RabbitMQ--SpringBoot--整合/使用/用法/实例/示例/实战

拼搏现实的明天。 2022-09-12 15:57 947阅读 0赞

原文网址：[RabbitMQ--SpringBoot--整合/使用/用法/实例/示例/实战\_IT利刃出鞘的博客-CSDN博客][RabbitMQ--SpringBoot--_IT_-CSDN]

# 其他网址 #

> [Spring AMQP 2.1.2.RELEASE 中文文档 - 1. Preface | Docs4dev][Spring AMQP 2.1.2.RELEASE _ - 1. Preface _ Docs4dev]
> 
> [RabbitMQ：@RabbitListener 与 @RabbitHandler 及 消息序列化 - 简书][RabbitMQ_RabbitListener _ _RabbitHandler _ _ -]
> 
> [SpringBoot+RabbitMq的几种姿势][SpringBoot_RabbitMq]  
> [RabbitMQ与SpringBoot整合实战 - 简书][RabbitMQ_SpringBoot_ -]

# 注解 #

**其他网址**

> [RabbitMQ：@RabbitListener 与 @RabbitHandler 及 消息序列化 - 简书][RabbitMQ_RabbitListener _ _RabbitHandler _ _ -]

**@RabbitListener**

> **用在方法上**
> 
> 当监听到队列中有消息时则会进行接收并处理，如果不存在，会报错。
> 
>     @Component
>     public class Receiver {
>         //也可监听多个队列：@RabbitListener(queues = {"hello", "hi"})
>         @RabbitListener(queues = "hello")
>         public void process(String hello) {
>             System.out.println ("Receiver : "  + hello);
>         }
>     }
> 
> @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 processMessage1(Message message) {
>         System.out.println(message);
>     }
> 
> **用在类上**
> 
> 1.  需配合 @RabbitHandler 注解一起使用
> 2.  @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));
>         }
>     }

**@Payload 与 @Headers**

> **其他网址**
> 
> [Spring AMQP 2.1.2.RELEASE 中文文档 - 3. Reference | Docs4dev][Spring AMQP 2.1.2.RELEASE _ - 3. Reference _ Docs4dev]  
> [RabbitMQ(三) RabbitMQ高级整合应用 - 盲目的拾荒者 - 博客园][RabbitMQ_ RabbitMQ_ - _ -]

**简介**

> @Headers 必须通过Map接收。  
> //@Header("amqp\_receivedRoutingKey") String rk 直接获取header中某一个key

> 获得消息中的 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);
> }

# 绑定 #

**其他网址**

> [SpringBoot 整合 rabbitmq][SpringBoot _ rabbitmq]

**简介**

> **1.创建交换机/队列/绑定 实例都有两种方式**
> 
> **交换机（下边两种方式等价）：**
> 
>     ExchangeBuilder.topicExchange(EXCHANGE_TOPIC_WELCOME).durable(true).build();
>     new TopicExchange(EXCHANGE_TOPIC_WELCOME, true, false)
> 
> **队列（下边两种方式等价）：**
> 
>     QueueBuilder.durable("Hi").build();
>     new Queue(QUEUE_HI, true)
> 
> **绑定（下边两种方式等价）：**
> 
> 注意：第一种的参数并不是字符串。
> 
>     BindingBuilder.bind(helloQueue).to(welcomExchange).with("hello.#")
>     new Binding("Queue@hello", Binding.DestinationType.QUEUE,
>                     "Exchange@topic.welcome", "hello.#", null)

推荐程度依次递减。

**法1：通过配置类配置（简洁）**

