示例工程【免费下载链接】rabbitmq-tutorialsTutorials for using RabbitMQ in various ways项目地址https://gitcode.com/gh_mirrors/ra/rabbitmq-tutorials点击查看免费下载本指南以 haskell/README.md 为核心脉络系统讲解如何用 Haskell 与Network.AMQP客户端库逐一套跑通 RabbitMQ 官方六套教程Hello World、工作队列、发布/订阅、路由、主题、RPC。文中给出的每个命令与代码示例均来自仓库中真实可运行的.hs文件读完你既能用stack或runhaskell直接复现全部示例也能理解Network.AMQP在队列声明、消息发布、消费确认、交换机绑定与 RPC 回调等环节的核心 API 用法。环境准备Network.AMQP 与本地 RabbitMQ运行本仓库的 Haskell 示例只需要一个关键依赖Network.AMQP它是 Haskell 生态中面向 AMQP 0-9-1 协议的客户端库提供了openConnection、openChannel、declareQueue、declareExchange、publishMsg、consumeMsgs等教程中反复使用的高层函数。在运行任何示例之前请确保满足以下前置条件本地 RabbitMQ 节点已启动且使用默认配置监听127.0.0.1、默认 vhost/、默认账号guest/guest。所有源码中的连接参数均与此默认值一一对应例如 send.hs 中的openConnection 127.0.0.1 / guest guest。安装 GHC 与stack。如果不想手动管理 GHC 版本和依赖包README 推荐直接使用stack它会在首次运行时自动为你安装ghc、amqp以及其他必需包。两种运行方式runhaskell与stack仓库中每个.hs文件都可以独立运行。README 给出了两种执行方式直接执行runhaskell FILE ARGS使用stack执行stack FILE ARGS等价于stack runghc FILE ARGS。实际上每个源文件头部都内嵌了 stack 所需的依赖声明注释例如 send.hs#!/usr/bin/env stack {- stack --install-ghc runghc --package amqp --package bytestring -}这意味着即使你用runhaskell send.hs运行该文件本身也保留了stack模式下可复现的完整依赖清单而用stack send.hs hello world这类命令运行时stack会读取该头部注释自动安装 GHC 与依赖无需额外手工stack install。所有示例都启用了OverloadedStrings扩展{-# LANGUAGE OverloadedStrings #-}因此字符串字面量可以直接当作Text/ByteString使用简化了与 AMQP 消息体BL.ByteString的交互。教程一Hello World —— 队列声明、发送与消费对应文件send.hs生产者与 receive.hs消费者运行命令runhaskell send.hs runhaskell receive.hs生产者的核心逻辑非常精简send.hsmain do conn - openConnection 127.0.0.1 / guest guest ch - openChannel conn declareQueue ch newQueue {queueName hello, queueAutoDelete False, queueDurable False} publishMsg ch hello (newMsg {msgBody Hello World!, msgDeliveryMode Just NonPersistent}) BL.putStrLn [x] Sent Hello World! closeConnection conn这里体现了Network.AMQP的三个关键 APIopenConnection host vhost user password建立到 RabbitMQ 的连接openChannel在连接上打开一个信道Channel后续所有操作都在信道上进行declareQueue ch newQueue {...}声明队列queueName设为helloqueueAutoDelete为False队列在消费方断开后不自动删除queueDurable为False非持久化队列Broker 重启后消失。publishMsg ch hello (newMsg {...})中的第二个参数是交换机名称传入空字符串即使用默认交换机此时 routing key第三个参数hello就是目标队列名。消息通过newMsg构造msgBody指定消息体msgDeliveryMode Just NonPersistent明确该消息为非持久化投递。消费者receive.hs声明同一个队列后通过consumeMsgs注册回调consumeMsgs ch hello NoAck deliveryHandler -- waits for keypresses getLineconsumeMsgs ch queueName ackMode handler的第三个参数NoAck表示使用自动确认模式回调函数deliveryHandler :: (Message, Envelope) - IO ()收到(消息体, 投递元数据)二元组后用msgBody msg取出内容并打印。程序通过getLine阻塞等待按键防止 main 提前退出导致连接被关闭。教程二工作队列 —— 持久化、预取限制与手动确认对应文件newTask.hs任务生产者与 worker.hs任务消费者运行命令runhaskell newTask.hs hello world runhaskell worker.hs生产者newTask.hs与前例最大的区别是消息持久化publishMsg ch task_queue (newMsg {msgBody body, msgDeliveryMode Just Persistent})msgDeliveryMode Just Persistent会把消息标记为持久化投递配合 Worker 端对task_queue的持久化声明任务在 RabbitMQ 重启后仍能恢复。命令行参数通过getArgs读取并拼接为消息体bodyFor [] Hello, world!无参数时用默认值有参数时BL.pack . unwords $ xs将参数用空格连接。消费者worker.hs引入了三项针对任务处理的关键配置declareQueue ch newQueue {queueName task_queue, queueAutoDelete False, queueDurable True} qos ch 0 1 False consumeMsgs ch task_queue Ack deliveryHandlerqueueDurable True队列本身持久化配合消息持久化形成任务不丢失的完整链路qos ch 0 1 False设置 QoS 预取限制RabbitMQ 每次只向该消费者推送1条消息prefetch count 1避免某个慢消费者积压大量消息从而实现任务在多个 Worker 间的公平分发consumeMsgs ... Ack ...改用手动确认模式回调在处理完成后必须显式调用ackEnv metadata告诉 Broker本条消息已处理完成worker.hs。若 Worker 在处理过程中崩溃消息不会被确认会重新投递给其他消费者。回调中通过threadDelay (1000000 * n)模拟耗时任务其中n countDots body即消息体中.的个数BL.count .。例如发送runhaskell newTask.hs Hello........Worker 就会休眠约 8 秒——这正是教程用来演示负载均衡的方式。教程三发布/订阅 —— fanout 交换机与临时队列对应文件emitLog.hs日志生产者与 receiveLogs.hs日志消费者运行命令runhaskell receiveLogs.hs runhaskell emitLog.hs hello world这一套教程引入**交换机exchange**概念。生产者emitLog.hs声明了一个名为logs、类型为fanout的交换机logsExchange logs declareExchange ch newExchange {exchangeName logsExchange, exchangeType fanout, exchangeDurable False} publishMsg ch logsExchange (newMsg {msgBody body, msgDeliveryMode Just NonPersistent})fanout交换机会把收到的每条消息广播给所有绑定它的队列因此发布时 routing key 传空字符串即可被忽略。exchangeDurable False表示该交换机为非持久化。消费者receiveLogs.hs展示了临时队列 绑定的标准组合declareExchange ch newExchange {exchangeName logsExchange, exchangeType fanout, exchangeDurable False} (q, _, _) - declareQueue ch newQueue {queueName , queueAutoDelete True, queueDurable False} bindQueue ch q logsExchange queueName 由 Broker 生成随机唯一队列名如amq.gen-xxx每个消费者一个专属队列queueAutoDelete True消费者断开连接后队列自动删除bindQueue ch q logsExchange 将临时队列绑定到logs交换机。由于是 fanoutbinding key 同样为空字符串。declareQueue返回三元组(队列名, 消息数, 消费者数)这里用(q, _, _)只取队列名q供后续消费使用。这样每个消费者拿到的都是独立队列从而保证一条日志被广播给所有在线的日志接收者。教程四路由 —— direct 交换机与多 routing key 绑定对应文件emitLogDirect.hs 与 receiveLogsDirect.hs运行命令runhaskell receiveLogsDirect.hs info warn runhaskell emitLogDirect.hs warn a warning教程四把日志系统升级为按严重级别路由。生产者emitLogDirect.hs改用direct类型交换机发布时以第一个命令行参数作为 routing key即严重级别logsExchange direct_logs declareExchange ch newExchange {exchangeName logsExchange, exchangeType direct, exchangeDurable False} publishMsg ch logsExchange severity (newMsg {msgBody body, msgDeliveryMode Just NonPersistent})参数解析使用了Safe库的atMayseverityFor xs maybe info DT.pack (atMay xs 0)即默认严重级别为info消息体bodyFor xs maybe Hello world BL.pack (atMay xs 1)取第二个参数缺省为Hello world。消费者receiveLogsDirect.hs允许传入多个严重级别并为每个级别都建立绑定(q, _, _) - declareQueue ch newQueue {queueName , queueAutoDelete True, queueDurable False} forM_ severities (bindQueue ch q logsExchange . DT.pack)关键在forM_ severities (bindQueue ch q logsExchange . DT.pack)对命令行传入的每个严重级别如info、warn都调用一次bindQueue使同一个队列绑定多个 routing key。direct交换机的路由规则是routing key 精确匹配的队列才会收到消息。例如运行runhaskell receiveLogsDirect.hs info warn后该队列只会收到级别为info或warn的日志而error级别则被过滤。回调中通过envRoutingKey metadata取回该消息实际的路由 key打印为[x] warn:a warning的格式receiveLogsDirect.hs并调用ackEnv metadata完成手动确认。教程五主题 —— topic 交换机与通配符匹配对应文件emitLogTopic.hs 与 receiveLogsTopic.hs运行命令runhaskell receiveLogsTopic.hs info warn runhaskell emitLogTopic.hs warn a warning教程五将路由升级为主题匹配。生产者emitLogTopic.hs把交换机类型改为topic并将 routing key 设计为设施.严重级别的层级结构logsExchange topic_logs declareExchange ch newExchange {exchangeName logsExchange, exchangeType topic, exchangeDurable False} publishMsg ch logsExchange severity (newMsg {msgBody body, msgDeliveryMode Just NonPersistent})与教程四的区别在于默认 keyseverityFor xs maybe anonymous.info DT.pack (atMay xs 0)缺省值为anonymous.info默认设施anonymous 默认级别info。消费者端receiveLogsTopic.hs的队列声明与绑定逻辑和 direct 版完全一致(q, _, _) - declareQueue ch newQueue {queueName , queueAutoDelete True, queueDurable False} forM_ severities (bindQueue ch q logsExchange . DT.pack)区别全部体现在绑定 key 的语义上。topic交换机按点分单词的 routing key 做部分匹配binding key 中允许两种通配符*恰好匹配一个单词#匹配零个或多个单词。例如绑定*.warn可收到a.warn、b.warn等所有设施产生的 warn 日志绑定kern.*只收内核日志绑定#则接收全部日志。这也是为什么教程建议把 receive 端的info warn换成*.info #之类模式来体验更灵活的路由。与 direct 一样topic模式也支持对同一队列绑定多个 key 实现并集接收。教程六RPC —— 回调队列、correlation id 与同步调用对应文件rpcServer.hsRPC 服务端与 rpcClient.hsRPC 客户端运行命令runhaskell rpcServer.hs runhaskell rpcClient.hs教程六实现一个经典的 Fibonacci RPC客户端把fib(30)请求发到rpc_queue服务端计算后把结果发回客户端专属的回调队列。这是六套教程中代码量最大、最能体现Network.AMQP异步模型的一组。服务端[rpcServer.hs](https://link.gitcode.com/i/846ceb55720d83d9553df239409b79b1#L11-L46的核心qos ch 0 1 False declareQueue ch newQueue {queueName rpcQueue} m - newEmptyMVar consumeMsgs ch rpcQueue Ack $ handleRequest ch m putStrLn [x] Awaiting RPC requests takeMVar mrpcQueue rpc_queue是请求队列qos ch 0 1 False保证同一时刻只处理一个请求consumeMsgs ... Ack ...以手动确认模式消费请求用MVar让主线程阻塞在takeMVar m上程序常驻等待请求。请求处理函数handleRequestrpcServer.hs完整展示了 RPC 应答链路handleRequest ch m (msg, envelope) do n - readIO . BL.unpack . msgBody $ msg putStrLn $ [.] fib( show n ) let result fib n let response newMsg { msgCorrelationID msgCorrelationID msg , msgBody BL.pack . show $ result } publishMsg ch replyTo response ackEnv envelope where replyTo fromJust $ msgReplyTo msg关键点有三个replyTo从请求消息头msgReplyTo msg取出回调队列名即客户端的专属队列应答消息就发到这个队列msgCorrelationID把请求中的 correlation id 原样回填到应答消息供客户端关联请求与响应fib采用迭代实现fib n go n 0 1尾递归避免指数爆炸计算完成后publishMsg ch replyTo response经默认交换机投递到回调队列最后ackEnv envelope确认请求已处理。客户端[rpcClient.hs](https://link.gitcode.com/i/5646957a7cb39ff68e31001a006d7830#L30-L57封装了callFib ch queue n这一同步风格调用callFib ch queue n do cid - genCorrelationId rqn - declareReplyQueue let body BL.pack . show $ n let message newMsg {msgCorrelationID Just cid, msgReplyTo Just rqn, msgBody body} publishMsg ch queue message m - newEmptyMVar consumeMsgs ch rqn Ack $ handleResponse cid m res - takeMVar m return resgenCorrelationId用Data.UUID.V4的nextRandom生成 UUID 作为 correlation idtoText $ nextRandom保证全局唯一declareReplyQueue声明一个临时回调队列queueAutoDelete True, queueExclusive True服务端应答就发到这里请求发出后在回调队列上注册handleResponse cid m当收到的消息msgCorrelationId cid时解析结果并putMVar m res外层takeMVar m拿到结果——用MVar把异步消费包装成同步等待消息头msgCorrelationID、msgReplyTo都通过Just包装的Maybe传递读取时用fromJust取出这与服务端fromJust $ msgReplyTo msg形成对称。小结一份可运行、可复现的 Haskell AMQP 学习地图从 haskell/README.md 到仓库内的 12 个源文件整套教程覆盖了 RabbitMQ 最核心的六种消息模式同时也勾勒出Network.AMQP的完整使用轮廓教程交换机类型关键 API / 概念源文件Hello World默认交换机declareQueue、publishMsg、consumeMsgs NoAcksend.hs、receive.hs工作队列默认交换机queueDurable、msgDeliveryMode Persistent、qos、手动AcknewTask.hs、worker.hs发布/订阅fanoutdeclareExchange、临时队列、bindQueueemitLog.hs、receiveLogs.hs路由directrouting key 精确匹配、多 key 绑定forM_bindQueueemitLogDirect.hs、receiveLogsDirect.hs主题topic*/#通配符、层级 keyemitLogTopic.hs、receiveLogsTopic.hsRPC默认交换机replyTo、correlationId、MVar同步封装rpcServer.hs、rpcClient.hs需要留意的是Network.AMQP在 vhost、队列/交换机的持久化、确认模式等参数上的语义均与 AMQP 0-9-1 协议一致示例默认连接127.0.0.1上使用默认配置的 RabbitMQ。若你的 Broker 不在本机或启用了 TLS、修改过默认账号需要相应调整每个文件中的openConnection参数。上手时建议按教程一到六的顺序逐个运行先起消费者再起生产者观察终端输出即可直观验证每种消息模式的路由行为。若需进一步学习协议层面的设计与概念可参考仓库根目录的 README.md 了解各语言版本的全貌并结合 RabbitMQ 官方文档阅读本仓库仅提供可运行代码概念讲解请查阅官方教程站点。赞分享示例工程【免费下载链接】rabbitmq-tutorialsTutorials for using RabbitMQ in various ways项目地址https://gitcode.com/gh_mirrors/ra/rabbitmq-tutorials点击查看免费下载相关推荐PDF.js 官方 Examples 指南从 Hello World 到完整页面渲染的实战教程PDF.js 官方 Examples 指南从 Hello World 到完整页面渲染的实战教程 本篇技术指南以 PDF.js 官方文档 Examples 篇前端ReScript 官方项目模板入门从 npm install 到 Hello World 的完整实战指南ReScript 官方项目模板入门从 npm install 到 Hello World 的完整实战指南 本指南以 ReScript 官方 starter 模编译器编程语言开发工具GCDWebServer终极实战指南从Hello World到企业级应用完整教程GCDWebServer终极实战指南从Hello World到企业级应用完整教程 GCDWebServer 是iOS、macOS和tvOS平台上排名第一的轻量后端创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考