使用 Scala 复刻 RabbitMQ 官方教程：从 Hello World 到 RPC 的完整实战指南

• 示例工程

rabbitmq-tutorials仓库的 scala/ 目录是 RabbitMQ 官方 Java 教程的一套极简 Scala 移植版，覆盖了消息队列入门最重要的六个主题：Hello World、Work Queues、Publish/Subscribe、Routing、Topics 与 RPC。读完本文，你将掌握如何用 Scala 2.12 + RabbitMQ Java 客户端（amqp-client）完成队列声明、消息发布与消费、持久化与手动确认、多种交换机路由模式，以及基于 correlationId 的远程过程调用，并能基于仓库源码逐行理解每个示例的实现细节。

一、项目背景：Java 教程的 Scala 移植版

在本仓库中，每个顶层目录都是一套独立的教程移植（tutorial port），scala/ 目录即其中的 Scala 版本。根据 scala/README.md 的说明，这是一份“minimalistic”（极简）的移植，代码风格与 Java 版本相当接近，原因在于 RabbitMQ Java 客户端至今仍兼容 JDK 6，并没有提供 lambda 友好的 API，因此移植时保留了大量命令式的调用方式（ConnectionFactory、Channel、DeliverCallback等）。

该目录共包含 14 个 Scala 源文件，分布在 scala/src/main/scala/ 下，对应关系如下：

除此之外，源码中还额外提供了基于 headers 交换机的示例 EmitLogHeader.scala 与 ReceiveLogHeader.scala，可作为深入学习的补充素材。

二、工程结构与构建配置

2.1 Maven 与 Scala 插件

整个 Scala 工程由 scala/pom.xml 定义，关键配置如下：

• Scala 版本：scala.version为2.12.7，依赖org.scala-lang:scala-library；
• RabbitMQ 客户端：com.rabbitmq:amqp-client，版本声明为(5.5.0,)。这是 Maven 的版本区间语法，表示“大于 5.5.0 的最新可用版本”，构建时会自动解析；
• 日志实现：org.slf4j:slf4j-simple（版本1.7.25），用于输出客户端内部日志；
• 构建插件：net.alchim31.maven:scala-maven-plugin（版本3.4.4）负责编译 Scala 源码；maven-compiler-plugin配置了<source>1.8</source>与<target>1.8</target>，即 Java 编译目标为 JDK 8。

工程的 Maven 元数据为com.rabbitmq:rabbitmq-tutorial-scala:0.1-SNAPSHOT，源码目录通过<sourceDirectory>src/main/scala</sourceDirectory>显式指定。

2.2 编译代码

项目内置了 Maven Wrapper（mvnw与mvnw.cmd），不需要事先安装 Maven 即可构建。wrapper 配置（scala/.mvn/wrapper/maven-wrapper.properties）指向 Maven 3.5.4 发行包，首次运行时由 wrapper 自动下载。

执行编译：

./mvnw compile

编译产物会输出到target/目录。注意所有命令都应在 scala/ 目录下执行。

三、运行前提与通用运行方式

3.1 前置条件

所有示例都假设本地localhost:5672上运行着一个 RabbitMQ 节点（默认端口、默认guest账号）。这也是整个仓库的统一约定，参见 AGENTS.md 中的说明：所有教程均需要 RabbitMQ 运行在localhost上并使用默认的guest凭据。

3.2 统一运行入口

每个示例都是一个包含main方法的 Scalaobject，通过 exec-maven-plugin 直接以主类名运行：

./mvnw exec:java -Dexec.mainClass="类名" [-Dexec.args="参数"]

• -Dexec.mainClass指定要执行的顶层对象名（即类名）；
• -Dexec.args可选，用于传递命令行参数，多个参数用空格分隔。

四、Tutorial 1：Hello World

这是最基础的“发送/接收”示例，消息直接发布到默认交换机、路由到名为hello的队列。

先在一个终端启动接收端：

./mvnw exec:java -Dexec.mainClass="Recv"

在另一个终端发送消息：

./mvnw exec:java -Dexec.mainClass="Send"

4.1 发送端源码解析

Send.scala 的完整流程如下：

val factory = new ConnectionFactory() factory.setHost("localhost") val connection = factory.newConnection() val channel = connection.createChannel() channel.queueDeclare(QUEUE_NAME, false, false, false, null) // QUEUE_NAME = "hello" val message = "Hello World!" channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8")) println(" [x] Sent '" + message + "'") channel.close() connection.close()

要点：

• queueDeclare("hello", false, false, false, null)声明一个非持久化（durable=false）、非排他（exclusive=false）、不自动删除（autoDelete=false）的队列；
• basicPublish("", QUEUE_NAME, ...)使用空字符串作为交换机名，即投递到 RabbitMQ 内置的默认交换机（direct 类型），以队列名作为路由键；
• 消息体统一以UTF-8编码的字节数组发送。

