⭐⭐⭐ Spring Boot 项目实战	⭐⭐⭐ Spring Cloud 项目实战
《Dubbo 实现原理与源码解析 —— 精品合集》	《Netty 实现原理与源码解析 —— 精品合集》
《Spring 实现原理与源码解析 —— 精品合集》	《MyBatis 实现原理与源码解析 —— 精品合集》
《Spring MVC 实现原理与源码解析 —— 精品合集》	《数据库实体设计合集》
《Spring Boot 实现原理与源码解析 —— 精品合集》	《Java 面试题 + Java 学习指南》

摘要: 原创出处 blog.csdn.net/weixin_43189971/article/details/126256175 「谁是黄黄」欢迎转载，保留摘要，谢谢！

🙂🙂🙂关注微信公众号：【芋道源码】有福利：

RocketMQ / MyCAT / Sharding-JDBC 所有源码分析文章列表

RocketMQ / MyCAT / Sharding-JDBC 中文注释源码 GitHub 地址

您对于源码的疑问每条留言都将得到认真回复。甚至不知道如何读源码也可以请教噢。

新的源码解析文章实时收到通知。每周更新一篇左右。

认真的源码交流微信群。

前面介绍了 RabbitMQ 基础入门、安装与简单使用、基础配置、管理界面介绍、常用 API等相关的知识点，今天我将详细的为大家介绍 RabbitMQ 消息收发与交换机相关知识，希望大家能够从中收获多多！如有帮助，请点在看、转发支持一波！！！

七种消息收发方式

Hello World

1.先由生产者发送到消息交换机（有默认，可配置）
2.消息交换机根据既定的策略帮你把消息路由到不同的队列上面去(意味着消息收发的形式更加的多样)
3.消息消费者就去监听不同的队列

work

默认情况下，一个消费者中，开启一个线程进行消息的处理，意味着同一时间只能处理一条消息。

我们可以通过配置，让一个消费者中，同时存在多个子线程，每一个线程都可以去消费消息。

即消费者类中，在添加一个消费者即可，关键代码 RabbitConsumer类中。

一个生产者，一个默认的交换机，一个队列，两个消费者
即两个消费者监听同一个队列

@Component
public class RabbitConsumer {
       private static  final Logger logger = LoggerFactory.getLogger(RabbitConsumer.class);

    /**
     * @RabbitListener(queues = "hello-queue")
     * 通过这个注解，来指定该方法需要监听的消息队列，注解中的参数就是要去监听的消息队列的名称。
     * @param msg
     */
    @RabbitListener(queues = "hello-queue")
       public  void  handleMsg(String msg){
           logger.info("msg1:{}",msg);
       }

    @RabbitListener(queues = "hello-queue")
    public  void  handleMsg2(String msg){
        logger.info("msg:{}",msg);
    }
}

界面显示的是两个通道

一个生产者，一个交换机，两个队列，两个消费者

1.声明一个Fanout类型的exchange，并且将exchange和queue绑定在一起，绑定的方式就是直接绑定。
2.让生产者创建一个exchange并且指定类型，和一个或多个队列绑定到一起。

1.4 Routing

一个生产者，一个交换机，两个队列，两个消费者与1.3.1中的Direct直连十分相似

Topics

一个生产者，一个交换机，两个队列，两个消费者，生产者创建 Topic 的Exchange 并且绑定到队列中，这次绑定可以通过 * 和 # 关键字，对指定 RoutingKey 内容，编写时注意格式 xxx.xxx.xxx去编写。

与主机交换Topic十分相似

RPC

什么是RPC：（Remote Procedure Call Protocol 远程过程调用协议），后面的跨进程调用都是用到了这个

消息中间件怎么实现解耦的功能？

1.a调用b的时候b必须在线不在线就失败了
2.我们可以用消息中间件解决，a调用b就是想用b的服务。
具体过程是：
a给消息中间件发送消息，b去消费那个消息，消费完了之后，b也给消息中间件发送一个消息，然后a再去消费那个消息，这样就通过消息中间件完成了解耦
就是a把消息扔到消息中间件后就可以不用管了，如果b在线立马就会消费，如果b不在线那个消息本身也不会丢失，等b上线了就会把a丢在消息中间件的消息给消费了。

