整合篇：零基础学习与使用RabbitMQ

冷不防 2023-01-05 14:15 210阅读 0赞

### 目录 ###

*  1、RabbitMQ介绍
 *  2、RabbitMQ安装
 *   *  2.1、安装Erlang
     *  2.2、安装RabbitMQ
 *  3、RabbitMQ集成
 *   *  3.1、单播模式
     *   *  3.1.1、消息生产者
         *  3.1.2、消息消费者
     *  3.2、主题模式
     *   *  3.2.1、消息生产者
         *  3.2.2、消息消费者
     *  3.3、广播模式
     *   *  3.3.1、消息生产者
         *  3.3.2、消息消费者
     *  3.4、头部模式
     *   *  3.4.1、消息生产者
         *  3.4.2、消息消费者
 *  4、RabbitMQ注解
 *   *  4.1、@RabbitListener 用法
     *  4.2、@RabbitHandler 用法
     *  4.3、@Payload 与 @Headers 用法

--------------------

**配套资料，免费下载**  
链接：https://pan.baidu.com/s/1jA217UgqXpONi\_fV-aOzqw  
提取码：bm2g  
复制这段内容后打开百度网盘手机App，操作更方便哦

# 1、RabbitMQ介绍 #

RabbitMQ

RabbitMQ是使用Erlang语言来编写的并且RabbitMQ是基于AMQP协议的。Erlang语言在数据交互方面性能优秀，有着和原生Socket一样的延迟，这也是RabbitMQ高性能的原因所在。可谓“人如其名”，RabbitMQ像兔子一样迅速。RabbitMQ除了像兔子一样跑的很快以外，还有这些特点：

*  开源、性能优秀，稳定性保障
 *  提供可靠性消息投递模式、返回模式
 *  与Spring AMQP完美整合，API丰富
 *  集群模式丰富，表达式配置，HA模式，镜像队列模型
 *  保证数据不丢失的前提做到高可靠性、可用性

AMQP协议

提到RabbitMQ，就不得不提AMQP协议。AMQP协议是具有现代特征的二进制协议。是一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。先了解一下AMQP协议中间的几个重要概念：

*  Server：接收客户端的连接，实现AMQP实体服务。
 *  Connection：连接，应用程序与Server的网络连接，TCP连接。
 *  Channel：信道，消息读写等操作在信道中进行。客户端可以建立多个信道，每个信道代表一个会话任务。
 *  Message：消息，应用程序和服务器之间传送的数据，消息可以非常简单，也可以很复杂。有Properties和Body组成。Properties为外包装，可以对消息进行修饰，比如消息的优先级、延迟等高级特性；Body就是消息体内容。
 *  Virtual Host：虚拟主机，用于逻辑隔离。一个虚拟主机里面可以有若干个Exchange和Queue，同一个虚拟主机里面不能有相同名称的Exchange或Queue。
 *  Exchange：交换器，接收消息，按照路由规则将消息路由到一个或者多个队列。如果路由不到，或者返回给生产者，或者直接丢弃。RabbitMQ常用的交换器常用类型有direct、topic、fanout、headers四种，后面详细介绍。
 *  Binding：绑定，交换器和消息队列之间的虚拟连接，绑定中可以包含一个或者多个RoutingKey。
 *  RoutingKey：路由键，生产者将消息发送给交换器的时候，会发送一个RoutingKey，用来指定路由规则，这样交换器就知道把消息发送到哪个队列。路由键通常为一个“.”分割的字符串，例如“com.rabbitmq”。
 *  Queue：消息队列，用来保存消息，供消费者消费。

我们完全可以直接使用 Connection 就能完成信道的工作，为什么还要引入信道呢?

试想这样一个场景， 一个应用程序中有很多个线程需要从 RabbitMQ 中消费消息，或者生产消息，那么必然需要建立很多个 Connection，也就是许多个 TCP 连接。然而对于操作系统而言，建立和销毁 TCP 连接是非常昂贵的开销，如果遇到使用高峰，性能瓶颈也随之显现。 RabbitMQ 采用 TCP 连接复用的方式，不仅可以减少性能开销，同时也便于管理 。