> package com.example.config;
> 
> import org.springframework.amqp.core.*;
> import org.springframework.beans.factory.annotation.Autowired;
> import org.springframework.context.annotation.Bean;
> import org.springframework.context.annotation.Configuration;
> 
> @Configuration
> public class RabbitRouterConfig {
> public static final String QUEUE_HELLO = "Queue@hello";
> public static final String QUEUE_HI = "Queue@hi";
> public static final String EXCHANGE_TOPIC_WELCOME = "Exchange@topic.welcome";
> public static final String ROUTINGKEY_HELLOS = "hello.#";
> 
> @Autowired
> AmqpAdmin amqpAdmin;
> 
> @Bean
> Object initBindingTest() {
> amqpAdmin.declareExchange(new TopicExchange(EXCHANGE_TOPIC_WELCOME, true, false));
> 
> amqpAdmin.declareQueue(new Queue(QUEUE_HI, true));
> amqpAdmin.declareQueue(new Queue(QUEUE_HELLO, true));
> 
> amqpAdmin.declareBinding(new Binding(QUEUE_HELLO, Binding.DestinationType.QUEUE,
> EXCHANGE_TOPIC_WELCOME, ROUTINGKEY_HELLOS, null));
> 
> return new Object();
> }
> }
> 
> **amqpAdmin.declareBinding**
> 
> 需要一个Binding对象作为参数
> 
> * exchange：交换器名称
> * type：交换器类型。BuiltinExchangeType枚举类，有以下4中类型交换器：DIRECT(“direct”), FANOUT(“fanout”), TOPIC(“topic”), HEADERS(“headers”)
> * durable：设置是否持久化。true：持久化，false：非持久化。持久化可以将交换器存盘，在服务器重启时不会丢失相关消息。
> * autoDelete：设置是否自动删除。true：自动删除，false：不自动删除。自动删除的前提是至少有一个队列或交换器与这个交换器绑定，之后所有与这个交换器绑定的队列或交换器都与此交换器解绑。
> * internal：设置是否内置的。true：内置交换器，false：非内置交换器。内置交换器，客户端无法直接发送消息到这个交换器中，只能通过交换器路由到交换器这种方式。
> * arguments：其他一些结构化参数。如备份交换器：alternate-exchange、超时时间。示例配置超时时间方法：
> 
> Map<String, Object> params = new HashMap();
> params.put("x-message-ttl", 2000);
> amqpAdmin.declareBinding(new Binding(QUEUE_HELLO, Binding.DestinationType.QUEUE,
> EXCHANGE_TOPIC_WELCOME, ROUTINGKEY_HELLOS, params));

**法2：通过配置类配置（繁琐）**

> **适用于队列和交换器不多时**
> 
>     package com.lly.order.message;
>     
>     import org.springframework.amqp.core.*;
>     import org.springframework.context.annotation.Bean;
>     import org.springframework.context.annotation.Configuration;
>     
>     @Configuration
>     public class RabbitMqConfig {
>         public final static String QUEUE_DIRECT     = "Queue@direct";
>         public final static String QUEUE_TOPIC_ONE  = "Queue@topic_one";
>         public final static String TOPIC_QUEUE_TWO  = "Queue@topic_two";
>         public final static String QUEUE_FANOUT_ONE = "Queue@fanout_one";
>         public final static String QUEUE_FANOUT_TWO = "Queue@fanout_two";
>     
>         public final static String EXCHANGE_TOPIC   = "Exchange@topic";
>         public final static String EXCHANGE_FANOUT  = "Exchange@fanout";
>     
>         public final static String ROUTINGKEY_TOPIC_ONE = "hello.key";
>         public final static String ROUTINGKEY_TOPIC_TWO = "*.key";
>     
>     	//  direct模式队列
>         @Bean
>         public Queue directQueue() {
>             return new Queue(QUEUE_DIRECT, true);
>         }
>     	
>     	//  topic 订阅者模式队列
>         @Bean
>         public Queue topicQueueOne() {
>             return new Queue(QUEUE_TOPIC_ONE, true);
>         }
>         @Bean
>         public Queue topicQueueTwo() {
>             return new Queue(TOPIC_QUEUE_TWO, true);
>         }
>     	
>     	//  fanout 广播者模式队列
>         @Bean
>         public Queue fanoutQueueOne() {
>             return new Queue(QUEUE_FANOUT_ONE, true);
>         }
>         @Bean
>         public Queue fanoutQueueTwo() {
>             return new Queue(QUEUE_FANOUT_TWO, true);
>         }
>     	
>     	//  topic 交换器
>         @Bean
>         public TopicExchange topExchange() {
>             return new TopicExchange(EXCHANGE_TOPIC);
>         }
>     	
>     	//  fanout 交换器
>         @Bean
>         public FanoutExchange fanoutExchange() {
>             return new FanoutExchange(EXCHANGE_FANOUT);
>         }
>     
>     	//   订阅者模式绑定
>         @Bean
>         public Binding topicExchangeBingingOne() {
>             return BindingBuilder.bind(topicQueueOne()).to(topExchange()).with(ROUTINGKEY_TOPIC_ONE);
>         }
>     
>         @Bean
>         public Binding topicExchangeBingingTwo() {
>             return BindingBuilder.bind(topicQueueTwo()).to(topicExchange()).with(ROUTINGKEY_TOPIC_TWO);
>         }
>     	
>     	//   广播模式绑定
>         @Bean
>         public Binding fanoutExchangeBingingOne() {
>             return BindingBuilder.bind(fanoutQueueOne()).to(fanoutExchange());
>         }
>         @Bean
>         public Binding fanoutExchangeBingingTwo() {
>             return BindingBuilder.bind(fanoutQueueTwo()).to(fanoutExchange());
>         }
>     }