RPC调用说的基本上就是这么个情况

简单架构图

首先Client发送一条消息，和普通的消息相比，这条消息多了两个关键内容：

correlation_id,这个表示这条消息的唯一id，还有一个内容是reply_to,这个表示消息回复队列的名字。
Server从消息发送队列获取消息并且处理响应的业务逻辑，处理完成之后，再将处理结果发送到reply_to指定的回调队列中。
Client从回调队列中读取消息，就可以知道消息的执行情况是什么样子了。

RPC 调用的小例子

建maven

我们先创建一个客户端：rpc_client ，再创建一个服务端：rpc_server。

依赖rpc_client

 <dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-amqp</artifactId>
  </dependency>
  <dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-web</artifactId>
  </dependency>

  <dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-test</artifactId>
   <scope>test</scope>
  </dependency>
  <dependency>
   <groupId>org.springframework.amqp</groupId>
   <artifactId>spring-rabbit-test</artifactId>
   <scope>test</scope>
  </dependency>
rpc_server
<dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-amqp</artifactId>
  </dependency>
  <dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-web</artifactId>
  </dependency>

  <dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-test</artifactId>
   <scope>test</scope>
  </dependency>
  <dependency>
   <groupId>org.springframework.amqp</groupId>
   <artifactId>spring-rabbit-test</artifactId>
   <scope>test</scope>
  </dependency>

application配置

rpc_client
spring.rabbitmq.host=1.12.235.192
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.virtual-host=/

#添加了这个属性之后，将来发送的消息会自动携带一个id,通过这个id我们可以识别出自己想要的消息
spring.rabbitmq.publisher-confirm-type=correlated
spring.rabbitmq.publisher-returns=true
rpc_server

1.配置correlation_id，将发送的消息和返回值之间关联起来
2.开启发送失败退回

spring.rabbitmq.host=1.12.235.192
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.virtual-host=/

server.port=8085
#只有开启了这个配置，将来的消息中才会带 correlation_id，只有通过correlation_id 我们才能将发送的消息和返回值之间关联起来
spring.rabbitmq.publisher-confirm-type=correlated
#是开启发送失败退回。
spring.rabbitmq.publisher-returns=true

设置配置配RpcConfig

rpc_client` 中的配置类`RpcConfig

关键点：

1.消息发送调用 sendAndReceive 方法，该方法自带返回值，返回值就是服务端返回的消息。
2.服务端返回的消息中，头信息中包含了 spring_returned_message_correlation 字段，这个就是消息发送时候的 correlation_id，通过消息发送时候correlation_id 以及返回消息头中的 spring_returned_message_correlation 字段值，我们就可以将返回的消息内容和发送的消息绑定到一起，确认出这个返回的内容就是针对这个发送的消息的。
3.客户端的开发最核心的就是sendAndReceive方法的调用。主要点就是发送的消息中要有correlation_id即我们的application配置类中要有其配置。

/**
 * client 发送消息，发到 RPC_MSG_QUEUE 队列，server 监听这个队列，就可以收到 client 发来的消息了，然后 server 对消息进行处理，处理完成后，将响应的内容发到 RPC_REPLY_QUEUE 队列上，client 监听 RPC_REPLY_QUEUE 队列，就可以收到 server 返回的信息了
 */
@Configuration
public class RpcConfig {
    //client 发送消息的时候，发到这个队列上（server 将来监听这个队列）
    public static final String RPC_MSG_QUEUE = "rpc_msg_queue";
    //server 响应消息的时候，发送到这个队列上（client 将来监听这个队列）
    public static final String RPC_REPLY_QUEUE = "rpc_reply_queue";
    public static final String RPC_EXCHANGE_NAME = "rpc_exchange_name";

    @Bean
    Queue rpcMsgQueue() {
        return new Queue(RPC_MSG_QUEUE, true, false, false);
    }

