消息队列在现今数据量大，并发量高的系统中是十分常用的。本文会对最常用到的几款消息队列框架 ActiveMQ、RabbitMQ、Kafka 进行分析对比。介绍 RabbitMQ 在 Spring 框架下的结构及实现原理，从Producer 端的事务、回调函数（ConfirmCallback / ReturnCallback）到 Consumer 端的 MessageListenerContainer 信息接收容器进行详细的分析。介绍 RabbitTemplate、SimpleMessageListenerContainer、DirectMessageListenerCon
前言
消息队列在现今数据量大，并发量高的系统中是十分常用的。本文将会对现时最常用到的几款消息队列框架 ActiveMQ、RabbitMQ、Kafka 进行分析对比。
详细介绍 RabbitMQ 在 Spring 框架下的结构及实现原理，从Producer 端的事务、回调函数（ConfirmCallback / ReturnCallback）到 Consumer 端的 MessageListenerContainer 信息接收容器进行详细的分析。通过对 RabbitTemplate、SimpleMessageListenerContainer、DirectMessageListenerContainer 等常用类型介绍，深入剖析在消息处理各个传输环节中的原理及注意事项。
并举以实例对死信队列、持久化操作进行一一介绍。
目录
一、RabbitMQ 与 AMQP 的关系
二、RabbitMQ 的实现原理
三、RabbitMQ 应用实例
四、Producer 端的消息发送与监控
五、Consumer 端的消息接收与监控
六、死信队列
七、持久化操作
一、RabbitMQ 与 AMQP 的关系
1.1 AMQP简介
AMQP（Advanced Message Queue Protocol 高级消息队列协议）是一个消息队列协议，它支持符合条件的客户端和消息代理中间件（message middleware broker）进行通讯。RabbitMQ 则是 AMQP 协议的实现者，主要用于在分布式系统中信息的存储发送与接收，RabbitMQ 的服务器端用 Erlang 语言编写，客户端支持多种开发语言：Python、.NET、Java、Ruby、C、PHP、ActionScript、XMPP、STOMP 等。
1.2 ActiveMQ、RabbitMQ、Kafka 对比
现在在市场上有 ActiveMQ、RabbitMQ、Kafka 等多个常用的消息队列框架，与其他框架对比起来，RabbitMQ 在易用性、扩展性、高可用性、多协议、支持多语言客户端等方面都有不俗表现。
1.2.1 AcitveMQ 特点
ActiveMQ 是 Apache 以 Java 语言开发的消息模型，它完美地支持 JMS（Java Message Service）消息服务，客户端支持 Java、C、C++、C#、Ruby、Perl、Python、PHP 等多种开主发语言，支持OpenWire、Stomp、REST、XMPP、AMQP 等多种协议。ActiveMQ 采用异步消息传递方式，在设计上保证了多主机集群，客户端-服务器，点对点等模式的有效通信。从开始它就是按照 JMS 1.1 和 J2EE 1.4 规范进行开发，实现了消息持久化，XA，事务支撑等功能。经历多年的升级完善，现今已成为 Java 应用开发中主流的消息解决方案。但相比起 RabbitMQ、Kafka 它的主要缺点表现为资源消耗比较大，吞吐量较低，在高并发的情况下系统支撑能力较弱。如果系统全程使用 Java 开发，其并发量在可控范围内，或系统需要支持多种不同的协议，使用 ActiveMQ 可更轻便地搭建起消息队列服务。
1.2.2 Kafka 特点
Kafka 天生是面向分布式系统开发的消息队列，它具有高性能、容灾性、可动态扩容等特点。Kafka 与生俱来的特点在于它会把每个Partition 的数据都备份到不同的服务器当中，并与 ZooKeeper 配合，当某个Broker 故障失效时，ZooKeeper 服务就会将通知生产者和消费者，从备份服务器进行数据恢复。在性能上 Kafka 也大大超越了传统的 ActiveMQ、RabbitMQ ，由于 Kafka 集群可支持动态扩容，在负载量到达峰值时可动态增加新的服务器进集群而无需重启服务。但由于 Kafka 属于分布式系统，所以它只能在同一分区内实现消息有序，无法实现全局消息有序。而且它内部的监控机制不够完善，需要安装插件，依赖ZooKeeper 进行元数据管理。如果系统属于分布式管理机制，数据量较大且并发量难以预估的情况下，建议使用 Kafka 队列。
1.2.3 RabbitMQ 对比
由于 ActiveMQ 过于依赖 JMS 的规范而限制了它的发展，所以 RabbitMQ 在性能和吞吐量上明显会优于 ActiveMQ。
由于上市时间较长，在可用性、稳定性、可靠性上 RabbitMq 会比 Kafka 技术成熟，而且 RabbitMq 使用 Erlang 开发，所以天生具备高并发高可用的特点。而 Kafka 属于分布式系统，它的性能、吞吐量、TPS 都会比 RabbitMq 要强。
回到目录
二、RabbitMQ 的实现原理
2.1 生产者（Producer）、消费者（Consumer）、服务中心（Broker）之间的关系
首先简单介绍 RabbitMQ 的运行原理，在 RabbitMQ 使用时，系统会先安装并启动 Broker Server，也就是 RabbitMQ 的服务中心。无论是生产者 （Producer），消费者（Consumer）都会通过连接池（Connection）使用 TCP/IP 协议（默认）来与 BrokerServer 进行连接。然后 Producer 会把 Exchange / Queue 的绑定信息发送到 Broker Server，Broker Server 根据 Exchange 的类型逻辑选择对应 Queue ，最后把信息发送到与 Queue 关联的对应 Consumer 。
2.2 交换器（Exchange）、队列（Queue）、信道（Channel）、绑定（Binding）的概念
2.2.1 交换器 Exchange
Producer 建立连接后，并非直接将消息投递到队列 Queue 中，而是把消息发送到交换器 Exchange，由 Exchange 根据不同逻辑把消息发送到一个或多个对应的队列当中。目前 Exchange 提供了四种不同的常用类型：Fanout、Direct、Topic、Header。
- Fanout类型
此类型是最为常见的交换器，它会将消息转发给所有与之绑定的队列上。比如，有N个队列与 Fanout 交换器绑定，当产生一条消息时，Exchange 会将该消息的N个副本分别发给每个队列，类似于广播机制。
- Direct类型
此类型的 Exchange 会把消息发送到 Routing_Key 完全相等的队列当中。多个 Cousumer 可以使用相同的关键字进行绑定，类似于数据库的一对多关系。比如，Producer 以 Direct 类型的 Exchange 推送 Routing_Key 为 direct.key1 的队列，系统再指定多个 Cousumer 绑定 direct.key1。如此，消息就会被分发至多个不同的 Cousumer 当中。
- Topic类型
此类型是最灵活的一种方式配置方式，它可以使用模糊匹配，根据 Routing_Key 绑定到包含该关键字的不同队列中。比如，Producer 使用 Topic类型的 Exchange 分别推送 Routing_Key 设置为 topic.guangdong.guangzhou 、topic.guangdong.shenzhen 的不同队列，Cousumer 只需要把 Routing_Key 设置为 topic.guangdong.# ，就可以把所有消息接收处理。
- Headers类型
该类型的交换器与前面介绍的稍有不同，它不再是基于关键字 Routing_Key 进行路由，而是基于多个属性进行路由的，这些属性比路由关键字更容易表示为消息的头。也就是说，用于路由的属性是取自于消息 Header 属性，当消息 Header 的值与队列绑定时指定的值相同时，消息就会路由至相应的队列中。
2.2.2 Queue 队列
Queue 队列是消息的载体，每个消息都会被投入到 Queue 当中，它包含 name，durable，arguments 等多个属性，name 用于定义它的名称，当 durable（持久化）为 true 时，队列将会持久化保存到硬盘上。反之为 false 时，一旦 Broker Server 被重启，对应的队列就会消失，后面还会有例子作详细介绍。
2.2.3 Channel 通道
当 Broker Server 使用 Connection 连接 Producer / Cousumer 时会使用到信道（Channel），一个 Connection上可以建立多个 Channel，每个 Channel 都有一个会话任务，可以理解为逻辑上的连接。主要用作管理相关的参数定义，发送消息，获取消息，事务处理等。
2.2.4 Binding 绑定
Binding 主要用于绑定交换器 Exchange 与 队列 Queue 之间的对应关系，并记录路由的 Routing-Key。Binding 信息会保存到系统当中，用于 Broker Server 信息的分发依据。
回到目录
三、RabbitMQ 应用实例
3.1 Rabbit 常用类说明
3.1.1 RabbitTemplate 类
Spring 框架已经封装了 RabbitTemplate 对 RabbitMQ 的绑定、队列发送、接收进行简化管理
方法说明void setExchange(String exchange) 设置绑定的 exchange 名称String getExchange()获取已绑定的 exchange 名称void setRoutingKey(String routingKey)设置绑定的 routingKeyString getRoutingKey()获取已绑定的 routingKeyvoid send(String exchange, String routingKey, Message message,CorrelationData data)以Message方式发送信息到 Broken Server,CorrelationData 为标示符可为空void convertAndSend(String exchange, String routingKey, Object object, CorrelationData data)以自定义对象方式发送信息到 Broken Server,系统将自动把 object转换成 Message,CorrelationData 为标示符可为空Message receive(String queueName, long timeoutMillis)根据queueuName接收队列发送Message信息Object receiveAndConvert(String queueName, long timeoutMillis)根据queueuName接收队列对象信息void setReceiveTimeout(long receiveTimeout)设置接收过期时间void setReplyTimeout(long replyTimeout)设置重发时间void setMandatory(boolean mandatory)开启强制委托模式（下文会详细说明）void setConfirmCallback(confirmCallback)绑定消息确认回调方法（下文会详细说明）void setReturnCallback(returnCallback)绑定消息退出回调方法（下文会详细说明）
3.2 初探 RabbitMQ
在官网下载并成功安装完 RabbitMQ 后，打开默认路径 http://localhost:15672/#/ 即可看到 RabbitMQ 服务中心的管理界面
3.2.1 Producer 端开发
先在 pom 中添加 RabbitMQ 的依赖，并在 application.yml 中加入 RabbitMQ 帐号密码等信息。此例子，我们尝试使用 Direct 交换器把队列发送到不同的 Consumer。
1 **********************pom ************************* 2 <project> 3 ............. 4 <dependency> 5 <groupId>org.springframework.bootgroupId> 6 <artifactId>spring-boot-starter-amqpartifactId> 7 <version>2.0.5.RELEASEversion> 8 dependency> 9 project>10 11 **************** application.yml ****************12 spring:13 application:14 name: rabbitMqProducer15 rabbitmq:16 host: localhost 17 port: 567218 username: admin19 password: 1234567820 virtual-host: /LeslieHost
首先使用 CachingConnectionFactory 建立链接，通过 BindingBuilder 绑定 Exchange、Queue、RoutingKey之间的关系。
然后通过 void convertAndSend (String exchange, String routingKey, Object object, CorrelationData data) 方法把信息发送到 Broken Server
1 @Configuration 2 public class ConnectionConfig { 3 @Value("${spring.rabbitmq.host}") 4 public String host; 5 6 @Value("${spring.rabbitmq.port}") 7 public int port; 8 9 @Value("${spring.rabbitmq.username}")10 public String username;11 12 @Value("${spring.rabbitmq.password}")13 public String password;14 15 @Value("${spring.rabbitmq.virtual-host}")16 public String virtualHost;17 18 @Bean19 public ConnectionFactory getConnectionFactory(){20 CachingConnectionFactory factory=new CachingConnectionFactory();21 factory.setHost(host);22 factory.setPort(port);23 factory.setUsername(username);24 factory.setPassword(password);25 factory.setVirtualHost(virtualHost);26 return factory;27 }28 }29 30 @Configuration31 public class BindingConfig {32 public final static String first="direct.first";33 public final static String second="direct.second";34 public final static String Exchange_NAME="directExchange";35 public final static String RoutingKey1="directKey1";36 public final static String RoutingKey2="directKey2";37 38 @Bean39 public Queue queueFirst(){40 return new Queue(first);41 }42 43 @Bean44 public Queue queueSecond(){45 return new Queue(second);46 }47 48 @Bean49 public DirectExchange directExchange(){50 return new DirectExchange(Exchange_NAME,true,true);51 }52 53 //利用BindingBuilder绑定Direct与queueFirst54 @Bean55 public Binding bindingExchangeFirst(Queue queueFirst, DirectExchange directExchange){56 return BindingBuilder.bind(queueFirst).to(directExchange).with(RoutingKey1);57 }58 59 //利用BindingBuilder绑定Direct与queueSecond60 @Bean61 public Binding bindingExchangeSecond(Queue queueSecond, DirectExchange directExchange){ 62 return BindingBuilder.bind(queueSecond).to(directExchange).with(RoutingKey2);63 } 64 }65 66 @Controller67 @RequestMapping("/producer")68 public class ProducerController {69 @Autowired70 private RabbitTemplate template;71 72 @RequestMapping("/send")73 public void send() {74 for(int n=0;n<100;n++){ 75 76 template.convertAndSend(BindingConfig.Exchange_NAME,BindingConfig.RoutingKey1,"I'm the first queue! "+String.valueOf(n),getCorrelationData());77 template.convertAndSend(BindingConfig.Exchange_NAME,BindingConfig.RoutingKey2,"I'm the second queue! "+String.valueOf(n),getCorrelationData());78 }79 }80 81 private CorrelationData getCorrelationData(){82 return new CorrelationData(UUID.randomUUID().toString());83 }84 }此时，打开 RabbitMQ 管理界面，可看到 Producer 已经向 Broken Server 的 direct.first / direct.second 两个 Queue 分别发送100 个 Message
3.2.2 Consumer 端开发
分别建立两个不同的 Consumer ，一个绑定 direct.first 别一个绑定 direct.second , 然后通过注解 @RabbitListener 监听不同的 queue，当接到到 Producer 推送队列时，显示队列信息。
1 @Configuration 2 public class ConnectionConfig { 3 @Value("${spring.rabbitmq.host}") 4 public String host; 5 6 @Value("${spring.rabbitmq.port}") 7 public int port; 8 9 @Value("${spring.rabbitmq.username}")10 public String username;11 12 @Value("${spring.rabbitmq.password}")13 public String password;14 15 @Value("${spring.rabbitmq.virtual-host}")16 public String virtualHost;17 18 @Bean19 public ConnectionFactory getConnectionFactory(){20 CachingConnectionFactory factory=new CachingConnectionFactory();21 factory.setHost(host);22 factory.setPort(port);23 factory.setUsername(username);24 factory.setPassword(password);25 factory.setVirtualHost(virtualHost);26 return factory;27 }28 }29 30 @Configuration31 public class BindingConfig {32 public final static String first="direct.first";33 public final static String Exchange_NAME="directExchange";34 public final static String RoutingKey1="directKey1";35 36 @Bean37 public Queue queueFirst(){38 return new Queue(first);39 }40 41 @Bean42 public DirectExchange directExchange(){43 return new DirectExchange(Exchange_NAME);44 }45 46 //利用BindingBuilder绑定Direct与queueFirst47 @Bean48 public Binding bindingExchangeFirst(Queue queueFirst, DirectExchange directExchange){49 return BindingBuilder.bind(queueFirst).to(directExchange).with(RoutingKey1);50 } 51 }52 53 @Configuration54 @RabbitListener(queues="direct.first")55 public class RabbitMqListener {56 57 @RabbitHandler58 public void handler(String message){59 System.out.println(message);60 }61 }62 63 @SpringBootApplication64 public class App {65 66 public static void main(String[] args){67 SpringApplication.run(App.class, args);68 }69 }运行后可以观察到不同的 Consumer 会收到不同队列的消息
如果觉得使用 Binding 代码绑定过于繁琐，还可以直接在监听类RabbitMqListener中使用 @QueueBinding 注解绑定
1 @Configuration 2 public class ConnectionConfig { 3 @Value("${spring.rabbitmq.host}") 4 public String host; 5 6 @Value("${spring.rabbitmq.port}") 7 public int port; 8 9 @Value("${spring.rabbitmq.username}")10 public String username;11 12 @Value("${spring.rabbitmq.password}")13 public String password;14 15 @Value("${spring.rabbitmq.virtual-host}")16 public String virtualHost;17 18 @Bean19 public ConnectionFactory getConnectionFactory(){20 CachingConnectionFactory factory=new CachingConnectionFactory();21 factory.setHost(host);22 factory.setPort(port);23 factory.setUsername(username);24 factory.setPassword(password);25 factory.setVirtualHost(virtualHost);26 return factory;27 }28 }29 30 @Configuration31 @RabbitListener(bindings=@QueueBinding(32 exchange=@Exchange(value="directExchange"),33 value=@Queue(value="direct.second"),34 key="directKey2"))35 public class RabbitMqListener {36 37 @RabbitHandler38 public void handler(String message){39 System.out.println(message);40 }41 }42 43 @SpringBootApplication44 public class App {45 46 public static void main(String[] args){47 SpringApplication.run(App.class, args);48 }49 }运行结果
回到目录
四、Producer 端的消息发送与监控
前面一节已经介绍了RabbitMQ的基本使用方法，这一节将从更深入的层面讲述 Producer 的应用。
试想一下这种的情形，如果因 RabbitTemplate 发送时 Exchange 名称绑定错误，或 Broken Server 因网络问题或服务负荷过大引发异常，Producer 发送的队列丢失，系统无法正常工作。此时，开发人员应该进行一系列应对措施进行监测，确保每个数据都能正常推送到 Broken Server 。有见及此，RabbitMQ 专门为大家提供了两种解决方案，一是使用传统的事务模式，二是使用回调函数，下面为大家作详介绍。
4.1 Producer 端的事务管理
在需要使用事务时，可以通过两种方法
第一可以调用 channel 类的方法以传统模式进行管理，事务开始时调用 channel.txSelect（），信息发送后进行确认 channel.txCommit（），一旦捕捉到异常进行回滚 channel.txRollback()，最后关闭事务。
1 @Controller 2 @RequestMapping("/producer") 3 public class ProducerController { 4 @Autowired 5 private RabbitTemplate template; 6 7 @RequestMapping("/send") 8 public void send1(HttpServletResponse response) 9 throws InterruptedException, IOException, TimeoutException{10 Channel channel=template.getConnectionFactory().createConnection().createChannel(true);11 .......12 try{13 channel.txSelect();14 channel.basicPublish("ErrorExchange", BindingConfig.Routing_Key_First, new AMQP.BasicProperties(),"Nothing".getBytes());15 channel.txCommit();16 }catch(Exception e){17 channel.txRollback();18 }finally{19 channel.close();20 }21 ......22 ......23 ......24 }25 }第二还可以直接通过 RabbitTemplate 的配置方法 void setChannelTransacted(bool isTransacted) 直接开启事务
1 public class ProducerController { 2 @Autowired 3 private ConnectionConfig connection; 4 5 @Autowired 6 @Bean 7 private RabbitTemplate template(){ 8 RabbitTemplate template=new RabbitTemplate(connection.getConnectionFactory()); 9 template.setChannelTransacted(true);10 return template;11 }12 13 @RequestMapping("/send")14 @Transactional(rollbackFor=Exception.class)15 public void send(HttpServletResponse response) throws InterruptedException, IOException，TimeoutException{16 ..........17 ..........18 ..........19 }20 }
4.2 利用 ConfirmCallback 回调确认消息是否成功发送到 Exchange
使用事务模式消耗的系统资源比较大，系统往往会处理长期等待的状态，在并发量较高的时候也有可能造成死锁的隐患。有见及此，系统提供了轻量级的回调函数方式进行异步处理。
当需要确认消息是否成功发送到 Exchange 的时候，可以使用 ConfirmCallback 回调函数。使用该函数，系统推送消息后，该线程便会得到释放，等 Exchange 接收到消息后系统便会异步调用 ConfirmCallback 绑定的方法进行处理。ConfirmCallback 只包含一个方法 void confirm(CorrelationData correlationData, boolean ack, String cause)，此方法会把每条数据发送到 Exchange 时候的 ack 状态（成功/失败），cause 成败原因，及对应的 correlationData（CorrelationData 只包含一个属性 id，是绑定发送对象的唯一标识符） 返还到 Producer，让Producer 进行相应处理。
注意：在绑定 ConfirmCallback 回调函数前，请先把 publisher-confirms 属性设置为 true
1 spring: 2 application: 3 name: rabbitmqproducer 4 rabbitmq: 5 host: 127.0.0.1 6 port: 5672 7 username: admin 8 password: 12345678 9 virtual-host: /LeslieHost
例如：下面的例子，特意将 RabbitTemplate 发送时所绑定的 Exchange 名称填写为错误名称 “ ErrorExchange ”，造成发送失败，然后在回调函数中检查失败的原因。
Producer 端代码：
1 @Configuration 2 public class ConnectionConfig { 3 @Value("${spring.rabbitmq.host}") 4 public String host; 5 6 @Value("${spring.rabbitmq.port}") 7 public int port; 8 9 @Value("${spring.rabbitmq.username}")10 public String username;11 12 @Value("${spring.rabbitmq.password}")13 public String password;14 15 @Value("${spring.rabbitmq.virtual-host}")16 public String virtualHost;17 18 @Bean19 public ConnectionFactory getConnectionFactory(){20 CachingConnectionFactory factory=new CachingConnectionFactory();21 System.out.println(host);22 factory.setHost(host);23 factory.setPort(port);24 factory.setUsername(username);25 factory.setPassword(password);26 factory.setVirtualHost(virtualHost);27 factory.setPublisherConfirms(true);28 factory.setPublisherReturns(true);29 return factory;30 }31 }32 33 @Configuration34 public class BindingConfig {35 public final static String first="direct.first";36 public final static String Exchange_NAME="directExchange";37 public final static String RoutingKey1="directKey1";38 39 @Bean40 public Queue queueFirst(){41 return new Queue(first);42 }43 44 @Bean45 public DirectExchange directExchange(){46 return new DirectExchange(Exchange_NAME);47 }48 49 @Bean50 public Binding bindingExchangeFirst(Queue queueFirst, DirectExchange directExchange){51 return BindingBuilder.bind(queueFirst).to(directExchange).with(RoutingKey1);52 } 53 }54 55 @Component56 public class MyConfirmCallback implements ConfirmCallback {57 58 @Override59 public void confirm(CorrelationData correlationData, boolean ack, String cause) {60 // TODO 自动生成的方法存根61 // TODO 自动生成的方法存根62 if(ack){63 System.out.println(correlationData.getId()+" ack is: true! \ncause:"+cause);64 }else65 System.out.println(correlationData.getId()+" ack is: false! \ncause:"+cause);66 }67 }68 69 @Controller70 @RequestMapping("/producer")71 public class ProducerController {72 @Autowired73 private RabbitTemplate template;74 @Autowired75 private MyConfirmCallback confirmCallback;76 77 @RequestMapping("/send")78 public void send() {79 template.setConfirmCallback(confirmCallback); 80 for(int n=0;n<2;n++){ 81 template.convertAndSend("ErrorExchange",82 BindingConfig.RoutingKey1,"I'm the first queue! "83 +String.valueOf(n),getCorrelationData());84 }85 }86 87 private CorrelationData getCorrelationData(){88 return new CorrelationData(UUID.randomUUID().toString());89 }90 }Consumer端代码
1 @Configuration 2 public class ConnectionConfig { 3 @Value("${spring.rabbitmq.host}") 4 public String host; 5 6 @Value("${spring.rabbitmq.port}") 7 public int port; 8 9 @Value("${spring.rabbitmq.username}")10 public String username;11 12 @Value("${spring.rabbitmq.password}")13 public String password;14 15 @Value("${spring.rabbitmq.virtual-host}")16 public String virtualHost;17 18 @Bean19 public ConnectionFactory getConnectionFactory(){20 CachingConnectionFactory factory=new CachingConnectionFactory();21 factory.setHost(host);22 factory.setPort(port);23 factory.setUsername(username);24 factory.setPassword(password);25 factory.setVirtualHost(virtualHost);26 return factory;27 }28 }29 30 @Configuration31 @RabbitListener(bindings=@QueueBinding(32 exchange=@Exchange(value="directExchange"),33 value=@Queue(value="direct.first"),34 key="directKey1"))35 public class RabbitMqListener {36 37 @RabbitHandler38 public void handler(String message){39 System.out.println(message);40 }41 }42 43 @SpringBootApplication44 public class App {45 46 public static void main(String[] args){47 SpringApplication.run(App.class, args);48 }49 }运行结果：
4.3 绑定 CorrelationData 与发送对象的关系
上面的例子当中，CorrelationData 只是用一个随机的 UUID 作为 CorrelationID，而在现实的应用场景中，由于 ConfirmCallback 只反回标识值 CorrelationData，而没有把队列里的对象值也一同返回。所以，在推送队列时可以先用 Key-Value 保存 CorrelationID 与所发送信息的关系，这样当 ConfirmCallback 回调时，就可根据 CorrelationID 找回对象，作进一步处理。
下面例子，我们把要发送的对象放在虚拟数据 DataSource 类中，用 DataRelation 记录 CorrelationID 与发送对象 OrderID 的关系，然后在回调函数 ConfirmCallback 中根据 CorrelationID 查找对应的 OrderEntity，如果发送成功，则删除绑定。如果发送失败，可以重新发送或根据情况再作处理。
Producer端代码：
1 @Configuration 2 public class ConnectionConfig { 3 @Value("${spring.rabbitmq.host}") 4 public String host; 5 6 @Value("${spring.rabbitmq.port}") 7 public int port; 8 9 @Value("${spring.rabbitmq.username}") 10 public String username; 11 12 @Value("${spring.rabbitmq.password}") 13 public String password; 14 15 @Value("${spring.rabbitmq.virtual-host}") 16 public String virtualHost; 17 18 @Bean 19 public ConnectionFactory getConnectionFactory(){ 20 CachingConnectionFactory factory=new CachingConnectionFactory(); 21 System.out.println(host); 22 factory.setHost(host); 23 factory.setPort(port); 24 factory.setUsername(username); 25 factory.setPassword(password); 26 factory.setVirtualHost(virtualHost); 27 factory.setPublisherConfirms(true); 28 factory.setPublisherReturns(true); 29 return factory; 30 } 31 } 32 33 @Configuration 34 public class BindingConfig { 35 public final static String first="direct.first"; 36 //Exchange 使用 direct 模式 37 public final static String Exchange_NAME="directExchange"; 38 public final static String RoutingKey1="directKey1"; 39 40 @Bean 41 public Queue queueFirst(){ 42 return new Queue(first); 43 } 44 45 @Bean 46 public DirectExchange directExchange(){ 47 return new DirectExchange(Exchange_NAME); 48 } 49 50 @Bean 51 public Binding bindingExchangeFirst(Queue queueFirst, DirectExchange directExchange){ 52 return BindingBuilder.bind(queueFirst).to(directExchange).with(RoutingKey1); 53 } 54 } 55 56 @Data 57 public class OrderEntity implements Serializable{ 58 private String id; 59 private String goods; 60 private Double price; 61 private Integer count; 62 63 public OrderEntity(String id,String goods,Double price,Integer count){ 64 this.id=id; 65 this.goods=goods; 66 this.price=price; 67 this.count=count; 68 } 69 70 public OrderEntity(){} 71 72 public String getId() { 73 return id; 74 } 75 public void setId(String id) { 76 this.id = id; 77 } 78 79 public String getGoods() { 80 return goods; 81 } 82 83 public void setGoodsId(String goods) { 84 this.goods = goods; 85 } 86 87 public Integer getCount() { 88 return count; 89 } 90 91 public void setCount(Integer count) { 92 this.count = count; 93 } 94 95 public Double getPrice() { 96 return price; 97 } 98 99 public void setPrice(Double price) {100 this.price = price;101 }102 }103 104 @Component105 public class DataSource {106 //加入虚拟数据107 private static Listlist=new ArrayList(108 Arrays.asList(new OrderEntity("001","Nikon D750",13990.00,1),109 new OrderEntity("002","Huwei P30 Plus",5400.00,1),110 ..........));111 112 public DataSource(){113 }114 115 public List getOrderList(){116 return list;117 }118 119 //根据Id获取对应order120 public OrderEntity getOrder(String id){121 for(OrderEntity order:list){122 if(order.getId()==id)123 return order;124 }125 return null;126 }127 }128 129 public class DataRelation {130 public static Map map=new HashMap();131 132 //绑定关系133 public static void add(String key,String value){134 if(!map.containsKey(key))135 map.put(key,value);136 }137 138 //返回orderId139 public static Object get(String key){140 if(map.containsKey(key))141 return map.get(key);142 else143 return null;144 }145 146 //根据 orderId 删除绑定关系147 public static void del(String key){148 if(map.containsKey(key))149 map.remove(key);150 }151 }152 153 @Component154 public class MyConfirmCallback implements ConfirmCallback {155 @Autowired156 private DataSource datasource;157 158 @Override159 public void confirm(CorrelationData correlationData, boolean ack, String cause) {160 String correlationId=correlationData.getId();161 //根据 correclationId取回对应的orderId162 String orderId=DataRelation.get(correlationId).toString();163 //在datasource中找回对应的order164 OrderEntity order=datasource.getOrder(orderId);165 166 if(ack){167 System.out.println("--------------------ConfirmCallback-------------------\n" 168 +" order's ack is true!\nId:"+order.getId()+" Goods:"+order.getGoods()169 +" Count:"+order.getCount().toString()+" Price:"+order.getPrice());170 DataRelation.del(correlationId); //操作完成删除对应绑定171 }else {172 System.out.println(order.getId()+" order's ack is: false! \ncause:"+cause);173 //可在记录日志后把Order推送到队列进行重新发送174 .......175 }176 }177 }178 179 @Controller180 @RequestMapping("/producer")181 public class ProducerController {182 @Autowired183 private RabbitTemplate template;184 @Autowired185 private MyConfirmCallback confirmCallback;186 @Autowired187 private DataSource dataSource;188 189 @RequestMapping("/send")190 public void send() throws InterruptedException, IOException{191 //绑定 ConfirmCallback 回调函数192 template.setConfirmCallback(confirmCallback);193 194 for(OrderEntity order:dataSource.getOrderList()){195 CorrelationData correlationData=getCorrelationData();196 //保存 CorrelationId 与 orderId关系197 DataRelation.add(correlationData.getId(), order.getId());198 //把 order 插入队列199 template.convertAndSend("directExchange",BindingConfig.RoutingKey1,order,correlationData);200 }201 }202 203 private CorrelationData getCorrelationData(){204 return new CorrelationData(UUID.randomUUID().toString());205 }206 }Consumer 端代码
1 @Configuration 2 public class ConnectionConfig { 3 @Value("${spring.rabbitmq.host}") 4 public String host; 5 6 @Value("${spring.rabbitmq.port}") 7 public int port; 8 9 @Value("${spring.rabbitmq.username}")10 public String username;11 12 @Value("${spring.rabbitmq.password}")13 public String password;14 15 @Value("${spring.rabbitmq.virtual-host}")16 public String virtualHost;17 18 @Bean19 public ConnectionFactory getConnectionFactory(){20 CachingConnectionFactory factory=new CachingConnectionFactory();21 factory.setHost(host);22 factory.setPort(port);23 factory.setUsername(username);24 factory.setPassword(password);25 factory.setVirtualHost(virtualHost);26 return factory;27 }28 }29 30 @Configuration31 @RabbitListener(bindings=@QueueBinding(32 exchange=@Exchange(value="directExchange"),33 value=@Queue(value="direct.first"),34 key="directKey1"))35 public class RabbitMqListener {36 37 @RabbitHandler38 public void handler(String message){39 System.out.println(message);40 }41 }42 43 @SpringBootApplication44 public class App {45 46 public static void main(String[] args){47 SpringApplication.run(App.class, args);48 }49 }运行结果
4.4 利用 ReturnCallback 处理队列 Queue 错误
使用 ConfirmCallback 函数只能判断消息是否成功发送到 Exchange，但并不能保证消息已经成功进行队列 Queue。所以，系统预备了另一个回调函数 ReturnCallback 来监听 Queue 队列处理的成败。如果队列错误绑定不存在的 queue，或者 Broken Server 瞬间出现问题末能找到对应的 queue，系统就会激发 Producer 端 ReturnCallback 的回调函数来进行错误处理。 ReturnCallback 回调接口只包含一个方法 void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey)，它会把出错的 replyCode，replyText，exchange，routingKey等值都一起返还。与 ConfirmCallback 不同的是，returnedMessage 会把队列中的对象保存到 Message 的 Body 属性中并返还到回调函数。
注意：在绑定 ReturnCallback 回调函数前，请先把 publisher-returns 及 mandatory 属性设置为 true 。 mandatory 参数默认为 false，用于判断 broken server是否把错误的对象返还到 Producer。如末进行设置，系统将把错误的消息丢弃。
下面例子我们在调用 convertAndSend 方法时特意把 routingKey 设置为 ErrorKey，触发 ReturnCallback 回调，然后在 ReturenCallback 的回调方法显示 replyCode，replyText，exchange，routingKey 等值，并把队列中对象属性一并显示。
Producer 端代码
1 @Configuration 2 public class ConnectionConfig { 3 @Value("${spring.rabbitmq.host}") 4 public String host; 5 6 @Value("${spring.rabbitmq.port}") 7 public int port; 8 9 @Value("${spring.rabbitmq.username}") 10 public String username; 11 12 @Value("${spring.rabbitmq.password}") 13 public String password; 14 15 @Value("${spring.rabbitmq.virtual-host}") 16 public String virtualHost; 17 18 @Bean 19 public ConnectionFactory getConnectionFactory(){ 20 CachingConnectionFactory factory=new CachingConnectionFactory(); 21 System.out.println(host); 22 factory.setHost(host); 23 factory.setPort(port); 24 factory.setUsername(username); 25 factory.setPassword(password); 26 factory.setVirtualHost(virtualHost); 27 factory.setPublisherConfirms(true); 28 factory.setPublisherReturns(true); 29 return factory; 30 } 31 } 32 33 @Configuration 34 public class BindingConfig { 35 public final static String first="direct.first"; 36 public final static String Exchange_NAME="directExchange"; 37 public final static String RoutingKey1="directKey1"; 38 39 @Bean 40 public Queue queueFirst(){ 41 return new Queue(first); 42 } 43 44 @Bean 45 public DirectExchange directExchange(){ 46 return new DirectExchange(Exchange_NAME); 47 } 48 49 @Bean 50 public Binding bindingExchangeFirst(Queue queueFirst, DirectExchange directExchange){ 51 return BindingBuilder.bind(queueFirst).to(directExchange).with(RoutingKey1); 52 } 53 } 54 55 @Data 56 public class OrderEntity implements Serializable{ 57 private String id; 58 private String goods; 59 private Double price; 60 private Integer count; 61 62 public OrderEntity(String id,String goods,Double price,Integer count){ 63 this.id=id; 64 this.goods=goods; 65 this.price=price; 66 this.count=count; 67 } 68 69 public OrderEntity(){} 70 71 public String getId() { 72 return id; 73 } 74 public void setId(String id) { 75 this.id = id; 76 } 77 78 public String getGoods() { 79 return goods; 80 } 81 82 public void setGoodsId(String goods) { 83 this.goods = goods; 84 } 85 86 public Integer getCount() { 87 return count; 88 } 89 90 public void setCount(Integer count) { 91 this.count = count; 92 } 93 94 public Double getPrice() { 95 return price; 96 } 97 98 public void setPrice(Double price) { 99 this.price = price;100 }101 }102 103 @Component104 public class DataSource {105 //虚拟数据106 private static Listlist=new ArrayList(107 Arrays.asList(new OrderEntity("001","Nikon D750",13990.00,1),108 new OrderEntity("002","Huwei P30 Plus",5400.00,1),109 ......));110 public DataSource(){111 }112 113 public List getOrderList(){114 return list;115 }116 117 //根据Id获取对应order118 public OrderEntity getOrder(String id){119 for(OrderEntity order:list){120 if(order.getId()==id)121 return order;122 }123 return null;124 }125 }126 127 @Component128 public class MyReturnCallback implements ReturnCallback {129 130 @Override131 public void returnedMessage(Message message, int replyCode, 132 String replyText, String exchange, String routingKey){133 //把messageBody反序列化为 OrderEntity对象134 OrderEntity order=convertToOrder(message.getBody());135 //显示错误原因136 System.out.println("-------------ReturnCallback!------------\n"137 +" exchange:"+exchange+" replyCode:"+String.valueOf(replyCode)138 +" replyText:"+replyText+" key:"+routingKey+"\n OrderId:"+order.getId()139 +" Goods:"+order.getGoods()+" Count:"+order.getCount().toString()140 +" Price:"+order.getPrice()+" ");141 }142 143 //把byte[]反序列化为 OrderEntity对象144 private OrderEntity convertToOrder(byte[] bytes){145 OrderEntity order=null;146 ByteArrayInputStream bis = new ByteArrayInputStream (bytes); 147 ObjectInputStream ois;148 try {149 ois = new ObjectInputStream (bis);150 Object obj = ois.readObject();151 order=(OrderEntity)obj;152 ois.close(); 153 bis.close(); 154 } catch (IOException | ClassNotFoundException e) {155 // TODO 自动生成的 catch 块156 e.printStackTrace();157 } 158 return order;159 }160 }161 162 @Controller163 @RequestMapping("/producer")164 public class ProducerController {165 @Autowired166 private RabbitTemplate template;167 @Autowired168 private MyReturnCallback returnCallback;169 @Autowired170 private DataSource dataSource;171 172 173 @RequestMapping("/send")174 public void send() throws InterruptedException, IOException{175 //把 mandatory 属性设定为true176 template.setMandatory(true);177 //绑定 ReturnCallback 回调函数178 template.setReturnCallback(returnCallback);179 180 for(OrderEntity order:dataSource.getOrderList()){181 CorrelationData correlationData=getCorrelationData();182 template.convertAndSend("directExchange","ErrorKey",orde�ƺ�,����r,correlationData);183 }184 }185 186 private CorrelationData getCorrelationData(){187 return new CorrelationData(UUID.randomUUID().toString());188 }189 }Consumer 代码
1 @Configuration 2 public class ConnectionConfig { 3 @Value("${spring.rabbitmq.host}") 4 public String host; 5 6 @Value("${spring.rabbitmq.port}") 7 public int port; 8 9 @Value("${spring.rabbitmq.username}")10 public String username;11 12 @Value("${spring.rabbitmq.password}")13 public String password;14 15 @Value("${spring.rabbitmq.virtual-host}")16 public String virtualHost;17 18 @Bean19 public ConnectionFactory getConnectionFactory(){20 CachingConnectionFactory factory=new CachingConnectionFactory();21 factory.setHost(host);22 factory.setPort(port);23 factory.setUsername(username);24 factory.setPassword(password);25 factory.setVirtualHost(virtualHost);26 return factory;27 }28 }29 30 @Configuration31 @RabbitListener(bindings=@QueueBinding(32 exchange=@Exchange(value="directExchange"),33 value=@Queue(value="direct.first"),34 key="directKey1"))35 public class RabbitMqListener {36 37 @RabbitHandler38 public void handler(String message){39 System.out.println(message);40 }41 }42 43 @SpringBootApplication44 public class App {45 46 public static void main(String[] args){47 SpringApplication.run(App.class, args);48 }49 }运行结果：
回到目录
五、Consumer 消息接收管控
在第四节主要介绍了 Producer 端的队列发送与监控，它只能管理 Producer 与 Broker Server 之间的通信，但并不能确认 Consumer 是否能成功接收到队列，在这节内容将介绍 Consumer 端的队列接收与监听。前面几节里，Consumer 端都是简单地直接使用 RabbitListener 对队列进行监听，其实 RabbitMQ 已经为用户准备了功能更强大的 MessageListenerContainer 容器用于管理 Message ，下面将为大家介绍。
5.1 AbstractMessageListenerContainer 介绍
AbstractMeessageListenerContainer 虚拟类是 RabbitMQ 封装好的一个容器，本身并没有对消息进行处理，而是把消息的处理方式交给了 MessageListener 。而它的主要功能是实现 MessageListener 的绑定，ApplicationContext 上下文的绑定，ErrorHandler 的错误处理方法的绑定、对消息消费的开始、结束等等默认参数进行配置，让开发人员可以在容器中对 Consumer 实现统一管理。SimpleMessageListenerContainer、DirectMessageLinstenerCoontainer 都是它的子类，分别应用于不同的场景，在下面会再作详细介绍。
方法说明void setAcknowledgeMode(AcknowledgeMode acknowledgeMode)设定消息接收确认的模式（下文会有详细介绍）AcknowledgeMode getAcknowledgeMode()获取消息接收确认的模式（下文会有详细介绍）void setPrefetchCount(int prefetchCount)设置每个 consumer 每次可接收到消息的最大数量void setQueues(Queue... queues)设定监听Queue队列void addQueues(Queue... queues)加入监听Queue队列void setMessageListener(Object messageListener)绑定MessageListener，对信息进行处理void setChannelAwareMessageListener(ChannelAwareMessageListener messageListener)绑定ChannelAwareMessageListener,对信息进行处理，同时可获取当前使用的channel信息Object getMessageListener()获取MessageListener对象void setMessageConverter(MessageConverter messageConverter)绑定MessageConverter消息转换对象 void setApplicationContext(ApplicationContext applicationContext)绑定ApplicationContext上下文ConnectionFactory getConnectionFactory()获取ConnectionFactory连接工厂void setListenerId(String listenerId)设定ListenerIdMessageListener 是监听消息最常用 Listener，它只包含了一个方法 void onMessage(Message message)，这是消息接收最常用的一个方法，开发者只需要实现此方法即可对接收到的 Message 进行处理。
ChannelAwareMessageListener 相当于是 MessageListener的一个扩展，包含了方法 void onMessage(Message message, Channel channel)，除了对 Message 进行处理外，还可以对接收此 Message 的 Channel 进行检测。
5.2 SimpleMessageListenerContainer 常用方法
SimpleMessageListenerContainer 是最常用的 MessageListener 容器，它可以通过下面的方法设置默认消费者数量与最大的消费者数量。下面例子中尝试把 consurrentConsumers 设置为3，把maxConcurrentConsumers 设置为4，并同时监控 direct 模式交换器的 direct.first，direct.second 队列。
方法说明void setConcurrentConsumers(final int concurrentConsumers)设定当前队列中消费者数量void setMaxConcurrentConsumers(int maxConcurrentConsumers)设定当前队列中最大消费者数量通过截图可以看到，系统默认会为每个 queue 都创建 3 个 consumers，不同的 queue 中的 consumers 是共享相同的 3 个 channel 。
当 Producer 端发送消息时，consumers 的实际数量可根据 maxConcurrentConsumers 的配置限制进行扩展。
Producer 端代码
1 @Configuration 2 public class BindingConfig { 3 public final static String first="direct.first"; 4 public final static String second="direct.second"; 5 public final static String Exchange_NAME="directExchange"; 6 public final static String RoutingKey1="directKey1"; 7 public final static String RoutingKey2="directKey2"; 8 9 @Bean10 public Queue queueFirst(){11 return new Queue(first);12 }13 14 @Bean15 public Queue queueSecond(){16 return new Queue(second);17 }18 19 @Bean20 public DirectExchange directExchange(){21 return new DirectExchange(Exchange_NAME);22 }23 24 //利用BindingBuilder绑定Direct与queueFirst25 @Bean26 public Binding bindingExchangeFirst(Queue queueFirst, DirectExchange directExchange){27 return BindingBuilder.bind(queueFirst).to(directExchange).with(RoutingKey1);28 }29 30 //利用BindingBuilder绑定Direct与queueSecond31 @Bean32 public Binding bindingExchangeSecond(Queue queueSecond, DirectExchange directExchange){ 33 return BindingBuilder.bind(queueSecond).to(directExchange).with(RoutingKey2);34 } 35 }36 37 @Configuration38 public class ConnectionConfig {39 @Value("${spring.rabbitmq.host}")40 public String host;41 42 @Value("${spring.rabbitmq.port}")43 public int port;44 45 @Value("${spring.rabbitmq.username}")46 public String username;47 48 @Value("${spring.rabbitmq.password}")49 public String password;50 51 @Value("${spring.rabbitmq.virtual-host}")52 public String virtualHost;53 54 @Bean55 public ConnectionFactory getConnectionFactory(){56 CachingConnectionFactory factory=new CachingConnectionFactory();57 factory.setHost(host);58 factory.setPort(port);59 factory.setUsername(username);60 factory.setPassword(password);61 factory.setVirtualHost(virtualHost);62 return factory;63 }64 }65 66 @Controller67 @RequestMapping("/producer")68 public class ProducerController {69 @Autowired70 private RabbitTemplate template;71 72 @RequestMapping("/send")73 public void send(HttpServletResponse response) throws InterruptedException, IOException{74 for(Integer n=0;n<100;n++){75 CorrelationData correlationData=getCorrelationData();76 template.convertAndSend("directExchange","directKey1", 77 "queue1"+" "+n.toString(),correlationData);78 template.convertAndSend("directExchange","directKey2"," queue2"+" "+n.toString(),correlationData); 79 Thread.currentThread().sleep(30);80 }81 }82 83 private CorrelationData getCorrelationData(){84 return new CorrelationData(UUID.randomUUID().toString());85 }86 }Consumer 端代码：
1 @Configuration 2 public class ConnectionConfig { 3 @Value("${spring.rabbitmq.host}") 4 public String host; 5 6 @Value("${spring.rabbitmq.port}") 7 public int port; 8 9 @Value("${spring.rabbitmq.username}") 10 public String username; 11 12 @Value("${spring.rabbitmq.password}") 13 public String password; 14 15 @Value("${spring.rabbitmq.virtual-host}") 16 public String virtualHost; 17 18 @Bean 19 public ConnectionFactory getConnectionFactory(){ 20 CachingConnectionFactory factory=new CachingConnectionFactory(); 21 factory.setHost(host); 22 factory.setPort(port); 23 factory.setUsername(username); 24 factory.setPassword(password); 25 factory.setVirtualHost(virtualHost); 26 return factory; 27 } 28 } 29 30 @Configuration 31 public class BindingConfig { 32 public final static String first="direct.first"; 33 public final static String second="direct.second"; 34 public final static String Exchange_NAME="directExchange"; 35 public final static String RoutingKey1="directKey1"; 36 public final static String RoutingKey2="directKey2"; 37 38 @Bean 39 public Queue queueFirst(){ 40 return new Queue(first); 41 } 42 43 @Bean 44 public Queue queueSecond(){ 45 return new Queue(second); 46 } 47 48 @Bean 49 public DirectExchange directExchange(){ 50 return new DirectExchange(Exchange_NAME); 51 } 52 53 //利用BindingBuilder绑定Direct与queueFirst 54 @Bean 55 public Binding bindingExchangeFirst(Queue queueFirst, DirectExchange directExchange){ 56 return BindingBuilder.bind(queueFirst).to(directExchange).with(RoutingKey1); 57 } 58 59 //利用BindingBuilder绑定Direct与queueSecond 60 @Bean 61 public Binding bindingExchangeSecond(Queue queueSecond, DirectExchange directExchange){ 62 return BindingBuilder.bind(queueSecond).to(directExchange).with(RoutingKey2); 63 } 64 } 65 @Configuration 66 public class SimpleMessListener { 67 @Autowired 68 private RabbitTemplate template; 69 private int index=0; 70 71 @Bean 72 public SimpleMessageListenerContainer messageContainer(){ 73 SimpleMessageListenerContainer container=new SimpleMessageListenerContainer(); 74 container.setConnectionFactory(connectionConfig.getConnectionFactory()); 75 // 绑定Queue1/Queue2 76 container.setQueueNames("direct.first"); 77 container.addQueueNames("direct.second"); 78 //设置默认 consumer 数为3 79 container.setConcurrentConsumers(3); 80 //设置最大 consumer 数为4 81 container.setMaxConcurrentConsumers(4); 82 //标记 consumerTag 83 container.setConsumerTagStrategy(queue -> "consumer"+(++index)); 84 //绑定MessageListener显示接收信息 85 container.setMessageListener(new MessageListener(){ 86 @Override 87 public void onMessage(Message message) { 88 // TODO 自动生成的方法存根 89 Thread thread=Thread.currentThread(); 90 MessageProperties messProp=message.getMessageProperties(); 91 try { 92 System.out.println(" ConsumerTag:"+messProp.getConsumerTag() 93 +" ThreadId is:"+thread.getId()+" Queue:"+messProp.getConsumerQueue() 94 +" "+new String(message.getBody(),"UTF-8")); 95 } catch (UnsupportedEncodingException e) { 96 // TODO 自动生成的 catch 块 97 e.printStackTrace(); 98 } 99 }100 101 });102 return container;103 }104 }运行结果
5.3 SimpleMessageListenerContainer 的运作原理
在 SimpleMessageListenerContainer 模式中，无论系统监听多少个 queue 队列，channel 都是共享的，类似上面的例子，4个 channel 会把接收到不同的队列请求并分发到对应的 consumer 进行处理。这样做的好处是系统可以通过 concurrentConsumers、maxConcurrentConsumers 灵活设定当前队列中消费者的数量，系统可以跟据实际需求灵活处理。但由于每个 channel 都是在固定线程中运行的，一个 channel 要游走于多个 consumer 当中，这无疑增加了系统在上下文切换中的开销。下面用系统提供的 ChannelAwareMessageListener 接口，以更直观的例子说明一下 SimpleMessageListenerContainer 当中 channel、queue、consumer 之间的关系。
Producer 端代码
1 @Configuration 2 public class BindingConfig { 3 public final static String first="direct.first"; 4 public final static String second="direct.second"; 5 public final static String Exchange_NAME="directExchange"; 6 public final static String RoutingKey1="directKey1"; 7 public final static String RoutingKey2="directKey2"; 8 9 @Bean10 public Queue queueFirst(){11 return new Queue(first);12 }13 14 @Bean15 public Queue queueSecond(){16 return new Queue(second);17 }18 19 @Bean20 public DirectExchange directExchange(){21 return new DirectExchange(Exchange_NAME);22 }23 24 //利用BindingBuilder绑定Direct与queueFirst25 @Bean26 public Binding bindingExchangeFirst(Queue queueFirst, DirectExchange directExchange){27 return BindingBuilder.bind(queueFirst).to(directExchange).with(RoutingKey1);28 }29 30 //利用BindingBuilder绑定Direct与queueSecond31 @Bean32 public Binding bindingExchangeSecond(Queue queueSecond, DirectExchange directExchange){ 33 return BindingBuilder.bind(queueSecond).to(directExchange).with(RoutingKey2);34 } 35 }36 37 @Configuration38 public class ConnectionConfig {39 @Value("${spring.rabbitmq.host}")40 public String host;41 42 @Value("${spring.rabbitmq.port}")43 public int port;44 45 @Value("${spring.rabbitmq.username}")46 public String username;47 48 @Value("${spring.rabbitmq.password}")49 public String password;50 51 @Value("${spring.rabbitmq.virtual-host}")52 public String virtualHost;53 54 @Bean55 public ConnectionFactory getConnectionFactory(){56 CachingConnectionFactory factory=new CachingConnectionFactory();57 factory.setHost(host);58 factory.setPort(port);59 factory.setUsername(username);60