企业官网建设流程全解析

• 示例工程

本指南以本仓库 java 目录下的官方 RabbitMQ 教程 Java 示例代码为主体，完整覆盖依赖获取、classpath 配置、编译以及七个经典教程场景（Hello World、Work Queues、Publish/Subscribe、Routing、Topics、RPC、Publisher Confirms）的终端运行方法，并逐例结合源码剖析其底层实现要点。读完本文，你将能够在本地 RabbitMQ 节点上独立编译并运行每一组示例，理解 Java 客户端（amqp-client）在队列声明、交换机绑定、消息确认、RPC 回调等环节的真实调用方式。

一、前置条件：本地 RabbitMQ 节点

所有示例均面向运行在 localhost 上的 RabbitMQ 节点（默认端口 5672，默认账号guest/guest），并要求使用默认配置。文档明确指出：To successfully use the examples you will need a RabbitMQ node running locally。因此动手前请先确认本机 RabbitMQ 服务已启动，可通过rabbitmqctl status或 RabbitMQ 管理界面（默认 http://localhost:15672）验证节点可用。

需要特别说明的是，本目录下的代码不依赖任何构建工具（无 Maven/Gradle），采用手动管理依赖 JAR 的方式，这也是它与仓库中 java-mvn（Maven 版）、java-gradle（Gradle 版）目录的最大区别。从根目录 README.md 可以看到，仓库同时维护了Java (with manual dependency management)、Java with Gradle、Java with Maven三种形态，本指南聚焦手动依赖管理这一种。

二、依赖获取：三个必需的 JAR 文件

示例运行需要从 Maven Central 下载三个 JAR：

使用wget下载（Linux/macOS 环境）：

wget https://repo1.maven.org/maven2/com/rabbitmq/amqp-client/5.21.0/amqp-client-5.21.0.jar wget https://repo1.maven.org/maven2/org/slf4j/slf4j-api/2.0.13/slf4j-api-2.0.13.jar wget https://repo1.maven.org/maven2/org/slf4j/slf4j-simple/2.0.13/slf4j-simple-2.0.13.jar

将这三个 JAR 与教程的 Java 源文件放在同一个工作目录下。文档强调了一个关键差异：编译阶段只需要 amqp-client 一个 JAR 在 classpath 上，而运行阶段则需要全部三个依赖。这是因为编译时javac只需解析客户端 API 的符号，而运行时 SLF4J 绑定必须存在，否则客户端库的日志调用会因找不到实现而报错。

三、classpath 配置：环境变量一劳永逸

为免去每条命令重复书写长串 classpath，文档推荐设置环境变量。Linux/macOS 用冒号:分隔：

export CP=.:amqp-client-5.21.0.jar:slf4j-api-2.0.13.jar:slf4j-simple-2.0.13.jar java -cp $CP Send

Windows 用分号;分隔：

set CP=.;amqp-client-5.21.0.jar;slf4j-api-2.0.13.jar;slf4j-simple-2.0.13.jar java -cp %CP% Send

注意CP中的第一个元素.（当前目录），它保证编译产物（.class文件）能被java命令找到。后续所有教程的运行命令均复用这个CP变量。

四、Tutorial One：Hello World（Send / Recv）

这是入门第一课：一个生产者向队列发消息，一个消费者从队列取消息。先编译两个类：

javac -cp amqp-client-5.21.0.jar Send.java Recv.java

然后开两个终端，先启动消费者再启动生产者：

# terminal tab 1（消费者） java -cp .:amqp-client-5.21.0.jar:slf4j-api-2.0.13.jar:slf4j-simple-2.0.13.jar Recv # terminal tab 2（生产者） java -cp .:amqp-client-5.21.0.jar:slf4j-api-2.0.13.jar:slf4j-simple-2.0.13.jar Send

消费者终端会先打印[*] Waiting for messages. To exit press CTRL+C，生产者发送后消费者打印[x] Received 'Hello World!'。

从源码看实现要点（Send.java、Recv.java）：

