专栏RabbitMQ 进阶之路个人主页手握风云目录一、七大工作模式1.1. Simple简单模式1.2. Work Queue 工作队列模式1.3. Publish/Subscribe 发布订阅模式1.4. Routing 路由模式1.5. Topics 通配符模式1.6. RPC 远程调用模式1.7. Publisher Confirms 发布确认二、Java API 代码实现2.1. Work Queue 工作队列2.2. Fanout 发布订阅2.3. Direct 路由2.4. Topic 通配符2.5. RPC 模式一、七大工作模式1.1. Simple简单模式Simple 简单模式是 RabbitMQ 最基础的点对点模型包含生产者 P、消息队列 Queue、消费者 C 三类核心角色生产者负责投递消息至队列消费者从队列取出消息进行处理。该模式遵循一对一规则仅存在一个生产者、一个队列和一个消费者每条消息只会被消费一次整个流程不使用自定义交换机消息通过 RabbitMQ 默认内置交换机直接发送到队列适用于仅需要单个程序处理消息的简单异步通知场景。1.2. Work Queue 工作队列模式Work Queue 工作队列模式属于简单模式的增强版本同样不使用自定义交换机由一个生产者向单一队列发送消息但支持多个消费者同时监听同一个队列多个消费者形成竞争消费关系。队列会以轮询的方式分发消息每条消息只会分配给其中一个消费者不会出现重复消费的情况默认会将奇数序号、偶数序号消息分开交给不同消费者搭配手动 ACK 与 QoS 机制还能实现公平分发。该模式主要用于集群环境下的异步任务分流、流量削峰典型场景是 12306 订单系统推送短信通知任务多个短信服务实例分摊队列消息提升整体处理效率。1.3. Publish/Subscribe 发布订阅模式Publish/Subscribe 发布订阅模式首次引入交换机 Exchange 角色使用 fanout 类型交换机生产者不会直接将消息发送至队列而是先投递到交换机交换机本身不具备消息存储能力仅负责转发路由若没有队列与交换机绑定消息会直接丢失。fanout 交换机采用广播策略会把收到的消息复制多份转发给所有绑定自身的队列绑定队列时 RoutingKey 和 BindingKey 统一填写空字符串不参与匹配逻辑。该模式架构为一个生产者、一个 fanout 交换机、多个独立队列每个队列对应单独消费者生产者发送一条消息后所有绑定队列的消费者都会收到完整相同的消息适合一条消息需要多系统同步接收的广播通知场景例如气象局推送气象数据新浪、百度、网易等多个门户网站同时订阅接收。1.4. Routing 路由模式Routing 路由模式是发布订阅模式的衍生升级方案该模式使用 Direct 类型交换机摒弃了 Fanout 交换机无差别广播的逻辑依靠路由键实现消息的精准定向分发。该模式包含生产者、Direct 交换机、多个独立队列与对应消费者四类组件其中有两个核心路由相关概念生产者发送消息时携带的字符串为 RoutingKey队列和交换机绑定时指定的匹配字符串为 BindingKey只有二者内容完全一致消息才会被路由至对应队列交换机本身不具备存储消息的能力若不存在匹配的绑定队列消息会直接丢失。同时一个队列可以绑定多个不同的 BindingKey能够接收多种路由键对应的消息实现多类消息统一汇总消费。路由模式的核心特点是消息按需过滤、精准分发单条消息只会流入匹配成功的队列不会出现全量广播的情况非常适合需要对消息做分类隔离处理的业务场景典型应用就是系统分级日志处理系统日志分为 error、warning、info、debug 四个等级每条日志携带对应等级的 RoutingKey错误日志队列仅绑定 error普通日志队列绑定 info 和 warning以此实现不同等级日志分开存储、输出至不同文件。1.5. Topics 通配符模式Topics 通配符模式是 Routing 路由模式的进一步扩展核心区别在于使用 Topic 类型交换机不再采用严格相等匹配而是支持通配符完成模糊匹配消息过滤的灵活度大幅提升。该模式对路由键的格式有强制规范无论是生产者的 RoutingKey 还是队列绑定的 BindingKey都必须是由英文点.分割的多段单词例如 order.error、order.pay.info同时定义了两种通配符*可以匹配任意单个单词#能够匹配 0 个或多个连续单词依靠两种符号可以组合出多样的订阅规则。通配符模式底层的消息分发逻辑和 Direct 交换机保持一致仅匹配规则做了扩展既可以实现精准匹配也能完成大范围模糊订阅业务扩展性极强主要适用于多模块、多维度的消息订阅场景比如订单、支付、用户模块分别产生 error、info 类型日志业务系统可以按需订阅全部错误日志或是仅订阅支付模块的信息日志。1.6. RPC 远程调用模式RPC 远程调用模式和前面单向投递消息的几种模式完全不同它基于 RabbitMQ 实现同步的请求 - 响应远程调用效果模拟接口调用的一问一答逻辑依靠两套队列完成双向通信全程不使用自定义交换机消息通过 RabbitMQ 默认内置交换机传输。整套调用流程分为客户端与服务端两大角色客户端作为调用发起方首先声明全局公共请求队列同时创建专属临时回调队列用于接收返回结果发送请求消息时会在消息属性中配置两个关键标识replyTo 字段填写回调队列名称用于告知服务端响应消息发送位置correlationId 是本次请求的唯一标识用来区分并发场景下多条请求对应的返回数据客户端发送消息后会阻塞持续监听回调队列等待处理结果。服务端作为业务处理方持续监听公共请求队列收到消息后执行业务逻辑读取消息内的 replyTo 和 correlationId将处理完成的响应消息发送至指定回调队列并携带相同的 correlationId处理完成后手动调用 ACK 确认消息避免消息丢失。客户端监听到回调消息后对比消息内的 correlationId匹配成功则取出结果结束本次同步调用。RPC 模式有几项关键机制保障调用稳定服务端会设置 basicQos (1)限制每次仅获取一条消息处理完毕后再接收下一条保证请求顺序处理消费时将 autoAck 设置为 false开启手动消息确认业务处理异常时可重新投递消息客户端内部使用阻塞队列实现同步等待达成同步调用的使用体验。该模式适合业务中需要同步获取处理结果的场景例如通过 MQ 完成库存校验、数据计算必须拿到返回结果才能执行后续业务逻辑。1.7. Publisher Confirms 发布确认Publisher Confirms 是 RabbitMQ 用来保障生产者消息可靠投递的专属机制专门解决生产者侧消息丢失问题。系统中消息丢失分为三类场景一是生产者程序崩溃、网络抖动消息根本没发送到 Broker二是 Broker 收到消息后未持久化服务器宕机造成消息丢失这类问题依靠队列、消息持久化解决三是消费者处理异常且未手动确认消息被丢弃依靠消费者ACK机制处理。而发布确认只针对第一种生产者投递失败的场景开启后Broker完成消息接收与持久落盘才会返回确认一旦投递失败会通知生产者适合金融、订单等对数据安全要求严苛的业务。发布确认拥有固定底层运行逻辑生产者需要调用 channel.confirmSelect() 将信道设置为confirm 模式该信道下所有发出的消息都会分配从1开始自增的唯一deliveryTag序号。生产者发送消息时无需原地等待可以持续批量推送数据Broker处理完消息、完成路由与磁盘持久化后会异步向生产者推送回执回执分为 ACK 和 Nack 两种ACK 代表消息成功接收Nack 代表 Broker内部异常导致消息丢失回执中 multiple 参数用来区分单条与批量确认multiple 为 true 时当前deliveryTag 及之前所有消息都已成功送达false则仅代表单条消息确认结果。发布确认提供三种实现策略性能、复杂度各有差异。第一种是单条同步确认每发送一条消息就阻塞线程等待 Broker 回执确认完成才发送下一条优点是能精准定位失败消息但串行发送效率极低仅适合少量消息调试场景。第二种是批量同步确认累计一定数量消息后统一等待确认发送速度相比单条确认大幅提升缺点是一旦一批消息出现失败无法定位具体哪一条出问题只能整体重发网络波动场景会产生大量重复消息。第三种是异步回调确认无需阻塞发送流程通过注册 ConfirmListener 监听器接收 Broker 异步回执使用有序集合缓存所有未确认消息收到ACK/Nack时自动清理或处理失败数据该方案性能最优消息量越大优势越明显只是代码编写逻辑更复杂适合高并发大批量消息业务。三种策略有着清晰的适用区分单条同步确认适合消息量极小的调试场景批量同步确认适用于网络稳定的中小批量业务异步回调确认是线上高并发场景的首选。使用发布确认还有多项关键注意点。confirm 属于信道级特性不同 Channel 相互独立建议为不同业务单独分配信道如果交换机没有绑定有效队列消息会直接丢弃但 Broker 仍会返回ACK需要配合 mandatory 参数捕获无路由的消息Nack 仅由 Broker 内部故障触发网络中断不会推送 Nack需要生产者自行增加超时判断异步模式必须维护未确认消息集合否则无法实现失败消息重发最后该机制仅保证消息成功抵达 Broker无法保证消费者正常处理完整可靠投递还需要结合消费者手动 ACK 机制。二、Java API 代码实现2.1. Work Queue 工作队列工作队列模式代码层面和简单模式差异很小生产者可批量循环发送多条消息只需额外启动多个消费者实例即可实现竞争消费。公共常量package rabbitmq.constant; public class Constants { public static final String HOST 8.152.201.3; public static final int PORT 5672; public static final String USER_NAME yang; public static final String PASSWORD study; public static final String VIRTUALHOST test; // 工作队列模式 public static final String WORK_QUEUE work.queue; }生产者package rabbitmq.work; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import rabbitmq.constant.Constants; import java.io.IOException; import java.util.concurrent.TimeoutException; public class Producer { public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory connectionFactory new ConnectionFactory(); connectionFactory.setHost(Constants.HOST); connectionFactory.setPort(Constants.PORT); connectionFactory.setUsername(Constants.USER_NAME); connectionFactory.setPassword(Constants.PASSWORD); connectionFactory.setVirtualHost(Constants.VIRTUALHOST); Connection connection connectionFactory.newConnection(); Channel channel connection.createChannel(); // 队列不存在则创建 channel.queueDeclare(Constants.WORK_QUEUE, true, false, false, null); for (int i 0; i 10; i) { String message Hello RabbitMQ... i; channel.basicPublish(, Constants.WORK_QUEUE, null, message.getBytes()); } System.out.println(消息发送成功); channel.close(); connection.close(); } }package rabbitmq.work; import com.rabbitmq.client.*; import rabbitmq.constant.Constants; import java.io.IOException; import java.util.concurrent.TimeoutException; public class Consumer1 { public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory connectionFactory new ConnectionFactory(); connectionFactory.setHost(Constants.HOST); connectionFactory.setPort(Constants.PORT); connectionFactory.setUsername(Constants.USER_NAME); connectionFactory.setPassword(Constants.PASSWORD); connectionFactory.setVirtualHost(Constants.VIRTUALHOST); Connection connection connectionFactory.newConnection(); Channel channel connection.createChannel(); channel.queueDeclare(Constants.WORK_QUEUE, true, false, false, null); DefaultConsumer consumer new DefaultConsumer(channel) { Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println(接收信息 new String(body)); } }; channel.basicConsume(Constants.WORK_QUEUE, true, consumer); /*channel.close(); connection.close();*/ } }package rabbitmq.work; import com.rabbitmq.client.*; import rabbitmq.constant.Constants; import java.io.IOException; import java.util.concurrent.TimeoutException; public class Consumer2 { public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory connectionFactory new ConnectionFactory(); connectionFactory.setHost(Constants.HOST); connectionFactory.setPort(Constants.PORT); connectionFactory.setUsername(Constants.USER_NAME); connectionFactory.setPassword(Constants.PASSWORD); connectionFactory.setVirtualHost(Constants.VIRTUALHOST); Connection connection connectionFactory.newConnection(); Channel channel connection.createChannel(); channel.queueDeclare(Constants.WORK_QUEUE, true, false, false, null); DefaultConsumer consumer new DefaultConsumer(channel) { Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println(接收信息 new String(body)); } }; channel.basicConsume(Constants.WORK_QUEUE, true, consumer); /*channel.close(); connection.close();*/ } }先启动 Consumer1再启动 Consumer2最后启动 Poducer。2.2. Fanout 发布订阅代码实现上需要先声明持久化 fanout 交换机创建多个独立队列并全部绑定至交换机生产者发送消息时路由 key 传空不同消费者分别监听各自队列就能同时获取同一条广播消息。package rabbitmq.fanout; import com.rabbitmq.client.BuiltinExchangeType; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import rabbitmq.constant.Constants; import java.io.IOException; import java.util.concurrent.TimeoutException; public class Producer { public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory connectionFactory new ConnectionFactory(); connectionFactory.setHost(Constants.HOST); connectionFactory.setPort(Constants.PORT); connectionFactory.setUsername(Constants.USER_NAME); connectionFactory.setPassword(Constants.PASSWORD); connectionFactory.setVirtualHost(Constants.VIRTUALHOST); Connection connection connectionFactory.newConnection(); Channel channel connection.createChannel(); channel.exchangeDeclare(Constants.FANOUT_EXCHANGE, BuiltinExchangeType.FANOUT, true); channel.queueDeclare(Constants.FANOUT_QUEUE1, true, false, false, null); channel.queueDeclare(Constants.FANOUT_QUEUE2, true, false, false, null); channel.queueBind(Constants.FANOUT_QUEUE1, Constants.FANOUT_EXCHANGE, ); channel.queueBind(Constants.FANOUT_QUEUE2, Constants.FANOUT_EXCHANGE, ); String message hello fanout....; channel.basicPublish(Constants.FANOUT_EXCHANGE, , null, message.getBytes()); System.out.println(消息发送成功); channel.close(); connection.close(); } }package rabbitmq.fanout; import com.rabbitmq.client.*; import rabbitmq.constant.Constants; import java.io.IOException; import java.util.concurrent.TimeoutException; public class Consumer1 { public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory connectionFactory new ConnectionFactory(); connectionFactory.setHost(Constants.HOST); connectionFactory.setPort(Constants.PORT); connectionFactory.setUsername(Constants.USER_NAME); connectionFactory.setPassword(Constants.PASSWORD); connectionFactory.setVirtualHost(Constants.VIRTUALHOST); Connection connection connectionFactory.newConnection(); Channel channel connection.createChannel(); channel.queueDeclare(Constants.FANOUT_QUEUE1, true, false, false, null); DefaultConsumer consumer new DefaultConsumer(channel) { Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println(接收信息 new String(body)); } }; channel.basicConsume(Constants.FANOUT_QUEUE1, true, consumer); } }package rabbitmq.fanout; import com.rabbitmq.client.*; import rabbitmq.constant.Constants; import java.io.IOException; import java.util.concurrent.TimeoutException; public class Consumer2 { public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory connectionFactory new ConnectionFactory(); connectionFactory.setHost(Constants.HOST); connectionFactory.setPort(Constants.PORT); connectionFactory.setUsername(Constants.USER_NAME); connectionFactory.setPassword(Constants.PASSWORD); connectionFactory.setVirtualHost(Constants.VIRTUALHOST); Connection connection connectionFactory.newConnection(); Channel channel connection.createChannel(); channel.queueDeclare(Constants.FANOUT_QUEUE2, true, false, false, null); DefaultConsumer consumer new DefaultConsumer(channel) { Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println(接收信息 new String(body)); } }; channel.basicConsume(Constants.FANOUT_QUEUE2, true, consumer); } }2.3. Direct 路由路由模式有固定执行流程首先需要声明持久化的 Direct 交换机再创建多个独立消息队列随后将队列与交换机绑定并为每组绑定配置专属 BindingKey生产者发送消息时主动传入目标 RoutingKey消费者分别监听不同队列仅能接收路由键完全匹配的消息。package rabbitmq.direct; import com.rabbitmq.client.BuiltinExchangeType; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import rabbitmq.constant.Constants; import java.io.IOException; import java.util.concurrent.TimeoutException; public class Producer { public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory connectionFactory new ConnectionFactory(); connectionFactory.setHost(Constants.HOST); connectionFactory.setPort(Constants.PORT); connectionFactory.setUsername(Constants.USER_NAME); connectionFactory.setPassword(Constants.PASSWORD); connectionFactory.setVirtualHost(Constants.VIRTUALHOST); Connection connection connectionFactory.newConnection(); Channel channel connection.createChannel(); channel.exchangeDeclare(Constants.DIRECT_EXCHANGE, BuiltinExchangeType.DIRECT, true); channel.queueDeclare(Constants.DIRECT_QUEUE1, true, false, false, null); channel.queueDeclare(Constants.DIRECT_QUEUE2, true, false, false, null); channel.queueBind(Constants.DIRECT_QUEUE1, Constants.DIRECT_EXCHANGE, a); channel.queueBind(Constants.DIRECT_QUEUE2, Constants.DIRECT_EXCHANGE, a); channel.queueBind(Constants.DIRECT_QUEUE2, Constants.DIRECT_EXCHANGE, b); channel.queueBind(Constants.DIRECT_QUEUE2, Constants.DIRECT_EXCHANGE, c); String messageA hello direct, my routingKey is a; channel.basicPublish(Constants.DIRECT_EXCHANGE, a, null, messageA.getBytes()); String messageB hello direct, my routingKey is b; channel.basicPublish(Constants.DIRECT_EXCHANGE, b, null, messageB.getBytes()); String messageC hello direct, my routingKey is c; channel.basicPublish(Constants.DIRECT_EXCHANGE, c, null, messageC.getBytes()); System.out.println(消息发送成功); channel.close(); connection.close(); } }package rabbitmq.direct; import com.rabbitmq.client.*; import rabbitmq.constant.Constants; import java.io.IOException; import java.util.concurrent.TimeoutException; public class Consumer1 { public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory connectionFactory new ConnectionFactory(); connectionFactory.setHost(Constants.HOST); connectionFactory.setPort(Constants.PORT); connectionFactory.setUsername(Constants.USER_NAME); connectionFactory.setPassword(Constants.PASSWORD); connectionFactory.setVirtualHost(Constants.VIRTUALHOST); Connection connection connectionFactory.newConnection(); Channel channel connection.createChannel(); channel.queueDeclare(Constants.DIRECT_QUEUE1, true, false, false, null); DefaultConsumer consumer new DefaultConsumer(channel) { Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println(接收信息 new String(body)); } }; channel.basicConsume(Constants.DIRECT_QUEUE1, true, consumer); } }package rabbitmq.direct; import com.rabbitmq.client.*; import rabbitmq.constant.Constants; import java.io.IOException; import java.util.concurrent.TimeoutException; public class Consumer2 { public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory connectionFactory new ConnectionFactory(); connectionFactory.setHost(Constants.HOST); connectionFactory.setPort(Constants.PORT); connectionFactory.setUsername(Constants.USER_NAME); connectionFactory.setPassword(Constants.PASSWORD); connectionFactory.setVirtualHost(Constants.VIRTUALHOST); Connection connection connectionFactory.newConnection(); Channel channel connection.createChannel(); channel.queueDeclare(Constants.DIRECT_QUEUE2, true, false, false, null); DefaultConsumer consumer new DefaultConsumer(channel) { Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println(接收信息 new String(body)); } }; channel.basicConsume(Constants.DIRECT_QUEUE2, true, consumer); } }2.4. Topic 通配符首先声明 Topic 类型交换机创建多个业务队列队列绑定交换机时配置带有通配符的 BindingKey生产者发送由点分隔的多层级 RoutingKey队列会自动接收所有符合通配规则的消息。package rabbitmq.topic; import com.rabbitmq.client.BuiltinExchangeType; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import rabbitmq.constant.Constants; import java.io.IOException; import java.util.concurrent.TimeoutException; public class Producer { public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory connectionFactory new ConnectionFactory(); connectionFactory.setHost(Constants.HOST); connectionFactory.setPort(Constants.PORT); connectionFactory.setUsername(Constants.USER_NAME); connectionFactory.setPassword(Constants.PASSWORD); connectionFactory.setVirtualHost(Constants.VIRTUALHOST); Connection connection connectionFactory.newConnection(); Channel channel connection.createChannel(); channel.exchangeDeclare(Constants.TOPIC_EXCHANGE, BuiltinExchangeType.TOPIC, true); channel.queueDeclare(Constants.TOPIC_QUEUE1, true, false, false, null); channel.queueDeclare(Constants.TOPIC_QUEUE2, true, false, false, null); channel.queueBind(Constants.TOPIC_QUEUE1, Constants.TOPIC_EXCHANGE, *.a.*); channel.queueBind(Constants.TOPIC_QUEUE2, Constants.TOPIC_EXCHANGE, *.*.b); channel.queueBind(Constants.TOPIC_QUEUE2, Constants.TOPIC_EXCHANGE, c.#); String messageA hello topic, my routingKey is ae.a.f....; channel.basicPublish(Constants.TOPIC_EXCHANGE, ae.a.f, null, messageA.getBytes()); String messageB hello topic, my routingKey is ef.a.b....; channel.basicPublish(Constants.TOPIC_EXCHANGE, ef.a.b, null, messageB.getBytes()); String messageC hello topic, my routingKey is c.ef.d....; channel.basicPublish(Constants.TOPIC_EXCHANGE, c.ef.d, null, messageC.getBytes()); System.out.println(消息发送成功); channel.close(); connection.close(); } }package rabbitmq.topic; import com.rabbitmq.client.*; import rabbitmq.constant.Constants; import java.io.IOException; import java.util.concurrent.TimeoutException; public class Consumer1 { public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory connectionFactory new ConnectionFactory(); connectionFactory.setHost(Constants.HOST); connectionFactory.setPort(Constants.PORT); connectionFactory.setUsername(Constants.USER_NAME); connectionFactory.setPassword(Constants.PASSWORD); connectionFactory.setVirtualHost(Constants.VIRTUALHOST); Connection connection connectionFactory.newConnection(); Channel channel connection.createChannel(); channel.queueDeclare(Constants.TOPIC_QUEUE1, true, false, false, null); DefaultConsumer consumer new DefaultConsumer(channel) { Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println(接收信息 new String(body)); } }; channel.basicConsume(Constants.TOPIC_QUEUE1, true, consumer); } }package rabbitmq.topic; import com.rabbitmq.client.*; import rabbitmq.constant.Constants; import java.io.IOException; import java.util.concurrent.TimeoutException; public class Consumer2 { public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory connectionFactory new ConnectionFactory(); connectionFactory.setHost(Constants.HOST); connectionFactory.setPort(Constants.PORT); connectionFactory.setUsername(Constants.USER_NAME); connectionFactory.setPassword(Constants.PASSWORD); connectionFactory.setVirtualHost(Constants.VIRTUALHOST); Connection connection connectionFactory.newConnection(); Channel channel connection.createChannel(); channel.queueDeclare(Constants.TOPIC_QUEUE2, true, false, false, null); DefaultConsumer consumer new DefaultConsumer(channel) { Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println(接收信息 new String(body)); } }; channel.basicConsume(Constants.TOPIC_QUEUE2, true, consumer); } }2.5. RPC 模式分为 RPC 客户端与 RPC 服务端两部分客户端负责创建连接、生成唯一请求 ID、创建临时回调队列、发送请求、阻塞监听并打印返回结果服务端负责监听请求队列、配置 QoS 参数、执行业务逻辑、向回调队列返回结果、手动确认消息。客户端代码package rabbitmq.rpc; import com.rabbitmq.client.*; import rabbitmq.constant.Constants; import java.io.IOException; import java.util.UUID; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; import java.util.concurrent.TimeoutException; public class RpcClient { public static void main(String[] args) throws IOException, TimeoutException, InterruptedException { ConnectionFactory connectionFactory new ConnectionFactory(); connectionFactory.setHost(Constants.HOST); connectionFactory.setPort(Constants.PORT); connectionFactory.setUsername(Constants.USER_NAME); connectionFactory.setPassword(Constants.PASSWORD); connectionFactory.setVirtualHost(Constants.VIRTUALHOST); Connection connection connectionFactory.newConnection(); Channel channel connection.createChannel(); channel.queueDeclare(Constants.RPC_REQUEST_QUEUE, true, false, false, null); channel.queueDeclare(Constants.RPC_RESPONSE_QUEUE, true, false, false, null); // 发送请求 String message Hello, RPC; // 设置请求的唯一标识 String correlationID UUID.randomUUID().toString(); // 设置请求的相关属性 AMQP.BasicProperties props new AMQP.BasicProperties().builder() .correlationId(correlationID) .replyTo(Constants.RPC_RESPONSE_QUEUE) .build(); channel.basicPublish(, Constants.RPC_REQUEST_QUEUE, props, message.getBytes()); // 接收响应 final BlockingQueueString response new ArrayBlockingQueue(1); DefaultConsumer consumer new DefaultConsumer(channel) { Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { String respMessage new String(body); System.out.println(接收到回调消息 respMessage); if (correlationID.equals(properties.getCorrelationId())) { // ID 校验一致 response.offer(respMessage); } } }; channel.basicConsume(Constants.RPC_RESPONSE_QUEUE, true, consumer); String ret response.take(); System.out.println([RPC Client 响应结果] ret); } }package rabbitmq.rpc; import com.rabbitmq.client.*; import rabbitmq.constant.Constants; import java.io.IOException; import java.util.concurrent.TimeoutException; public class RpcServer { public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory connectionFactory new ConnectionFactory(); connectionFactory.setHost(Constants.HOST); connectionFactory.setPort(Constants.PORT); connectionFactory.setUsername(Constants.USER_NAME); connectionFactory.setPassword(Constants.PASSWORD); connectionFactory.setVirtualHost(Constants.VIRTUALHOST); Connection connection connectionFactory.newConnection(); Channel channel connection.createChannel(); // 接收请求 channel.basicQos(1); DefaultConsumer consumer new DefaultConsumer(channel) { Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { String request new String(body, UTF-8); System.out.println(接收到请求 request); String response 针对 request request 响应成功; AMQP.BasicProperties basicProperties new AMQP.BasicProperties().builder() .correlationId(properties.getCorrelationId()) .build(); channel.basicPublish(, Constants.RPC_RESPONSE_QUEUE, basicProperties, response.getBytes()); channel.basicAck(envelope.getDeliveryTag(), false); } }; channel.basicConsume(Constants.RPC_REQUEST_QUEUE, false, consumer); } }