4.2 接收端源码解析

Recv.scala 使用回调式消费：

val deliverCallback: DeliverCallback = (_, delivery) => { val message = new String(delivery.getBody, "UTF-8") println(" [x] Received '" + message + "'") } channel.basicConsume(QUEUE_NAME, true, deliverCallback, _ => { })

• basicConsume的第二个参数true表示自动确认（autoAck），这里因为只是打印消息、不涉及失败重试，所以采用最简单的确认方式；
• Scala 版本利用 Java 客户端的函数式接口，将DeliverCallback写成 lambda 形式，这也是本移植版中比较“Scala 化”的地方；
• 启动后程序会一直阻塞等待消息，按Ctrl+C退出。

五、Tutorial 2：Work Queues（工作队列）

工作队列用于把耗时任务分发给多个 Worker 并行处理。任务文本中每个.代表一秒的模拟耗时。

发送一个立即完成的任务：

./mvnw exec:java -Dexec.mainClass="NewTask"

发送一个耗时约 4 秒的任务（rabbit1 ....中有 4 个点号）：

./mvnw exec:java -Dexec.mainClass="NewTask" -Dexec.args="rabbit1 ...."

启动一个 Worker（在独立终端运行）：

./mvnw exec:java -Dexec.mainClass="Worker"

按同样的方式再启动更多 Worker，消息会以**轮询（round robin）**的方式在多个 Worker 间分发。

5.1 生产者：持久化与消息属性

NewTask.scala 相比 Hello World 多了两个关键点：

channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null) // durable = true val message = if (argv.length < 1) "Hello World!" else argv.mkString(" ") channel.basicPublish("", TASK_QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes("UTF-8"))

• 持久化队列：queueDeclare的第二个参数为true，队列本身在 RabbitMQ 重启后依然存在；
• 持久化消息：发布时使用MessageProperties.PERSISTENT_TEXT_PLAIN，它携带deliveryMode = 2（持久化模式）与text/plain内容类型，保证 Broker 重启后消息不丢失；
• 消息内容拼接自命令行参数：无参数时默认"Hello World!"，有参数时用空格连接所有参数。

5.2 消费者：公平分发与手动确认

Worker.scala 是工作队列的核心：

channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null) channel.basicQos(1) // 一次只预取一条消息 ... val deliverCallback: DeliverCallback = (_, delivery) => { val message = new String(delivery.getBody, "UTF-8") println(" [x] Received '" + message + "'") try { doWork(message) } finally { println(" Done") channel.basicAck(delivery.getEnvelope.getDeliveryTag, false) } } channel.basicConsume(TASK_QUEUE_NAME, false, deliverCallback, _ => {})

三个关键设计：

1. basicQos(1)公平分发：默认情况下 RabbitMQ 会按轮询把消息依次发给每个 Worker，不考虑各 Worker 的处理速度；basicQos(1)限制每个 Worker 同时只预取 1 条消息，处理完并确认后才接收下一条，从而让快的 Worker 处理更多任务；
2. 手动确认：basicConsume的 autoAck 参数为false，处理完成后在finally中调用basicAck(deliveryTag, false)。若 Worker 在确认前崩溃，消息会被重新投递给其他 Worker，实现“处理失败不丢失”；
3. 模拟耗时：doWork遍历消息中的每个.字符，每遇到一个点Thread.sleep(1000)一秒，从而模拟不同长度的任务。

六、Tutorial 3：Publish/Subscribe（发布/订阅）

本教程引入**交换机（exchange）**概念，使用fanout类型交换机把每条消息广播给所有绑定到它的队列。

# 终端 1：接收所有日志 ./mvnw exec:java -Dexec.mainClass="ReceiveLogs" # 终端 2：发送日志 ./mvnw exec:java -Dexec.mainClass="EmitLog" -Dexec.args="rabbit1 msg1"

6.1 生产者：fanout 交换机

EmitLog.scala 的核心调用：

channel.exchangeDeclare(EXCHANGE_NAME, "fanout") // EXCHANGE_NAME = "logs" val message = if (argv.length < 1) "Hello World!" else argv.mkString(" ") channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8"))

• exchangeDeclare("logs", "fanout")声明fanout类型交换机，它忽略路由键，把消息复制到所有绑定的队列；
• 发布时路由键传空字符串即可，因为fanout根本不参与路由。

6.2 消费者：临时队列 + 绑定

ReceiveLogs.scala 展示了“临时匿名队列”的典型用法：

channel.exchangeDeclare(EXCHANGE_NAME, "fanout") val queueName = channel.queueDeclare().getQueue // 由服务端生成随机队列名 channel.queueBind(queueName, EXCHANGE_NAME, "")