• 两端都通过ConnectionFactory设置factory.setHost("localhost")建立连接，再createChannel()获取 Channel；
• 队列声明使用了channel.queueDeclare(QUEUE_NAME, true, false, false, Map.of("x-queue-type", "quorum"))：true表示持久化队列，并显式指定x-queue-type为quorum 队列（基于 Raft 的复制队列，RabbitMQ 3.8+ 支持），这是本仓库示例相比早期教程版本的新特性；
• 生产者通过channel.basicPublish("", QUEUE_NAME, null, ...)向默认交换机（空字符串名称）按队列名路由消息；
• 消费者用DeliverCallback回调接口异步接收消息，basicConsume的true表示自动确认（autoAck）。

五、Tutorial Two：Work Queues（NewTask / Worker）

工作队列用于把耗时任务分发给多个 Worker，实现负载均衡。编译与运行：

javac -cp $CP NewTask.java Worker.java # terminal tab 1（第一个 Worker） java -cp $CP Worker # terminal tab 2（第二个 Worker，可再开多个模拟多消费者） java -cp $CP NewTask

NewTask 的可执行参数即消息内容，例如java -cp $CP NewTask First message.，其中点号.越多，任务模拟耗时越长。

源码要点（NewTask.java、Worker.java）：

• 消息持久化：NewTask 发布时使用MessageProperties.PERSISTENT_TEXT_PLAIN（deliveryMode=2），确保消息在 broker 重启后不丢失；队列同样声明为持久化 quorum 队列；
• 公平分发：Worker 调用channel.basicQos(1)，告诉 broker 在收到确认前只向该消费者推送 1 条消息，避免某个慢消费者积压任务（Round-Robin 退化为能者多劳）；
• 手动确认：basicConsume第二参数为false（关闭自动确认），任务完成后调用channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false)显式确认；若 Worker 中途崩溃，未确认消息会重新投递给其他消费者；
• 任务模拟：doWork遍历消息字符，每遇到一个.就Thread.sleep(1000)，从而可以用点号数量控制任务耗时（Worker.java）。

六、Tutorial Three：Publish/Subscribe（EmitLog / ReceiveLogs）

发布/订阅模式：一份日志广播给所有消费者。核心是fanout 类型交换机，它不关心 routing key，把消息复制给所有绑定队列。

javac -cp $CP EmitLog.java ReceiveLogs.java # terminal tab 1（第一个订阅者） java -cp $CP ReceiveLogs # terminal tab 2（第二个订阅者，可多开几个验证广播效果） java -cp $CP EmitLog

源码要点（EmitLog.java、ReceiveLogs.java）：

• 生产者和消费者两端都执行channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT)声明logs交换机（幂等操作，重复声明无害）；
• 消费者通过无参channel.queueDeclare().getQueue()获得一个服务端生成的随机名、独占、自动删除的临时队列，再用channel.queueBind(queueName, EXCHANGE_NAME, "")以空 routing key 绑定；
• 生产者basicPublish(EXCHANGE_NAME, "", ...)只指定交换机，不关心路由键；
• 消费者basicConsume(queueName, true, ...)使用自动确认，适合日志这类可丢弃消息；
• 生产者默认消息为info: Hello World!，也可用命令行参数自定义，如java -cp $CP EmitLog "A log message"。

七、Tutorial Four：Routing（EmitLogDirect / ReceiveLogsDirect）

在发布/订阅基础上引入direct 交换机：消息按 routing key 精确匹配绑定键，实现按日志级别选择性接收。

javac -cp $CP ReceiveLogsDirect.java EmitLogDirect.java # terminal tab 1：只接收 warning 和 error java -cp $CP ReceiveLogsDirect warning error # terminal tab 2：接收 info、warning、error java -cp $CP ReceiveLogsDirect info warning error # terminal tab 3：发送一条 error 级别日志 java -cp $CP EmitLogDirect error "Run. Run. Or it will explode."

源码要点（EmitLogDirect.java、ReceiveLogsDirect.java）：

