示例工程【免费下载链接】rabbitmq-tutorialsTutorials for using RabbitMQ in various ways项目地址https://gitcode.com/gh_mirrors/ra/rabbitmq-tutorials点击查看免费下载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 portscala/ 目录即其中的 Scala 版本。根据 scala/README.md 的说明这是一份“minimalistic”极简的移植代码风格与 Java 版本相当接近原因在于 RabbitMQ Java 客户端至今仍兼容 JDK 6并没有提供 lambda 友好的 API因此移植时保留了大量命令式的调用方式ConnectionFactory、Channel、DeliverCallback等。该目录共包含 14 个 Scala 源文件分布在 scala/src/main/scala/ 下对应关系如下教程主题生产者 / 服务端消费者 / 客户端Hello WorldSend.scalaRecv.scalaWork QueuesNewTask.scalaWorker.scalaPublish/SubscribeEmitLog.scalaReceiveLogs.scalaRoutingEmitLogDirect.scalaReceiveLogsDirect.scalaTopicsEmitLogTopic.scalaReceiveLogsTopic.scalaRPCRPCServer.scalaRPCClient.scala除此之外源码中还额外提供了基于 headers 交换机的示例 EmitLogHeader.scala 与 ReceiveLogHeader.scala可作为深入学习的补充素材。二、工程结构与构建配置2.1 Maven 与 Scala 插件整个 Scala 工程由 scala/pom.xml 定义关键配置如下Scala 版本scala.version为2.12.7依赖org.scala-lang:scala-libraryRabbitMQ 客户端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配置了source1.8/source与target1.8/target即 Java 编译目标为 JDK 8。工程的 Maven 元数据为com.rabbitmq:rabbitmq-tutorial-scala:0.1-SNAPSHOT源码目录通过sourceDirectorysrc/main/scala/sourceDirectory显式指定。2.2 编译代码项目内置了 Maven Wrappermvnw与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 1Hello World这是最基础的“发送/接收”示例消息直接发布到默认交换机、路由到名为hello的队列。先在一个终端启动接收端./mvnw exec:java -Dexec.mainClassRecv在另一个终端发送消息./mvnw exec:java -Dexec.mainClassSend4.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)声明一个非持久化durablefalse、非排他exclusivefalse、不自动删除autoDeletefalse的队列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 化”的地方启动后程序会一直阻塞等待消息按CtrlC退出。五、Tutorial 2Work Queues工作队列工作队列用于把耗时任务分发给多个 Worker 并行处理。任务文本中每个.代表一秒的模拟耗时。发送一个立即完成的任务./mvnw exec:java -Dexec.mainClassNewTask发送一个耗时约 4 秒的任务rabbit1 ....中有 4 个点号./mvnw exec:java -Dexec.mainClassNewTask -Dexec.argsrabbit1 ....启动一个 Worker在独立终端运行./mvnw exec:java -Dexec.mainClassWorker按同样的方式再启动更多 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, _ {})三个关键设计basicQos(1)公平分发默认情况下 RabbitMQ 会按轮询把消息依次发给每个 Worker不考虑各 Worker 的处理速度basicQos(1)限制每个 Worker 同时只预取 1 条消息处理完并确认后才接收下一条从而让快的 Worker 处理更多任务手动确认basicConsume的 autoAck 参数为false处理完成后在finally中调用basicAck(deliveryTag, false)。若 Worker 在确认前崩溃消息会被重新投递给其他 Worker实现“处理失败不丢失”模拟耗时doWork遍历消息中的每个.字符每遇到一个点Thread.sleep(1000)一秒从而模拟不同长度的任务。六、Tutorial 3Publish/Subscribe发布/订阅本教程引入**交换机exchange**概念使用fanout类型交换机把每条消息广播给所有绑定到它的队列。# 终端 1接收所有日志 ./mvnw exec:java -Dexec.mainClassReceiveLogs # 终端 2发送日志 ./mvnw exec:java -Dexec.mainClassEmitLog -Dexec.argsrabbit1 msg16.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 4Routing路由direct类型交换机按路由键精确匹配路由消息本示例用日志级别info/warning/error作为路由键。接收info、warning、error三个级别./mvnw exec:java -Dexec.mainClassReceiveLogsDirect -Dexec.argsinfo warning error发送一条error级别的日志./mvnw exec:java -Dexec.mainClassEmitLogDirect -Dexec.argserror 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在无参数时默认返回infogetMessage在参数不足两个时返回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 5Topics主题topic交换机在direct的基础上支持通配符模式的路由键匹配*匹配一个单词#匹配零个或多个单词单词之间以.分隔。接收所有日志#匹配一切./mvnw exec:java -Dexec.mainClassReceiveLogsTopic -Dexec.args#只接收kern设备的所有日志./mvnw exec:java -Dexec.mainClassReceiveLogsTopic -Dexec.argskern.*只接收所有critical级别的日志./mvnw exec:java -Dexec.mainClassReceiveLogsTopic -Dexec.args*.critical同时绑定多个模式./mvnw exec:java -Dexec.mainClassReceiveLogsTopic -Dexec.argskern.* *.critical发送一条路由键为kern.critical的日志./mvnw exec:java -Dexec.mainClassEmitLogTopic -Dexec.argskern.critical A critical kernel error8.1 生产者与消费者的实现EmitLogTopic.scala 与 ReceiveLogsTopic.scala 的结构与 Routing 示例几乎一致仅两点不同交换机类型改为topicchannel.exchangeDeclare(EXCHANGE_NAME, topic)生产者第一个参数是路由键默认anonymous.info其余参数为消息内容消费者把每个命令行参数作为绑定模式逐个queueBind。九、Tutorial 6RPC远程过程调用RPC 教程实现了一个经典模式客户端发送请求到rpc_queue同时声明一个临时replyTo回调队列并通过correlationId关联请求与响应服务端计算斐波那契数后把结果发回。在一个终端启动 RPC 服务端./mvnw exec:java -Dexec.mainClassRPCServer在另一个终端启动客户端./mvnw exec:java -Dexec.mainClassRPCClient客户端默认请求fib(30)并打印计算结果。9.1 服务端回调类与斐波那契实现RPCServer.scala 定义了一个独立的ServerCallback类实现DeliverCallbackclass 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指定的回调队列即默认交换机 回调队列名处理完成后手动basicAckobject 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方法的调用链生成UUID.randomUUID().toString作为corrId构造包含correlationId与replyTo临时回调队列名的消息属性把请求发布到rpc_queue在回调队列上消费ResponseCallback用ArrayBlockingQueue阻塞等待匹配corrId的响应忽略不匹配corrId的响应避免“迟到”的旧响应污染结果。由于回调队列是channel.queueDeclare().getQueue生成的临时队列客户端进程退出后自动清理。十、源码中的额外示例Headers 交换机除了 README 中列出的六个教程scala/src/main/scala/ 还包含基于headers 交换机的 EmitLogHeader.scala 与 ReceiveLogHeader.scala。headers 交换机不按路由键匹配而是按消息的键值对属性匹配。生产者用法参数依次为消息内容、队列名、若干对 “键 值”./mvnw exec:java -Dexec.mainClassEmitLogHeader -Dexec.argsmessage 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))消费者则把“键值对”作为绑定参数传给queueBindchannel.queueBind(queueName, EXCHANGE_NAME, routingKeyFromUser, headers)这个示例可以看作 Topic 教程的延伸当你需要按消息的任意属性而非格式化路由键路由时headers 交换机是合适的选择。十一、运行提示与常见问题每个角色一个终端生产者和消费者、RPC 客户端和服务端都应分别在独立 shell 中运行README 中的命令都是“在一个终端 / 在另一个终端”的配合方式退出方式所有消费者程序Recv、Worker、ReceiveLogs系列在打印[*] Waiting for messages. To exit press CTRLC后持续阻塞按CtrlC退出依赖下载./mvnw首次运行会下载 Maven 3.5.4 与全部依赖Scala 库、amqp-client、slf4j-simple需要网络连接amqp-client采用(5.5.0,)版本区间实际解析到哪个版本取决于构建时 Maven Central 上可用的最新 5.5.0 以上版本Broker 前置条件所有示例默认连接localhost的 RabbitMQguest账号请先确保 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 打下扎实基础。赞分享示例工程【免费下载链接】rabbitmq-tutorialsTutorials for using RabbitMQ in various ways项目地址https://gitcode.com/gh_mirrors/ra/rabbitmq-tutorials点击查看免费下载相关推荐Haskell 实现 RabbitMQ 六套官方教程从 Hello World 到 RPC 的完整实战指南Haskell 实现 RabbitMQ 六套官方教程从 Hello World 到 RPC 的完整实战指南 本指南以 haskell/README.md ht示例工程RabbitMQ 官方教程 C / .NET 8 实战指南从 Hello World 到 Publisher Confirms 的完整落地RabbitMQ 官方教程 C / .NET 8 实战指南从 Hello World 到 Publisher Confirms 的完整落地 本指南基于当前仓库示例工程RabbitMQ 官方 Java 教程Maven 版实战指南从 Hello World 到 Publisher Confirms 全覆盖RabbitMQ 官方 Java 教程Maven 版实战指南从 Hello World 到 Publisher Confirms 全覆盖 导读 本文以 r示例工程上一篇BepInEx游戏插件框架5分钟完成Unity游戏模组安装的终极指南下一篇Windows右键菜单大瘦身3分钟学会用开源工具告别杂乱创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考