    @Bean
    Queue rpcReplyQueue() {
        return new Queue(RPC_REPLY_QUEUE, true, false, false);
    }

    @Bean
    TopicExchange topicExchange() {
        return new TopicExchange(RPC_EXCHANGE_NAME, true, false);
    }

    @Bean
    Binding rpcReplyBinding() {
        return BindingBuilder.bind(rpcReplyQueue())
                .to(topicExchange())
                .with(RPC_REPLY_QUEUE);
    }

    @Bean
    Binding rpcMsgBinding() {
        return BindingBuilder.bind(rpcMsgQueue())
                .to(topicExchange())
                .with(RPC_MSG_QUEUE);
    }

    @Bean
    RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        //设置响应队列名
        rabbitTemplate.setReplyAddress(RPC_REPLY_QUEUE);
        //设置响应超时时间
        rabbitTemplate.setReplyTimeout(6000);
        return rabbitTemplate;
    }

    /**
     * 当 server 响应数据的时候，数据是直接发送到队列中的，按照旧的方法就应该写一个消费者，但是这样会将消息的发送和接受分裂开。
     *
     * 所以，我们配置这个监听器，就可以实现消息的发送和接受在一起
     * @param connectionFactory
     * @return
     */
    @Bean
    SimpleMessageListenerContainer simpleMessageListenerContainer(ConnectionFactory connectionFactory) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        container.setQueueNames(RPC_REPLY_QUEUE);
        container.setMessageListener(rabbitTemplate(connectionFactory));
        return container;
    }
}

rpc_server这里的关键是配置一下消息队列并且把消息队列和消息交换机绑定起来。

/**
 * client 发送消息，发到 RPC_MSG_QUEUE 队列，server 监听这个队列，就可以收到 client 发来的消息了，然后 server 对消息进行处理，处理完成后，将响应的内容发到 RPC_REPLY_QUEUE 队列上，client 监听 RPC_REPLY_QUEUE 队列，就可以收到 server 返回的信息了
 */
@Configuration
public class RpcConfig {

    //client 发送消息的时候，发到这个队列上（server 将来监听这个队列）
    public static final String RPC_MSG_QUEUE = "rpc_msg_queue";
    //server 响应消息的时候，发送到这个队列上（client 将来监听这个队列）
    public static final String RPC_REPLY_QUEUE = "rpc_reply_queue";
    public static final String RPC_EXCHANGE_NAME = "rpc_exchange_name";

    @Bean
    Queue rpcMsgQueue() {
        return new Queue(RPC_MSG_QUEUE, true, false, false);
    }

    @Bean
    Queue rpcReplyQueue() {
        return new Queue(RPC_REPLY_QUEUE, true, false, false);
    }

    @Bean
    TopicExchange topicExchange() {
        return new TopicExchange(RPC_EXCHANGE_NAME, true, false);
    }

    @Bean
    Binding rpcReplyBinding() {
        return BindingBuilder.bind(rpcReplyQueue())
                .to(topicExchange())
                .with(RPC_REPLY_QUEUE);
    }

    @Bean
    Binding rpcMsgBinding() {
        return BindingBuilder.bind(rpcMsgQueue())
                .to(topicExchange())
                .with(RPC_MSG_QUEUE);
    }

配置Controller

rpc_client 中的RpcClientController

@RestController
public class RpcClientController {

    private static final Logger logger = LoggerFactory.getLogger(RpcClientController.class);

    @Autowired
    RabbitTemplate rabbitTemplate;