• 生产者把命令行第一个参数作为 severity（routing key），默认info；消息为其余参数拼接，默认Hello World!；
• 消费者为每个命令行参数执行一次channel.queueBind(queueName, EXCHANGE_NAME, severity)，从而对同一队列建立多个绑定键；
• 接收端打印时通过delivery.getEnvelope().getRoutingKey()展示实际路由键，方便对照验证匹配结果；
• 若消费者不带任何参数启动，会打印Usage: ReceiveLogsDirect [info] [warning] [error]并退出（ReceiveLogsDirect.java）。

八、Tutorial Five：Topics（EmitLogTopic / ReceiveLogsTopic）

Topics 使用topic 交换机，binding key 支持两个通配符：*（匹配一个单词）和#（匹配零个或多个单词），实现按主题模式订阅。routing key 由点号分隔的单词组成，如kern.critical。

# To compile: javac -cp $CP ReceiveLogsTopic.java EmitLogTopic.java # To receive all the logs（匹配所有消息）: java -cp $CP ReceiveLogsTopic "#" # To receive all logs from the facility "kern": java -cp $CP ReceiveLogsTopic "kern.*" # Or if you want to hear only about "critical" logs: java -cp $CP ReceiveLogsTopic "*.critical" # You can create multiple bindings（同时建立多个绑定）: java -cp $CP ReceiveLogsTopic "kern.*" "*.critical" # And to emit a log with a routing key "kern.critical" type: java -cp $CP EmitLogTopic "kern.critical" "A critical kernel error"

源码要点（EmitLogTopic.java、ReceiveLogsTopic.java）：

• 生产者默认 routing key 为anonymous.info，默认消息Hello World!；
• 消费者遍历argv为每个 binding key 调用一次queueBind，支持任意数量绑定；
• 无参数启动时打印Usage: ReceiveLogsTopic [binding_key]...并退出；
• 该示例同样使用服务端生成的临时队列（无参queueDeclare()），消费者退出后队列自动清理。

九、Tutorial Six：RPC（RPCClient / RPCServer）

RPC 教程演示如何用 RabbitMQ 实现远程过程调用：客户端发送请求、服务器计算斐波那契数并返回结果。

# Compile and set up the classpath as usual (see tutorial one): javac -cp $CP RPCClient.java RPCServer.java # Our RPC service is now ready. We can start the server: java -cp $CP RPCServer # To request a fibonacci number run the client: java -cp $CP RPCClient

运行后，客户端会循环请求fib(0)到fib(31)，打印[x] Requesting fib(N)与[.] Got '结果'；服务器打印[.] fib(N)。

源码要点（RPCClient.java、RPCServer.java）：

• correlationId 关联请求与响应：客户端每次调用生成UUID.randomUUID().toString()作为 correlationId，服务器在回包中原样带回；
• replyTo 回调队列：客户端通过channel.queueDeclare().getQueue()创建一个临时私有队列，把队列名写入消息属性replyTo；服务器将结果发布到该队列；
• 同步等待：客户端用CompletableFuture<String>挂起等待匹配 correlationId 的响应，收到后basicCancel取消订阅（RPCClient.java）；
• 服务端健壮性：RPCServer声明rpc_queue为持久化 quorum 队列并先queuePurge清空历史消息；basicQos(1)限制并发；fib采用迭代算法避免递归栈溢出；响应发布与basicAck放在finally块中，确保异常时也能回包（RPCServer.java）；
• 消息属性构造：两端都使用AMQP.BasicProperties.Builder链式构建属性。

十、Tutorial Seven：Publisher Confirms（PublisherConfirms）

第七个教程演示发布者确认（publisher confirms），用于可靠发布。它与其他示例不同——不涉及消费者，只测试发布端确认的吞吐表现：

javac -cp $CP PublisherConfirms.java java -cp $CP PublisherConfirms

程序固定发布50,000条消息（MESSAGE_COUNT = 50_000），依次执行五种确认策略并各自打印耗时（PublisherConfirms.java）：