下图是AMQP的协议模型：

![img][]

正如图中所看到的，AMQP协议模型有三部分组成：生产者、消费者和服务端。

生产者是投递消息的一方，首先连接到Server，建立一个连接，开启一个信道；然后生产者声明交换器和队列，设置相关属性，并通过路由键将交换器和队列进行绑定。同理，消费者也需要进行建立连接，开启信道等操作，便于接收消息。接着生产者就可以发送消息，发送到服务端中的虚拟主机，虚拟主机中的交换器根据路由键选择路由规则，然后发送到不同的消息队列中，这样订阅了消息队列的消费者就可以获取到消息，进行消费。最后还要关闭信道和连接。

RabbitMQ是基于AMQP协议实现的，其结构如下图所示，和AMQP协议简直就是一模一样。

![img][img 1]

常用交换器

RabbitMQ常用的交换器类型有direct、topic、fanout、headers四种。

**Direct Exchange**

该类型的交换器将所有发送到该交换器的消息被转发到RoutingKey指定的队列中，也就是说路由到BindingKey和RoutingKey完全匹配的队列中。

![img][img 2]

**Topic Exchange**

该类型的交换器将所有发送到Topic Exchange的消息被转发到所有RoutingKey中指定的Topic的队列上面。Exchange将RoutingKey和某Topic进行模糊匹配，其中“\*”用来匹配一个词，“\#”用于匹配一个或者多个词。例如“com.\#”能匹配到“com.rabbitmq.oa”和“com.rabbitmq”；而"login.\*"只能匹配到“com.rabbitmq”。

![img][img 3]

**Fanout Exchange**

该类型不处理路由键，会把所有发送到交换器的消息路由到所有绑定的队列中。优点是转发消息最快，性能最好。

![img][img 4]

**Headers Exchange**

该类型的交换器不依赖路由规则来路由消息，而是根据消息内容中的headers属性进行匹配。headers类型交换器性能差，在实际中并不常用。

# 2、RabbitMQ安装 #

## 2.1、安装Erlang ##

下载地址：https://www.erlang.org/downloads

> 温馨提示：如果下载不下来，配套资料中已经帮你都下载好了

![image-20210108150617494][]

安装步骤：打开安装包，默认下一步即可。

环境变量：新建ERLANG\_HOME，变量值为安装目录

![image-20210108150919351][]

修改 Path 环境变量，新增一条 `%ERLANG_HOME%\bin` ，如下所示：

![image-20210108151054179][]

测试是否安装成功，请重新打开一个 cmd 命令行窗口，输入 `erl`

![image-20210108151650390][]

## 2.2、安装RabbitMQ ##

下载地址的首页：https://www.rabbitmq.com/download.html

安装版下载地址：https://www.rabbitmq.com/install-windows.html

解压版下载地址：https://www.rabbitmq.com/install-windows-manual.html