**法3：通过代码配置**

>     @Component
>     public class Receiver {
>         @RabbitListener(queues = "hello")
>         public void process(String hello) {
>             System.out.println ("Receiver : "  + hello);
>         }
>     
>     	@RabbitListener(bindings = @QueueBinding(
>     			exchange = @Exchange(value = "Exchange@topic.Hello",durable = "true",type = "topic"),
>     			value    = @Queue(value = "Queue@Hello",durable = "true"),
>     			key      = "key.#"
>     	))
>     	public void processMessage1(Message message) {
>     		System.out.println(message);
>     	}
>     }

**法4：通过MQ网页配置**

> 添加交换器：[http://localhost:15672/\#/exchanges][http_localhost_15672_exchanges] //例如：Exchange@topic.Hello  
> 添加队列：[http://localhost:15672/\#/queues][http_localhost_15672_queues] //例如：Queue@Hello  
> 交换器添加路由键：[http://localhost:15672/\#/exchanges][http_localhost_15672_exchanges]=> 点击交换器名字=> Binding=> 添加队列与路由

> ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlaXlpbmcwY2FuZ2xhbmc_size_16_color_FFFFFF_t_70][]
> 
> ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlaXlpbmcwY2FuZ2xhbmc_size_16_color_FFFFFF_t_70 1][]

# 配置 #

**RabbitTemplate配置**

> <table> 
> <tbody> 
> <tr> 
> <td> 实例 </td> 
> <td> 作用 </td> 
> </tr> 
> <tr> 
> <td> rabbitTemplate.setConfirmCallback(rabbitConfirmCallback); </td> 
> <td> 设置confirm回调函数。 </td> 
> </tr> 
> <tr> 
> <td> rabbitTemplate.setReturnCallback(rabbitReturnCallback); </td> 
> <td> 设置return回调函数。 </td> 
> </tr> 
> <tr> 
> <td> rabbitTemplate.setMandatory(true); </td> 
> <td> 当mandatory设置为true时，若exchange根据自身类型和消息routingKey无法找到一个合适的queue存储消息，那么broker会调用basic.return方法将消息返还给生产者。当mandatory设置为false时，出现上述情况broker会直接将消息丢弃。 </td> 
> </tr> 
> <tr> 
> <td> rabbitTemplate.setUsePublisherConnection(true);&nbsp; </td> 
> <td> 使用单独的发送连接，避免生产者由于各种原因阻塞而导致消费者同样阻塞 </td> 
> </tr> 
> </tbody> 
> </table>

# 实例 #

## 操作RabbitMQ ##

**启动RabbitMQ**

> 命令行运行：
> 
> rabbitmq-server.bat
> 
> //若是第一次开启，还要运行：rabbitmq-plugins.bat enable rabbitmq\_management

**配置一个管理员用户**

> 用户名：admin 密码：123456
> 
> 权限配置为：/
> 
> **详见**：[RabbitMQ系列--综述\_feiying0canglang的博客-CSDN博客][RabbitMQ_--_feiying0canglang_-CSDN]

## 代码 ##

**依赖**

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

**application.yml**

>     server:
>     #  port: 9100
>       port: 9101
>     spring:
>       application:
>     #    name: demo-rabbitmq-sender
>         name: demo-rabbitmq-receiver
>       rabbitmq:
>         host: localhost
>         port: 5672
>         username: admin
>         password: 123456
>     #    virtualHost: /

**队列配置**

**Controller**

>     package com.example.controller;
>     
>     import com.example.config.RabbitConfig;
>     import com.example.mq.Sender;
>     import org.springframework.beans.factory.annotation.Autowired;
>     import org.springframework.web.bind.annotation.PostMapping;
>     import org.springframework.web.bind.annotation.RestController;
>     
>     import java.time.LocalDateTime;
>     
>     @RestController
>     public class HelloController {
>         @Autowired
>         private Sender sender;
>     
>         @PostMapping("/hi")
>         public void hi() {
>             sender.send(RabbitConfig.QUEUE_HI, "hi1 message:" + LocalDateTime.now());
>         }
>     
>         @PostMapping("/hello1")
>         public void hello1() {
>             sender.send("hello.a", "hello1 message:" + LocalDateTime.now());
>         }
>     
>         @PostMapping("/hello2")
>         public void hello2() {
>             sender.send("Exchange@topic.welcome","hello.b", "hello2 message:" + LocalDateTime.now());
>         }
>     }

**sender**

>     package com.example.mq;
>     
>     import org.springframework.amqp.core.AmqpTemplate;
>     import org.springframework.beans.factory.annotation.Autowired;
>     import org.springframework.stereotype.Component;
>     
>     import java.util.Date;
>     
>     @Component
>     public class Sender {
>         @Autowired
>         private AmqpTemplate rabbitTemplate;
>     
>         public void send(String routingKey, String message) {
>             this.rabbitTemplate.convertAndSend(routingKey, message);
>         }
>     
>         public void send(String exchange, String routingKey, String message) {
>             this.rabbitTemplate.convertAndSend(exchange, routingKey, message);
>         }
>     }

**receiver**

>     package com.example.mq;
>     
>     import com.example.config.RabbitRouterConfig;
>     import com.rabbitmq.client.Channel;
>     import org.springframework.amqp.core.Message;
>     import org.springframework.amqp.rabbit.annotation.RabbitListener;
>     import org.springframework.messaging.handler.annotation.Payload;
>     import org.springframework.stereotype.Component;
>     
>     @Component
>     public class Receiver {
>         @RabbitListener(queues = RabbitRouterConfig.QUEUE_HELLO)
>         public void hello(String hello) {
>             System.out.println ("Receiver(hello) : "  + hello);
>         }
>     
>         @RabbitListener(queues = RabbitRouterConfig.QUEUE_HI)
>         public void hi(String payload) {
>             System.out.println ("Receiver(hi) : "  + payload);
>             throw new RuntimeException("手动造的异常");
>         }
>     
>     //    @RabbitListener(queues = RabbitRouterConfig.QUEUE_HELLO)
>     //    public void helloAll(@Payload String payload, Message message, Channel channel) {
>     //        System.out.println("Receiver(hello)：");
>     //        System.out.println("payload:" + payload);
>     //        System.out.println("message:" + message);
>     //        System.out.println("channel:" + channel);
>     //    }
>     //    @RabbitListener(queues = RabbitRouterConfig.QUEUE_HI)
>     //    public void hiAll(@Payload String payload, Message message, Channel channel) {
>     //        System.out.println("Receiver(hi)：");
>     //        System.out.println("payload:" + payload);
>     //        System.out.println("message:" + message);
>     //        System.out.println("channel:" + channel);
>     //    }
>     }

## **基础测试** ##

**运行sender**

>     server:
>       port: 9100
>     #  port: 9101
>     spring:
>       application:
>         name: demo-rabbitmq-sender
>     #    name: demo-rabbitmq-receiver
>       rabbitmq:
>         host: localhost
>         port: 5672
>         username: admin
>         password: 123456
>     #    virtualHost: /

**运行receiver**

**MQ页面**

> ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlaXlpbmcwY2FuZ2xhbmc_size_16_color_FFFFFF_t_70 2][]
> 
> ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlaXlpbmcwY2FuZ2xhbmc_size_16_color_FFFFFF_t_70 3][]
> 
> ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlaXlpbmcwY2FuZ2xhbmc_size_16_color_FFFFFF_t_70 4][]