1. publishMessagesIndividually：每条消息发布后立即waitForConfirmsOrDie(5_000)等待确认，最慢但实现最简单；
2. publishMessagesInBatch：每累计 100 条未确认消息批量等待一次确认（batchSize = 100），显著降低往返开销；
3. handlePublishConfirmsAsynchronously：使用ConcurrentSkipListMap<Long, String>以发布序号跟踪未确认消息，通过addConfirmListener异步处理 ack/nack；nack 时打印消息体与序号便于重发（PublisherConfirms.java）；
4. handlePublishConfirmsWithWindow：引入确认窗口MAX_OUTSTANDING = 1000，窗口满时挂起等待确认腾出空间，兼顾吞吐与内存上限（PublisherConfirms.java）；
5. handlePublishConfirmsWithAdaptiveThrottling：在窗口机制上叠加自适应节流——当窗口使用率达到 50%（THROTTLING_PERCENTAGE）后，按占用比例线性增加最大 1000ms（MAX_DELAY_MS）的等待延迟，平滑应对 broker 处理压力（PublisherConfirms.java）。

所有策略的关键前置动作都是ch.confirmSelect()开启发布者确认模式，以及ch.getNextPublishSeqNo()获取每条消息的递增序号用于映射确认结果。运行时需注意：程序会为每个策略创建一次性临时队列（queueDeclare(queue, false, false, true, null)），测试结束即删除。

十一、仓库扩展：Header 交换机示例与一键编译脚本

除了文档提及的七个教程，java 目录还包含两个基于headers 交换机的进阶示例，可作为路由能力延伸的补充学习材料：

• EmitLogHeader.java：命令行以消息 键1 值1 键2 值2 ...形式传入，通过AMQP.BasicProperties.Builder.headers(map)将头部附加到消息属性，并继承MessageProperties.PERSISTENT_TEXT_PLAIN的 deliveryMode 与 priority；注释明确指出 headers 交换机的路由不依赖 routing key，routing key 仅作信息载体（EmitLogHeader.java）；
• ReceiveLogHeader.java：绑定队列时用channel.queueBind(queueName, EXCHANGE_NAME, routingKey, headers)携带匹配头，且必须在命令行中显式给出x-match头（any或all）来决定多个头的匹配语义，例如java -cp $CP ReceiveLogHeader testQueue1 x-match any header1 value1（ReceiveLogHeader.java）。

此外，仓库提供了 recompile.sh 一键编译脚本，等价于执行：

javac -cp .:amqp-client-5.21.0.jar:slf4j-api-2.0.13.jar:slf4j-simple-2.0.13.jar *.java

它会把目录下全部.java源文件一次编译为 class 文件，适合反复修改后批量重编译的场景。

十二、构建工具与 AMQP 1.0 变体：按需选择

若你更习惯构建工具管理依赖，仓库提供了两套等价实现：

• java-mvn：Maven 工程，pom.xml声明 amqp-client 依赖，可用./mvnw运行；
• java-gradle：Gradle 工程，settings.gradle配置项目，源文件与 java 目录一致；
• java-amqp：使用AMQP 1.0 Java 客户端的变体，要求 RabbitMQ 4.0+，运行方式为./mvnw compile exec:java -Dexec.mainClass=Send（详见 java-amqp/README.md），其源码位于src/main/java下并由 Amqp10TutorialIntegrationTest.java 提供集成测试验证。

十三、常见问题与排查建议

• ConnectException: Connection refused：说明本地 5672 端口无 RabbitMQ 节点，请先启动 broker 再运行；
• SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder"：运行 classpath 缺少 slf4j-simple（或其它绑定实现），请确认三个 JAR 都在CP中；
• NoClassDefFoundError: com/rabbitmq/client/...：编译或运行时 classpath 缺少 amqp-client JAR，检查 jar 文件名与CP是否一致；
• quorum 队列不可用：x-queue-type=quorum需要 RabbitMQ 3.8+，老版本节点会拒绝队列声明，可改用普通持久化队列（移除该参数）；
• 多终端顺序：订阅/消费类程序（Recv、Worker、ReceiveLogs 系）应先行启动等待，发布类程序（Send、NewTask、EmitLog 系）随后执行，便于观察消息流转。

以上所有命令均来自 java/README.md 原文，配合仓库源码 java 目录下的 15 个 Java 文件即可在本地完整复现 RabbitMQ 的七大核心消息模式，并为后续深入学习交换机类型、确认机制与 RPC 模式提供可直接运行的实验基础。

• 示例工程