将 [rabbitmq-server-windows-3.8.9.zip][] 解压到 `C:\DevTools\`

![image-20210108152136278][]

环境变量：新建RABBITMQ\_SERVER，变量值为解压后目录

![image-20210108152310740][]

修改 Path 环境变量，新增一条 `%RABBITMQ_SERVER%\sbin` ，如下所示：

![image-20210108152429914][]

安装RabbitMQ控制台，请到 `C:\DevTools\rabbitmq_server-3.8.9\sbin` 目录下，输入命令 `rabbitmq-plugins enable rabbitmq_management`

![image-20210108152926193][]

启动RabbitMQ服务，输入命令：`rabbitmq-server.bat`

![image-20210108153056493][]

启动成功后，如下图所示：

![image-20210108153256234][]

Rabbitmq启动成功，浏览器中[http://localhost:15672][http_localhost_15672]，登录账户：guest，登录密码：guest

![image-20210108153403011][]

![image-20210108153422541][]

# 3、RabbitMQ集成 #

## 3.1、单播模式 ##

DirectExchange是RabbitMQ的默认交换机，直接使用routingKey匹配队列，是一种精准匹配。

### 3.1.1、消息生产者 ###

![image-20210108155100624][]

![image-20210108155124888][]

![image-20210108155153322][]

![image-20210108155213895][]

编写配置信息：application.yaml

spring:
      rabbitmq:
        host: localhost
        port: 5672
        username: guest
        password: guest

编写配置类：com.caochenlei.integrationrabbitmqproducer.config.DirectConfig

*  配置一个q.direct的消息队列并绑定在e.direct交换机上。（交换机的匹配规则为rk.direct）

@Configuration
    public class DirectConfig { 
        @Bean
        public Queue directQueue() { 
            //队列名字，是否持久化
            return new Queue("q.direct", false);
        }
    
        @Bean
        public DirectExchange directExchange() { 
            //交换器名称、是否持久化、是否自动删除
            return new DirectExchange("e.direct", false, false);
        }
    
        @Bean
        public Binding directBinding(Queue directQueue, DirectExchange directExchange) { 
            //将指定的交换器绑定到指定的队列，路由键是rk.direct
            return BindingBuilder.bind(directQueue).to(directExchange).with("rk.direct");
        }
    }

编写生产者：com.caochenlei.integrationrabbitmqproducer.bean.DirectProducer

*  添加一个send()方法，发送消息至e.direct交换机且routingKey为rk.direct

@Component
    public class DirectProducer { 
        @Autowired
        private AmqpTemplate amqpTemplate;
    
        public void send(String message) { 
            amqpTemplate.convertAndSend("e.direct", "rk.direct", message);
        }
    }

测试生产者：com.caochenlei.integrationrabbitmqproducer.IntegrationRabbitmqProducerApplicationTests

@SpringBootTest
    class IntegrationRabbitmqProducerApplicationTests { 
        @Autowired
        private DirectProducer directProducer;
    
        @Test
        void testDirectSend() { 
            directProducer.send("张三");
        }
    }

查看队列：http://localhost:15672/\#/queues

![image-20210108210554905][]

查看交换机：http://localhost:15672/\#/exchanges

![image-20210108210623414][]

查看绑定：http://localhost:15672/\#/exchanges/%2F/e.direct

![image-20210108210659840][]

### 3.1.2、消息消费者 ###

![image-20210108163258502][]

![image-20210108163329596][]

![image-20210108163407144][]

![image-20210108163429065][]

编写配置信息：application.yaml

spring:
      rabbitmq:
        host: localhost
        port: 5672
        username: guest
        password: guest

编写消费者：com.caochenlei.integrationrabbitmqconsumer.bean.DirectConsumer

*  添加receive()方法，监听队列名称为q.direct的队列消息

@Component
    public class DirectConsumer { 
        @RabbitListener(queues = "q.direct")
        public void receive(String message) { 
            System.out.println("接收消息：" + message);
        }
    }

重新启动这个工程，然后进行观察网页队列和控制台情况：http://localhost:15672/\#/queues

![image-20210108210838389][]

![image-20210108210807545][]

## 3.2、主题模式 ##

TopicExchange是按规则转发消息，是交换机中最灵活的一个，也是最常用的一个。

### 3.2.1、消息生产者 ###

com.caochenlei.integrationrabbitmqproducer.config.TopicConfig

*  配置一个api.core的消息队列并绑定在coreExchange交换机上。（交换机的匹配规则为api.core.\*）
 *  配置一个api.payment的消息队列并绑定在paymentExchange交换机上。（交换机的匹配规则为api.payment.\#）

@Configuration
    public class TopicConfig { 
        @Bean
        public Queue coreQueue() { 
            return new Queue("api.core");
        }
    
        @Bean
        public Queue paymentQueue() { 
            return new Queue("api.payment");
        }
    
        @Bean
        public TopicExchange coreExchange() { 
            return new TopicExchange("coreExchange");
        }
    
        @Bean
        public TopicExchange paymentExchange() { 
            return new TopicExchange("paymentExchange");
        }
    
        @Bean
        public Binding bindingCoreExchange(Queue coreQueue, TopicExchange coreExchange) { 
            return BindingBuilder.bind(coreQueue).to(coreExchange).with("api.core.*");
        }
    
        @Bean
        public Binding bindingPaymentExchange(Queue paymentQueue, TopicExchange paymentExchange) { 
            return BindingBuilder.bind(paymentQueue).to(paymentExchange).with("api.payment.#");
        }
    }

com.caochenlei.integrationrabbitmqproducer.bean.TopicProducer

*  添加一个user()方法，发送消息至coreExchange交换机且routingKey为api.core.user
 *  添加一个userQuery()方法，发送消息至coreExchange交换机且routingKey为api.core.user.query
 *  添加一个order()方法，发送消息至paymentExchange交换机且routingKey为api.payment.order
 *  添加一个orderQuery()方法，发送消息至paymentExchange交换机且routingKey为api.payment.order.query

@Component
    public class TopicProducer { 
        @Autowired
        private AmqpTemplate amqpTemplate;
    
        public void user(String message) { 
            amqpTemplate.convertAndSend("coreExchange", "api.core.user", message);
        }
    
        public void userQuery(String message) { 
            amqpTemplate.convertAndSend("coreExchange", "api.core.user.query", message);
        }
    
        public void order(String message) { 
            amqpTemplate.convertAndSend("paymentExchange", "api.payment.order", message);
        }
    
        public void orderQuery(String message) { 
            amqpTemplate.convertAndSend("paymentExchange", "api.payment.order.query", message);
        }
    }

com.caochenlei.integrationrabbitmqproducer.IntegrationRabbitmqProducerApplicationTests

@SpringBootTest
    class IntegrationRabbitmqProducerApplicationTests { 
        @Autowired
        private TopicProducer topicProducer;
    
        @Test
        void testTopicUserSend() { 
            topicProducer.user("用户编号：987654321");
        }
    
        @Test
        void testTopicUserQuerySend() { 
            topicProducer.userQuery("查询用户：987654321");
        }
    
        @Test
        void testTopicOrderSend() { 
            topicProducer.order("订单编号：123456789");
        }
    
        @Test
        void testTopicOrderQuerySend() { 
            topicProducer.orderQuery("查询订单：123456789");
        }
    }

测试testTopicUserSend()，测试完毕，打开队列查看当前队列情况：http://localhost:15672/\#/queues

![image-20210109105753205][]

测试testTopicUserQuerySend()，测试完毕，打开队列查看当前队列情况：http://localhost:15672/\#/queues

> 注意：因为在TopicConfig配置类中，我们对api.core队列绑定的交换机规则是api.core.\*，而通配符“\*”只能向后多匹配一层路径。

![image-20210109105836026][]

测试testTopicOrderSend()，测试完毕，打开队列查看当前队列情况：http://localhost:15672/\#/queues

![image-20210109110110942][]

测试testTopicOrderQuerySend()，测试完毕，打开队列查看当前队列情况：http://localhost:15672/\#/queues

![image-20210109110148585][]

### 3.2.2、消息消费者 ###

com.caochenlei.integrationrabbitmqconsumer.bean.TopicConsumer

*  添加user()方法，监听队列名称为api.core的队列消息
 *  添加order()方法，监听队列名称为api.payment的队列消息

@Component
    public class TopicConsumer { 
        @RabbitListener(queues = "api.core")
        public void user(String message) { 
            System.out.println(message);
        }
    
        @RabbitListener(queues = "api.payment")
        public void order(String message) { 
            System.out.println(message);
        }
    }

重新启动这个工程，然后进行观察网页队列和控制台情况：http://localhost:15672/\#/queues

![image-20210109110640315][]

![image-20210109110712891][]

## 3.3、广播模式 ##

FanoutExchange交换机是转发消息到所有绑定队列（广播模式，和routingKey没有关系）。

### 3.3.1、消息生产者 ###

com.caochenlei.integrationrabbitmqproducer.config.FanoutConfig

*  配置一个api.report.payment的消息队列并绑定在reportExchange交换机上。
 *  配置一个api.report.refund的消息队列并绑定在reportExchange交换机上。

@Configuration
    public class FanoutConfig { 
        @Bean
        public Queue reportPaymentQueue() { 
            return new Queue("api.report.payment");
        }
    
        @Bean
        public Queue reportRefundQueue() { 
            return new Queue("api.report.refund");
        }
    
        @Bean
        public FanoutExchange reportExchange() { 
            return new FanoutExchange("reportExchange");
        }
    
        @Bean
        public Binding bindingReportPaymentExchange(Queue reportPaymentQueue, FanoutExchange reportExchange) { 
            return BindingBuilder.bind(reportPaymentQueue).to(reportExchange);
        }
    
        @Bean
        public Binding bindingReportRefundExchange(Queue reportRefundQueue, FanoutExchange reportExchange) { 
            return BindingBuilder.bind(reportRefundQueue).to(reportExchange);
        }
    }

com.caochenlei.integrationrabbitmqproducer.bean.TopicProducer

*  添加一个generateReports()方法，发送消息至reportExchange交换机

@Component
    public class FanoutProducer { 
        @Autowired
        private AmqpTemplate amqpTemplate;
    
        public void generateReports(String message) { 
            amqpTemplate.convertAndSend("reportExchange", "api.report", message);
        }
    }

com.caochenlei.integrationrabbitmqproducer.IntegrationRabbitmqProducerApplicationTests\#fanoutProducer

@SpringBootTest
    class IntegrationRabbitmqProducerApplicationTests { 
        @Autowired
        private FanoutProducer fanoutProducer;
    
        @Test
        void testGenerateReportsSend() { 
            fanoutProducer.generateReports("开始生成报表！");
        }
    }

测试testGenerateReportsSend()，测试完毕，打开队列查看当前队列情况：http://localhost:15672/\#/queues

![image-20210109113716679][]

### 3.3.2、消息消费者 ###

com.caochenlei.integrationrabbitmqconsumer.bean.FanoutConsumer

*  添加payment()方法，监听队列名称为api.report.payment的队列消息
 *  添加refund()方法，监听队列名称为api.report.refund的队列消息

@Component
    public class FanoutConsumer { 
        @RabbitListener(queues = "api.report.payment")
        public void payment(String message) { 
            System.out.println(message);
        }
    
        @RabbitListener(queues = "api.report.refund")
        public void refund(String message) { 
            System.out.println(message);
        }
    }

重新启动这个工程，然后进行观察网页队列和控制台情况：http://localhost:15672/\#/queues

![image-20210109113949037][]

![image-20210109114012779][]

## 3.4、头部模式 ##

HeadersExchange交换机是根据请求消息中设置的header attribute参数类型来匹配的（和routingKey没有关系）。

### 3.4.1、消息生产者 ###

com.caochenlei.integrationrabbitmqproducer.config.HeadersConfig

*  配置一个credit.bank的消息队列并绑定在creditBankExchange交换机上。

@Configuration
    public class HeadersConfig { 
        @Bean
        public Queue creditBankQueue() { 
            return new Queue("credit.bank");
        }
    
        @Bean
        public HeadersExchange creditBankExchange() { 
            return new HeadersExchange("creditBankExchange");
        }
    
        @Bean
        public Binding bindingCreditBankExchange(Queue creditBankQueue, HeadersExchange creditBankExchange) { 
            Map<String, Object> headerValues = new HashMap<>();
            headerValues.put("type", "cash");
            headerValues.put("aging", "fast");
            return BindingBuilder.bind(creditBankQueue).to(creditBankExchange).whereAll(headerValues).match();
        }
    }

com.caochenlei.integrationrabbitmqproducer.bean.HeadersProducer

*  添加一个creditBank()方法，发送消息至creditBankExchange交换机且routingKey为credit.bank

@Component
    public class HeadersProducer { 
        @Autowired
        private AmqpTemplate amqpTemplate;
    
        private Message getMessage(Map<String, Object> head, String message) { 
            //创建信息属性对象
            MessageProperties messageProperties = new MessageProperties();
            //循环添加头部信息
            Set<Map.Entry<String, Object>> headerEntries = head.entrySet();
            for (Map.Entry<String, Object> header : headerEntries) { 
                String key = header.getKey();
                Object value = header.getValue();
                messageProperties.setHeader(key, value);
            }
            //返回指定头部信息对象
            return new Message(message.getBytes(), messageProperties);
        }
    
        public void creditBank(Map<String, Object> head, String message) { 
            amqpTemplate.convertAndSend("creditBankExchange", "credit.bank", getMessage(head, message));
        }
    }

com.caochenlei.integrationrabbitmqproducer.IntegrationRabbitmqProducerApplicationTests

@SpringBootTest
    class IntegrationRabbitmqProducerApplicationTests { 
        @Autowired
        private HeadersProducer headersProducer;
    
        @Test
        public void testCreditBankSend() { 
            Map<String, Object> head = new HashMap<>();
            head.put("type", "cash");
            headersProducer.creditBank(head, "银行授信(部分匹配)");
        }
    
        @Test
        public void testCreditBankAllSend() { 
            Map<String, Object> head = new HashMap<>();
            head.put("type", "cash");
            head.put("aging", "fast");
            headersProducer.creditBank(head, "银行授信(全部匹配)");
        }
    }

测试testCreditBankSend()，测试完毕，打开队列查看当前队列情况：http://localhost:15672/\#/queues

> 注意：部分匹配是无法将消息发送到指定队列中的。

![image-20210109124919564][]

测试testCreditBankAllSend()，测试完毕，打开队列查看当前队列情况：http://localhost:15672/\#/queues

![image-20210109125005752][]

### 3.4.2、消息消费者 ###

com.caochenlei.integrationrabbitmqconsumer.bean.HeadersConsumer

*  添加creditBank()方法，监听队列名称为credit.bank的队列消息

@Component
    public class HeadersConsumer { 
        @RabbitListener(queues = "credit.bank")
        public void creditBank(String message) { 
            System.out.println(message);
        }
    }

重新启动这个工程，然后进行观察网页队列和控制台情况：http://localhost:15672/\#/queues

![image-20210109125419637][]

![image-20210109125438897][]

# 4、RabbitMQ注解 #

## 4.1、@RabbitListener 用法 ##

使用 @RabbitListener 注解标记方法，当监听到队列 debug 中有消息时则会进行接收并处理。

@RabbitListener(queues = "debug")
    public void processMessage(Message bytes) { 
        System.out.println(new String(bytes));
    }

消息处理方法参数是由 MessageConverter 转化，默认 Spring 使用的实现是 SimpleRabbitListenerContainerFactory，SimpleMessageConverter 对于要发送的消息体 body 为 byte\[\] 时不进行处理，如果是 String 则转成字节数组，如果是 Java 对象，则使用 Jdk 序列化将消息转成字节数组，转出来的结果较大，含class类名，类相应方法等信息，因此性能较差。消息的 content\_type 属性表示消息 body 数据以什么数据格式存储，接收消息除了使用 Message 对象接收消息（包含消息属性等信息）之外，还可直接使用对应类型接收消息 body 内容，但若方法参数类型不正确会抛异常：

*  application/octet-stream：二进制字节数组存储，使用 byte\[\]

![image-20210109133337550][]

*  text/plain：文本数据类型存储，使用 String

![image-20210109133715050][]

*  application/x-java-serialized-object：java 对象序列化格式存储，使用 Object、相应类型（反序列化时类型应该同包同名，否者会抛出找不到类异常）

![image-20210109134157683][]

*  application/json：JSON 格式，使用 Object、相应类型，需要自己配置 Jackson2JsonMessageConverter

![image-20210109134541750][]

当使用 RabbitMQ 作为中间件时，数据量比较大，此时就要考虑使用 Jackson2JsonMessageConverter 等序列化形式以此提高性能，序列化与反序列化配置：

import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
    import org.springframework.amqp.support.converter.MessageConverter;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    @Configuration
    public class AMQPConfig { 
        @Bean
        public MessageConverter messageConverter(){ 
            return new Jackson2JsonMessageConverter();
        }
    }

生产者和消费者都需要配置上边这一段配置，使用了之后，所有序列化和反序列化都将使用 Jackson2JsonMessageConverter ，如果启动报错，请加入依赖：

<dependency>
        <groupId>com.fasterxml.jackson.core</groupId>
        <artifactId>jackson-databind</artifactId>
    </dependency>

通过 @RabbitListener 的 bindings 属性声明 Binding（若 RabbitMQ 中不存在该绑定所需要的 Queue、Exchange、RouteKey 则自动创建，若存在则抛出异常）

@RabbitListener(bindings = @QueueBinding(
            exchange = @Exchange(value = "topic.exchange",durable = "true",type = "topic"),
            value = @Queue(value = "consumer_queue",durable = "true"),
            key = "key.#"
    ))
    public void processMessage(Message message) { 
        System.out.println(message);
    }

## 4.2、@RabbitHandler 用法 ##

@RabbitListener 可以标注在类上面，需配合 @RabbitHandler 注解一起使用，@RabbitListener 标注在类上面表示当有收到消息的时候，就交给 @RabbitHandler 的方法处理，具体使用哪个方法处理，根据 MessageConverter 转换后的参数类型。

@Component
    @RabbitListener(queues = "consumer_queue")
    public class Receiver { 
        @RabbitHandler
        public void processMessage1(String message) { 
            System.out.println(message);
        }
    
        @RabbitHandler
        public void processMessage2(byte[] message) { 
            System.out.println(new String(message));
        }
    }

## 4.3、@Payload 与 @Headers 用法 ##

使用 @Payload 和 @Headers 注解可以消息中的 body 与 headers 信息。

@RabbitListener(queues = "debug")
    public void processMessage1(@Payload String body, @Headers Map<String,Object> headers) { 
        System.out.println("body："+body);
        System.out.println("Headers："+headers);
    }

也可以获取单个 Header 属性。

@RabbitListener(queues = "debug")
    public void processMessage1(@Payload String body, @Header String token) { 
        System.out.println("body："+body);
        System.out.println("token："+token);
    }

[img]: /images/20221119/7fc224f8654b474dbae4f2258b2842c7.png
[img 1]: /images/20221119/7ba0f815511948a5bb39d1494e7d032b.png
[img 2]: /images/20221119/38c85cf0072f468e8208440753e00521.png
[img 3]: /images/20221119/112b391b21ad41e08598697c702291d6.png
[img 4]: /images/20221119/50d7a4c1e2c44f80909ce559660c715f.png
[image-20210108150617494]: /images/20221119/21ed34a110c4490eb9a838115947329e.png
[image-20210108150919351]: https://img-blog.csdnimg.cn/img_convert/b8b2829cee5bceba4f8a2a471c12ccec.png
[image-20210108151054179]: https://img-blog.csdnimg.cn/img_convert/bb9edbbdc4494b86b889989d9bae6d2d.png
[image-20210108151650390]: https://img-blog.csdnimg.cn/img_convert/0e85be886474d6e17e788cc14c551f41.png
[rabbitmq-server-windows-3.8.9.zip]: https://github.com/rabbitmq/rabbitmq-server/releases/download/v3.8.9/rabbitmq-server-windows-3.8.9.zip
[image-20210108152136278]: https://img-blog.csdnimg.cn/img_convert/5db15989428e9d34cb73ce61e35d4836.png
[image-20210108152310740]: https://img-blog.csdnimg.cn/img_convert/b512a70c35aed0073066d9985eb84b74.png
[image-20210108152429914]: https://img-blog.csdnimg.cn/img_convert/8cc56db8e4604b2e4b1aec51461d7801.png
[image-20210108152926193]: https://img-blog.csdnimg.cn/img_convert/c441211695b9e962f2d1a34d8fd1f60b.png
[image-20210108153056493]: https://img-blog.csdnimg.cn/img_convert/d5bb1b4332de10715b0d36bb5a654b50.png
[image-20210108153256234]: https://img-blog.csdnimg.cn/img_convert/9695bc13f5d7b32841c205b47af8af8d.png
[http_localhost_15672]: http://localhost:15672/
[image-20210108153403011]: https://img-blog.csdnimg.cn/img_convert/2fb999c03e18fda48267ba513d892809.png
[image-20210108153422541]: https://img-blog.csdnimg.cn/img_convert/37d430bd1100aa44ab3545be335435e2.png
[image-20210108155100624]: https://img-blog.csdnimg.cn/img_convert/ef914db946fce64d7641e12490ce00c3.png
[image-20210108155124888]: https://img-blog.csdnimg.cn/img_convert/d46dc386126b6b187afc14fc3395eb8d.png
[image-20210108155153322]: https://img-blog.csdnimg.cn/img_convert/ee5c3e15d852094edd86d45b0ece7d97.png
[image-20210108155213895]: https://img-blog.csdnimg.cn/img_convert/f77e1d37d4bd3eff74d5ecd93caaddb7.png
[image-20210108210554905]: https://img-blog.csdnimg.cn/img_convert/f2495848f68e6a7ae1a4ff33717a0592.png
[image-20210108210623414]: https://img-blog.csdnimg.cn/img_convert/a5d8e58f9d163cf5399c58b757b10ebb.png
[image-20210108210659840]: https://img-blog.csdnimg.cn/img_convert/cf2da7d17aeb925921eaab60e127b5c9.png
[image-20210108163258502]: https://img-blog.csdnimg.cn/img_convert/a05d3e66a97821e02c7fa0a774cdd5c6.png
[image-20210108163329596]: https://img-blog.csdnimg.cn/img_convert/bcbc5ad5278193a6bf0ae40e837ed411.png
[image-20210108163407144]: https://img-blog.csdnimg.cn/img_convert/df1e974c3d3bf39c2e64c4331f6cc69c.png
[image-20210108163429065]: https://img-blog.csdnimg.cn/img_convert/8409eb00e15082d1cf3df8b455fc1283.png
[image-20210108210838389]: https://img-blog.csdnimg.cn/img_convert/35351022e6c0e6cd0a6713413284f08a.png
[image-20210108210807545]: https://img-blog.csdnimg.cn/img_convert/3b83aaf1e19f007c9af4667489c5743e.png
[image-20210109105753205]: https://img-blog.csdnimg.cn/img_convert/e0c13ec66710fc82a8ca63044c18ebba.png
[image-20210109105836026]: https://img-blog.csdnimg.cn/img_convert/568f74a9f6732c6063edc691c011f824.png
[image-20210109110110942]: https://img-blog.csdnimg.cn/img_convert/d244f8fa4d3de4201ac5c5cd8db753ac.png
[image-20210109110148585]: https://img-blog.csdnimg.cn/img_convert/eeef50fa8bd70d9b4798482529200ecd.png
[image-20210109110640315]: https://img-blog.csdnimg.cn/img_convert/89a373bbbd3fbf10ab207ef572da3f02.png
[image-20210109110712891]: https://img-blog.csdnimg.cn/img_convert/d9c2fff12ace279f214255273ef9cb9c.png
[image-20210109113716679]: https://img-blog.csdnimg.cn/img_convert/1e3be5dd07b5e471920297a3f2912bff.png
[image-20210109113949037]: https://img-blog.csdnimg.cn/img_convert/3738f653735be885b2275967870d848e.png
[image-20210109114012779]: https://img-blog.csdnimg.cn/img_convert/165761ac984986cc17a5c3bf631d5568.png
[image-20210109124919564]: https://img-blog.csdnimg.cn/img_convert/bf178100ea0f352c40bd53a26d827e6c.png
[image-20210109125005752]: https://img-blog.csdnimg.cn/img_convert/0bd252995b914b18607685fb128911c8.png
[image-20210109125419637]: https://img-blog.csdnimg.cn/img_convert/d3f5f719376fc4e43e49352cddcd62e0.png
[image-20210109125438897]: https://img-blog.csdnimg.cn/img_convert/5c644a66a74f0f39a9d70287435c9fe9.png
[image-20210109133337550]: https://img-blog.csdnimg.cn/img_convert/ad3956a3bd485763722c4c2b2ab46e5e.png
[image-20210109133715050]: https://img-blog.csdnimg.cn/img_convert/da14a52858f0aa061848743c77717436.png
[image-20210109134157683]: https://img-blog.csdnimg.cn/img_convert/28c844fea5a50df709af42d1333aed38.png
[image-20210109134541750]: https://img-blog.csdnimg.cn/img_convert/379b3632d05a4f76df3eb52d27471117.png