    @GetMapping("/send")
    public void sendMsg(String msg) {
        Message message = MessageBuilder.withBody(msg.getBytes()).build();
        //这里拿到的响应值，就是服务端返回的消息
        Message result = rabbitTemplate.sendAndReceive(RpcConfig.RPC_EXCHANGE_NAME, RpcConfig.RPC_MSG_QUEUE, message);
        if (result != null) {
            //现在判断一下这个响应的数据，是不是刚刚发送的请求的响应
            //这个是刚刚发送的消息自动携带的 id
            String correlationId = message.getMessageProperties().getCorrelationId();
            String result_correlationId = (String) result.getMessageProperties().getHeaders().get("spring_returned_message_correlation");
            if (correlationId.equals(result_correlationId)) {
                //说明响应的这个消息就是刚刚发送消息的响应
                byte[] body = result.getBody();
                logger.info("服务端返回数据：{}", new String(body, 0, body.length));
            }
        }
    }

rpc_server中的RpcServerController

这里的逻辑

服务端首先收到消息并打印出来。
服务端提取出原消息中的 correlation_id。
服务端利用sendAndReceive方法，将消息发送给RPC_EXCHANGE_NAME和RPC_REPLY_QUEUE队列，同时带上correlationId参数
服务端的消息发出后，客户端会受到服务端返回的结果

@RestController
public class RpcServerController {

    private static final Logger logger = LoggerFactory.getLogger(RpcServerController.class);

    @Autowired
    RabbitTemplate rabbitTemplate;

    @RabbitListener(queues = RpcConfig.RPC_MSG_QUEUE)
    public void handleMsg(Message message) {
        byte[] body = message.getBody();
        //打印出来信息
        logger.info("客户端发送来消息:{}", new String(body, 0, body.length));
        //获取客户端发送来的消息的 correlationId
        String correlationId = message.getMessageProperties().getCorrelationId();
        Message msg = MessageBuilder.withBody("收到消息".getBytes()).build();
        //服务端利用sendAndReceive方法，将消息发送给RPC_EXCHANGE_NAME和RPC_REPLY_QUEUE队列，同时带上correlationId参数
        rabbitTemplate.sendAndReceive(RpcConfig.RPC_EXCHANGE_NAME,RpcConfig.RPC_REPLY_QUEUE,msg,new CorrelationData(correlationId));
    }
}

Publish/Subscribe的四种交换机

Direct直连交换机

DirectExchange 的路由策略是将消息队列绑定到一个DirectExchange 上，当一条消息到达DirectExchange 时会被转发到与该条消息routing key相同的Queue上。

例如消息队列名为“hello-queue”，则routingkey为“hello-queue”的消息会被该消息队列接收。

具体步骤

发送的时候给消息设置一个东西叫做routingkey。
生产者把消息发到交换机上面的时候，交换机会找一个和routingkey同名的队列把这条消息给他转发出去。
DirectConfig类

步骤:

定义列和交换机名字
列和交换机注入bean容器
绑定方法注入bean容器

@Configuration
public class DirectConfig {
    //定义两个队列
    public  static final String MY_QUEUE_NAME_02 ="my queue name 02";
    public  static final String MY_QUEUE_NAME_03 ="my queue name 03";
    //交换机的名字
    public  static final String DIRECT_EXCHANGE_NAME="direct exchange name";

    @Bean
    Queue queue02(){
        return new Queue(MY_QUEUE_NAME_02,true,false,false);
    }

    @Bean
    Queue queue03(){
        return  new Queue(MY_QUEUE_NAME_03,true,false,false);
    }

    @Bean
    DirectExchange directExchange(){
        /**
         * 1.交换机的名字
         * 2. 交换机是否持久化
         * 3. 没有队列的时候，交换机是否自动化删除
         */
        return  new DirectExchange(DIRECT_EXCHANGE_NAME,true ,false);
    }

    @Bean
    Binding bingdingQueue02(){
        //将 queue02 和交换机绑定起来
        return BindingBuilder.bind(queue02())
                .to(directExchange())
        //为这个 Binding 设置一个 routingKey，将来消息发送的时候，会携带一个 routingKey，这个消息达到交换机之后，会自动将这个消息路由到与之同名
                .with(MY_QUEUE_NAME_02);
    }

    @Bean
    Binding bingdingQueue03(){
        //将 queue02 和交换机绑定起来
        return BindingBuilder.bind(queue03())
                .to(directExchange())
                .with(MY_QUEUE_NAME_03);
    }
}

DirectConsumer

监听两个消息队列

@Component
public class DirectConsumer {
    private static final Logger logger = LoggerFactory.getLogger(DirectConsumer.class);