**测试只获得数据（payload）**

> **postman访问：**[Node Exporter][]
> 
> receiver打印结果
> 
>     Receiver(hi) : hi message:2020-10-21T18:52:24.766

> **postman访问：**[Node Exporter][Node Exporter 1]
> 
> receiver打印结果：无打印
> 
> 即：若队列已经与交换器绑定，则必须指定交换器和路由键。

> **postman访问：**[Node Exporter][Node Exporter 2]
> 
> receiver打印结果
> 
>     Receiver(hello) : hello2 message:2020-10-21T18:52:41.434

## 测试详细数据 ##

**测试获得payload/message/channel**

> 使用receiver的第二种代码。

> **postman访问：**[Node Exporter][]
> 
> receiver打印结果
> 
>     Receiver(hi)：
>     payload:hi1 message:2020-10-22T15:41:11.796
>     message:(Body:'hi1 message:2020-10-22T15:41:11.796' MessageProperties [headers={}, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=, receivedRoutingKey=Queue@hi, deliveryTag=1, consumerTag=amq.ctag-K0Yka5vHxrq6JzNXlg3ncQ, consumerQueue=Queue@hi])
>     channel:Cached Rabbit Channel: AMQChannel(amqp://admin@127.0.0.1:5672/,1), conn: Proxy@4f4c88f9 Shared Rabbit Connection: SimpleConnection@7c52fc81 [delegate=amqp://admin@127.0.0.1:5672/, localPort= 62067]

> **postman访问：**[Node Exporter][Node Exporter 1]
> 
> receiver打印结果：无打印
> 
> 即：若队列已经与交换器绑定，则必须指定交换器和路由键。

> **postman访问：**[Node Exporter][Node Exporter 2]
> 
> receiver打印结果
> 
>     Receiver(hello)：
>     payload:hello2 message:2020-10-22T15:42:13.126
>     message:(Body:'hello2 message:2020-10-22T15:42:13.126' MessageProperties [headers={}, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=Exchange@topic.welcome, receivedRoutingKey=hello.b, deliveryTag=1, consumerTag=amq.ctag-yZryIR801zNV5f0f9mJd9g, consumerQueue=Queue@hello])
>     channel:Cached Rabbit Channel: AMQChannel(amqp://admin@127.0.0.1:5672/,2), conn: Proxy@4f4c88f9 Shared Rabbit Connection: SimpleConnection@7c52fc81 [delegate=amqp://admin@127.0.0.1:5672/, localPort= 62067]

[RabbitMQ--SpringBoot--_IT_-CSDN]: https://knife.blog.csdn.net/article/details/120495800
[Spring AMQP 2.1.2.RELEASE _ - 1. Preface _ Docs4dev]: https://www.docs4dev.com/docs/zh/spring-amqp/2.1.2.RELEASE/reference/_preface.html
[RabbitMQ_RabbitListener _ _RabbitHandler _ _ -]: https://www.jianshu.com/p/911d987b5f11
[SpringBoot_RabbitMq]: https://juejin.im/post/6844903843612852232
[RabbitMQ_SpringBoot_ -]: https://www.jianshu.com/p/dca28ce0b3ec
[Spring AMQP 2.1.2.RELEASE _ - 3. Reference _ Docs4dev]: https://www.docs4dev.com/docs/zh/spring-amqp/2.1.2.RELEASE/reference/_reference.html
[RabbitMQ_ RabbitMQ_ - _ -]: https://www.cnblogs.com/niugang0920/p/13043708.html
[SpringBoot _ rabbitmq]: https://juejin.im/post/6844903784624160775#heading-13
[http_localhost_15672_exchanges]: http://localhost:15672/#/exchanges
[http_localhost_15672_queues]: http://localhost:15672/#/queues
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlaXlpbmcwY2FuZ2xhbmc_size_16_color_FFFFFF_t_70]: /images/20220828/442b539e313e445780349cca76560ae9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlaXlpbmcwY2FuZ2xhbmc_size_16_color_FFFFFF_t_70 1]: /images/20220828/658336ace96040158eebd40091fd01e5.png
[RabbitMQ_--_feiying0canglang_-CSDN]: https://blog.csdn.net/feiying0canglang/article/details/106932556
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlaXlpbmcwY2FuZ2xhbmc_size_16_color_FFFFFF_t_70 2]: /images/20220828/7f9bfa673f4e442fb07014ce16ece628.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlaXlpbmcwY2FuZ2xhbmc_size_16_color_FFFFFF_t_70 3]: /images/20220828/eb7273c3cb5449918236f98c756cbd22.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZlaXlpbmcwY2FuZ2xhbmc_size_16_color_FFFFFF_t_70 4]: /images/20220828/9321271995154946a2cda2a204389902.png
[Node Exporter]: http://localhost:9100/hi
[Node Exporter 1]: http://localhost:9100/hello1
[Node Exporter 2]: http://localhost:9100/hello2