• queueDeclare()不带参数时，RabbitMQ 会创建一个随机命名、排他、自动删除的临时队列，getQueue返回其名称；
• queueBind把该队列绑定到logs交换机，路由键为空；
• 每个ReceiveLogs进程都创建独立的临时队列，因此每个消费者都能收到广播消息的完整副本；进程退出（或连接断开）后临时队列自动消失。

七、Tutorial 4：Routing（路由）

direct类型交换机按路由键精确匹配路由消息，本示例用日志级别（info/warning/error）作为路由键。

接收info、warning、error三个级别：

./mvnw exec:java -Dexec.mainClass="ReceiveLogsDirect" -Dexec.args="info warning error"

发送一条error级别的日志：

./mvnw exec:java -Dexec.mainClass="EmitLogDirect" -Dexec.args="error Run. Run. Or it will explode."

7.1 生产者：参数解析与路由键

EmitLogDirect.scala 中，第一个命令行参数是严重级别（routing key），其余参数拼接为消息内容：

val severity = getSeverity(argv) // 默认 "info" val message = getMessage(argv) // 默认 "Hello World!" channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes("UTF-8"))

源码中getSeverity在无参数时默认返回"info"，getMessage在参数不足两个时返回"Hello World!"，并用joinStrings把从下标 1 开始的参数以空格连接。

7.2 消费者：多绑定键

ReceiveLogsDirect.scala 对每个命令行参数分别执行一次绑定：

if (argv.length < 1) { System.err.println("Usage: ReceiveLogsDirect [info] [warning] [error]") System.exit(1) } for (severity <- argv) { channel.queueBind(queueName, EXCHANGE_NAME, severity) }

• 同一队列可以绑定多个路由键，任一绑定匹配即可收到消息；
• 消费回调中通过delivery.getEnvelope.getRoutingKey打印实际的路由键，便于观察哪条消息命中了哪个绑定；
• 无参数时打印用法提示并退出，这是所有“接收端示例”的通用防御逻辑。

八、Tutorial 5：Topics（主题）

topic交换机在direct的基础上支持通配符模式的路由键匹配：*匹配一个单词，#匹配零个或多个单词，单词之间以.分隔。

接收所有日志（#匹配一切）：

./mvnw exec:java -Dexec.mainClass="ReceiveLogsTopic" -Dexec.args="#"

只接收kern设备的所有日志：

./mvnw exec:java -Dexec.mainClass="ReceiveLogsTopic" -Dexec.args="kern.*"

只接收所有critical级别的日志：

./mvnw exec:java -Dexec.mainClass="ReceiveLogsTopic" -Dexec.args="*.critical"

同时绑定多个模式：

./mvnw exec:java -Dexec.mainClass="ReceiveLogsTopic" -Dexec.args="kern.* *.critical"

发送一条路由键为kern.critical的日志：

./mvnw exec:java -Dexec.mainClass="EmitLogTopic" -Dexec.args="kern.critical A critical kernel error"

8.1 生产者与消费者的实现

EmitLogTopic.scala 与 ReceiveLogsTopic.scala 的结构与 Routing 示例几乎一致，仅两点不同：

• 交换机类型改为"topic"：channel.exchangeDeclare(EXCHANGE_NAME, "topic")；
• 生产者第一个参数是路由键（默认"anonymous.info"），其余参数为消息内容；
• 消费者把每个命令行参数作为绑定模式，逐个queueBind。

九、Tutorial 6：RPC（远程过程调用）

RPC 教程实现了一个经典模式：客户端发送请求到rpc_queue，同时声明一个临时replyTo回调队列，并通过correlationId关联请求与响应，服务端计算斐波那契数后把结果发回。

在一个终端启动 RPC 服务端：

./mvnw exec:java -Dexec.mainClass="RPCServer"

在另一个终端启动客户端：

./mvnw exec:java -Dexec.mainClass="RPCClient"

客户端默认请求fib(30)并打印计算结果。

9.1 服务端：回调类与斐波那契实现

RPCServer.scala 定义了一个独立的ServerCallback类实现DeliverCallback：