    @RabbitListener(queues = DirectConfig.MY_QUEUE_NAME_02)
    public void handleMsg(String msg){
        logger.info("{}:{}",DirectConfig.MY_QUEUE_NAME_02,msg);
    }

    @RabbitListener(queues = DirectConfig.MY_QUEUE_NAME_03)
    public  void handleMsg2(String msg){
        logger.info("{}:{}",DirectConfig.MY_QUEUE_NAME_03,msg);
    }
}

生产者依旧使用第三小节中的producer01在producer01的测试类中加入方法：让其同时给customer02中的queue02和queue03发送数据如下：

 /**
 * 第一个参数：交换机
 * 第二个参数： routingKey ---》为了方便一般都是路径的名字
 * 第三个参数：生产者要发送的数据
 */
@Test
void test01() {
    rabbitTemplate.convertAndSend(DirectConfig.DIRECT_EXCHANGE_NAME, DirectConfig.MY_QUEUE_NAME_02, "hello queue02");
    rabbitTemplate.convertAndSend(DirectConfig.DIRECT_EXCHANGE_NAME, DirectConfig.MY_QUEUE_NAME_03, "hello queue03");
}

测试结果为：queue02和queue03同时收到了消息，且消息不同，说明两个消费者消费的是不同的队列。今天的不同于昨天的是，今天是每个消费者消费自己的队列，且可控。

Fanout扇形交换机

FanoutExchange 的数据交换策略是把所有到达 FanoutExchange 的消息转发给所有与它绑定的 Queue 上，在这种策略中，routingkey 将不起任何作用。

fanout类型交换机会将接收到的消息广播给所有与之绑定的队列，也就是和路由键没有很大关系了`消息到达交换机之后，会把消息自动转发给与扇形交换机绑定的所有队列。![](https://static.iocoder.cn/200662888309d3df010306896ead0118.jpg)`FanoutConfig步骤与Direct直连交换机基本相同

定义所有队列和扇形交换机的名字
把扇形交换机和队列注入Bean中
绑定方法注入bean

@Configuration
public class FanoutConfig {

    //1.定义所有队列和扇形交换机的名字
    public  static final  String MY_FANOUT_QUEUE_NAME_01 ="my fanout queue name 01";
    public  static final  String MY_FANOUT_QUEUE_NAME_02 ="my fanout queue name 02";
    public  static final  String MY_FANOUT_EXCHANGE_NAME ="my_fanout_exchange_name";

    //2.把扇形交换机和队列注入Bean中
    @Bean
    Queue fanoutQueue01(){
        return new Queue(MY_FANOUT_QUEUE_NAME_01,true,false,false);
    }

    @Bean
    Queue fanoutQueue02(){
        return new Queue(MY_FANOUT_QUEUE_NAME_02,true,false,false);
    }

    @Bean
    FanoutExchange fanoutExchange(){
        return new FanoutExchange(MY_FANOUT_EXCHANGE_NAME, true, false);
    }

    //3.绑定方法注入bean
    @Bean
    Binding fanoutBind01(){
        return BindingBuilder.bind(fanoutQueue01())
                .to(fanoutExchange());
    }

    @Bean
    Binding fanoutBind02(){
        return BindingBuilder.bind(fanoutQueue02())
                .to(fanoutExchange());
    }

}

FanoutConsumer也基本相同

@Component
public class FanoutConsumer {

    private static final Logger logger = LoggerFactory.getLogger(FanoutConsumer.class);

    @RabbitListener(queues = FanoutConfig.MY_FANOUT_QUEUE_NAME_01)
    public void handleMsg(String msg) {
        logger.info("{}:{}", FanoutConfig.MY_FANOUT_QUEUE_NAME_01, msg);
    }

