企业官网建设流程全解析

• 示例工程

本指南以 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等教程中反复使用的高层函数。

在运行任何示例之前，请确保满足以下前置条件：

1. 本地 RabbitMQ 节点已启动，且使用默认配置（监听127.0.0.1、默认 vhost/、默认账号guest/guest）。所有源码中的连接参数均与此默认值一一对应，例如 send.hs 中的openConnection "127.0.0.1" "/" "guest" "guest"。
2. 安装 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.hs）：

main = 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的三个关键 API：

• openConnection host vhost user password：建立到 RabbitMQ 的连接；
• openChannel：在连接上打开一个信道（Channel），后续所有操作都在信道上进行；
• declareQueue ch newQueue {...}：声明队列，queueName设为"hello"，queueAutoDelete为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 getLine

consumeMsgs 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 deliveryHandler

• queueDurable = 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交换机。由于是 fanout，binding 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库的atMay：severityFor 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})

与教程四的区别在于默认 key：severityFor 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.hs（RPC 服务端）与 rpcClient.hs（RPC 客户端），运行命令：

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 m

• rpcQueue = "rpc_queue"是请求队列，qos ch 0 1 False保证同一时刻只处理一个请求；
• consumeMsgs ... Ack ...以手动确认模式消费请求；
• 用MVar让主线程阻塞在takeMVar m上，程序常驻等待请求。

请求处理函数handleRequest（rpcServer.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

关键点有三个：

1. replyTo：从请求消息头msgReplyTo msg取出回调队列名（即客户端的专属队列），应答消息就发到这个队列；
2. msgCorrelationID：把请求中的 correlation id 原样回填到应答消息，供客户端关联请求与响应；
3. 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 res

• genCorrelationId：用Data.UUID.V4的nextRandom生成 UUID 作为 correlation id（toText <$> 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的完整使用轮廓：

需要留意的是，Network.AMQP在 vhost、队列/交换机的持久化、确认模式等参数上的语义均与 AMQP 0-9-1 协议一致，示例默认连接127.0.0.1上使用默认配置的 RabbitMQ。若你的 Broker 不在本机或启用了 TLS、修改过默认账号，需要相应调整每个文件中的openConnection参数。上手时建议按教程一到六的顺序逐个运行：先起消费者再起生产者，观察终端输出即可直观验证每种消息模式的路由行为。若需进一步学习协议层面的设计与概念，可参考仓库根目录的 README.md 了解各语言版本的全貌，并结合 RabbitMQ 官方文档阅读（本仓库仅提供可运行代码，概念讲解请查阅官方教程站点）。

• 示例工程