class ServerCallback(val ch: Channel, val latch: CountDownLatch) extends DeliverCallback { override def handle(consumerTag: String, delivery: Delivery): Unit = { var response: String = null val replyProps = new BasicProperties.Builder() .correlationId(delivery.getProperties.getCorrelationId) // 回传相同的 correlationId .build ... ch.basicPublish("", delivery.getProperties.getReplyTo, replyProps, response.getBytes("UTF-8")) ch.basicAck(delivery.getEnvelope.getDeliveryTag, false) latch.countDown() } }

要点：

• 从请求消息的属性中取出correlationId，构造响应属性时原样回填，供客户端匹配；
• 响应发布到delivery.getProperties.getReplyTo指定的回调队列，即默认交换机 + 回调队列名；
• 处理完成后手动basicAck；
• object Fibonacci中实现了迭代版fib(n)（非递归），避免栈溢出。

主流程中，服务端声明rpc_queue、设置basicQos(1)，并借助CountDownLatch(1)在消费完一条请求后即退出——源码注释明确说明“stop after one consumed message since this is example code”，即这是刻意为之的示例行为，真实服务应持续循环消费。

9.2 客户端：correlationId 与阻塞等待

RPCClient.scala 定义ResponseCallback与RPCClient两个类：

class ResponseCallback(val corrId: String) extends DeliverCallback { val response: BlockingQueue[String] = new ArrayBlockingQueueString override def handle(consumerTag: String, message: Delivery): Unit = { if (message.getProperties.getCorrelationId.equals(corrId)) { response.offer(new String(message.getBody, "UTF-8")) } } def take(): String = response.take() }

客户端call方法的调用链：

1. 生成UUID.randomUUID().toString作为corrId；
2. 构造包含correlationId与replyTo（临时回调队列名）的消息属性；
3. 把请求发布到rpc_queue；
4. 在回调队列上消费，ResponseCallback用ArrayBlockingQueue阻塞等待匹配corrId的响应；
5. 忽略不匹配corrId的响应，避免“迟到”的旧响应污染结果。

由于回调队列是channel.queueDeclare().getQueue生成的临时队列，客户端进程退出后自动清理。

十、源码中的额外示例：Headers 交换机

除了 README 中列出的六个教程，scala/src/main/scala/ 还包含基于headers 交换机的 EmitLogHeader.scala 与 ReceiveLogHeader.scala。headers 交换机不按路由键匹配，而是按消息的键值对属性匹配。

生产者用法（参数依次为消息内容、队列名、若干对 “键 值”）：

./mvnw exec:java -Dexec.mainClass="EmitLogHeader" -Dexec.args="message queueName format pdf type report"

生产者源码中的关键点：

channel.exchangeDeclare(EXCHANGE_NAME, "headers") // EXCHANGE_NAME = "header_test" val builder = new AMQP.BasicProperties.Builder() builder.deliveryMode(MessageProperties.PERSISTENT_TEXT_PLAIN.getDeliveryMode) builder.priority(MessageProperties.PERSISTENT_TEXT_PLAIN.getPriority) builder.headers(headers) // 从命令行参数两两构造的 HashMap channel.basicPublish(EXCHANGE_NAME, routingKey, theProps, message.getBytes("UTF-8"))

消费者则把“键值对”作为绑定参数传给queueBind：

channel.queueBind(queueName, EXCHANGE_NAME, routingKeyFromUser, headers)

这个示例可以看作 Topic 教程的延伸：当你需要按消息的任意属性（而非格式化路由键）路由时，headers 交换机是合适的选择。

十一、运行提示与常见问题

• 每个角色一个终端：生产者和消费者、RPC 客户端和服务端都应分别在独立 shell 中运行，README 中的命令都是“在一个终端 / 在另一个终端”的配合方式；
• 退出方式：所有消费者程序（Recv、Worker、ReceiveLogs系列）在打印[*] Waiting for messages. To exit press CTRL+C后持续阻塞，按Ctrl+C退出；
• 依赖下载：./mvnw首次运行会下载 Maven 3.5.4 与全部依赖（Scala 库、amqp-client、slf4j-simple），需要网络连接；amqp-client采用(5.5.0,)版本区间，实际解析到哪个版本取决于构建时 Maven Central 上可用的最新 5.5.0 以上版本；
• Broker 前置条件：所有示例默认连接localhost的 RabbitMQ（guest账号），请先确保 Broker 已启动；生产环境使用时应通过ConnectionFactory显式配置主机、端口与凭据（本移植版为保持教程简洁，统一硬编码localhost）；
• 示例代码定位：仓库 AGENTS.md 明确说明这些代码是教学材料而非生产代码，刻意保持简单、自包含；例如 RPC 服务端只处理一条请求即退出，实际产品中需要循环消费与更完善的错误处理。

十二、小结

从Send/Recv的默认交换机直连，到NewTask/Worker的持久化、公平分发与手动确认，再到fanout、direct、topic、headers四种交换机的逐级演进，最后到基于correlationId/replyTo的 RPC——这套 Scala 移植版以极小的代码量完整复刻了 RabbitMQ 的核心消息模式。对照 scala/README.md 中的命令逐条运行，再结合 scala/pom.xml 与 scala/src/main/scala/ 下的源码逐行阅读，即可系统掌握 AMQP 0-9-1 协议下的队列、交换机、绑定与消息确认机制，为后续在真实 Scala 项目中接入 RabbitMQ 打下扎实基础。

• 示例工程