    @RabbitListener(queues = FanoutConfig.MY_FANOUT_QUEUE_NAME_02)
    public void handleMsg2(String msg) {
        logger.info("{}:{}", FanoutConfig.MY_FANOUT_QUEUE_NAME_02, msg);
    }
}

生产者的测试类

不同点:

fanout类型交换机会将接收到的消息广播给所有与之绑定的队列，也就是和路由键没有很大关系了,故此在这种策略中，routingkey将不起任何作用可以随意的填写。
@Test
    void test02() {
        rabbitTemplate.convertAndSend(FanoutConfig.MY_FANOUT_EXCHANGE_NAME, null, "hello fanout!");
    }

结果：

Topic主机交换机（用的很多）

TopicExchange 是比较复杂但是也比较灵活的一种路由策略，在 TopicExchange 中，Queue 通过 routingkey 绑定到 TopicExchange 上，当消息到达 TopicExchange 后，TopicExchange 根据消息的 routingkey 将消息路由到一个或者多个 Queue 上。

一个生产者，一个交换机，两个队列，两个消费者

#####

配置类 TopicConfig

每一个交换机都有一个主题，消息队列可以订阅不同的主题。
表示通配符，将来发送消息的时候，如果消息的 routingKey 是 xiaomi.xxx，那么这个消息就会被转发到 xiaomiNewsQueue 队列上。

#####

/**
 * 主题交换机
 * 每一个交换机都有一个主题，消息队列可以订阅不同的主题
 */
@Configuration
public class TopicConfig {
    public static final String XIAOMI_NEWS_QUEUE_NAME = "xiaomi_news_queue_name_01";
    public static final String HUAWEI_NEWS_QUEUE_NAME = "huawei_news_queue_name_01";
    public static final String PHONE_NEWS_QUEUE_NAME = "phone_news_queue_name_01";
    public static final String TOPIC_EXCHANGE_NAME = "topic_exchange_name";

    @Bean
    Queue xiaomiNewsQueue() {
        return new Queue(XIAOMI_NEWS_QUEUE_NAME, true, false, false);
    }

    @Bean
    Queue huaweiNewsQueue() {
        return new Queue(HUAWEI_NEWS_QUEUE_NAME, true, false, false);
    }

    @Bean
    Queue phoneNewsQueue() {
        return new Queue(PHONE_NEWS_QUEUE_NAME, true, false, false);
    }

    @Bean
    TopicExchange topicExchange() {
        return new TopicExchange(TOPIC_EXCHANGE_NAME, true, false);
    }

    @Bean
    Binding xiaomiBinding() {
        return BindingBuilder.bind(xiaomiNewsQueue())
                .to(topicExchange())
                //# 表示通配符，将来发送消息的时候，如果消息的 routingKey 是 xiaomi.xxx，那么这个消息就会被转发到 xiaomiNewsQueue 队列上
                .with("xiaomi.#");
    }

    @Bean
    Binding huaweiBinding() {
        return BindingBuilder.bind(huaweiNewsQueue())
                .to(topicExchange())
                .with("huawei.#");
    }

    @Bean
    Binding phoneBinding() {
        return BindingBuilder.bind(phoneNewsQueue())
                .to(topicExchange())
                .with("#.phone.#");
    }
}

实体类TopicConsumer

@Component
public class TopicConsumer {
    private static final Logger logger = LoggerFactory.getLogger(TopicConsumer.class);

    @RabbitListener(queues = TopicConfig.XIAOMI_NEWS_QUEUE_NAME)
    public void handleMsg(String msg) {
        logger.info("{}:{}", TopicConfig.XIAOMI_NEWS_QUEUE_NAME, msg);
    }

    @RabbitListener(queues = TopicConfig.HUAWEI_NEWS_QUEUE_NAME)
    public void handleMsg2(String msg) {
        logger.info("{}:{}", TopicConfig.HUAWEI_NEWS_QUEUE_NAME, msg);
    }

    @RabbitListener(queues = TopicConfig.PHONE_NEWS_QUEUE_NAME)
    public void handleMsg3(String msg) {
        logger.info("{}:{}", TopicConfig.PHONE_NEWS_QUEUE_NAME, msg);
    }
}

生产者的测试类

注意：

第一个参数还是交换机的名字
第二个参数是你的routingkey
第三个参数是你要传入的信息

@Test
  void test03() {
      //将被转发到xaiomi 的队列中
      rabbitTemplate.convertAndSend(TopicConfig.TOPIC_EXCHANGE_NAME, "xiaomi.news", "小米新闻");
      //将被转发到 huawei 的队列中
      rabbitTemplate.convertAndSend(TopicConfig.TOPIC_EXCHANGE_NAME, "huawei.news", "华为新闻");
      //将被转发到 xiaomi+phone 两个队列中
      rabbitTemplate.convertAndSend(TopicConfig.TOPIC_EXCHANGE_NAME, "xiaomi.phone", "小米手机新闻");

  }

测试结果

HeadersExchange 是一种使用较少的路由策略，HeadersExchange 会根据消息的 Header 将消息路由到不同的 Queue 上，这种策略也和 routingkey无关。

配置类HeaderConfig

@Configuration
public class HeaderConfig {

    public static final String HEADER_QUEUE_NAME01 = "header_queue_name01";
    public static final String HEADER_QUEUE_NAME02 = "header_queue_name02";
    public static final String HEADER_EXCHANGE_NAME = "header_exchange_name";

    @Bean
    Queue headerQueue01() {
        return new Queue(HEADER_QUEUE_NAME01, true, false, false);
    }

    @Bean
    Queue headerQueue02() {
        return new Queue(HEADER_QUEUE_NAME02, true, false, false);
    }

    @Bean
    HeadersExchange headersExchange() {
        return new HeadersExchange(HEADER_EXCHANGE_NAME, true, false);
    }

    @Bean
    Binding headerBinding01() {
        return BindingBuilder.bind(headerQueue01())
                .to(headersExchange())
                //如果将来的消息，消息头中有name 字段，就会进入到 headerQueue01 队列中
                .where("name").exists();
    }

    @Bean
    Binding headerBinding02() {
        return BindingBuilder.bind(headerQueue02())
                .to(headersExchange())
                //如果将来的消息，消息头中有age 字段且值为99，就会进入到 headerQueue02 队列中
                .where("age").matches(99);
    }

}

HeadConsumer

@Configuration
public class HeadConsumer {
    private static final Logger logger = LoggerFactory.getLogger(HeadConsumer.class);

    @RabbitListener(queues = HeaderConfig.HEADER_QUEUE_NAME01)
    public void handleMsg(byte[] msg) {
        logger.info("{}:{}", HeaderConfig.HEADER_QUEUE_NAME01, new  String(msg,0,msg.length));
    }

    @RabbitListener(queues = HeaderConfig.HEADER_QUEUE_NAME02)
    public void handleMsg2(byte[] msg) {
        logger.info("{}:{}", HeaderConfig.HEADER_QUEUE_NAME02, new  String(msg,0,msg.length));
    }      
}

生产者测试类

@Test
void test04() {
    //构建消息，并设置消息头
    Message msg01 = MessageBuilder.withBody("hello name".getBytes()).setHeader("name", "zhangsan").build();
    Message msg02 = MessageBuilder.withBody("hello age".getBytes()).setHeader("age", 99).build();
    rabbitTemplate.send(HeaderConfig.HEADER_EXCHANGE_NAME,null, msg01);
    rabbitTemplate.send(HeaderConfig.HEADER_EXCHANGE_NAME,null, msg02);
}

测试结果

芋道源码 —— 纯源码解析博客

硬卷消息中间件系列（六）：RabbitMQ 消息收发与交换机详解

七种消息收发方式

Hello World

work

1.4 Routing

Topics

RPC

消息中间件怎么实现解耦的功能？

简单架构图

RPC 调用的小例子

关键点：

这里的逻辑

Publish/Subscribe的四种交换机

Direct直连交换机

具体步骤

Fanout扇形交换机

生产者的测试类

Topic主机交换机（用的很多）

配置类 TopicConfig

实体类TopicConsumer

生产者的测试类

Header头部交换机（用的少）

配置类HeaderConfig

HeadConsumer

生产者测试类