devbox
小丑西瓜9
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

27.JavaWeb-RabbitMQ消息队列_java 整合rabbitmq

作者：小丑西瓜9 | 2024-05-23 22:12:11

赞

踩

java 整合rabbitmq

目录

1.MQ介绍

1.1 应用场景

1.1.1 系统之间通过MQ进行消息通信，异步解耦

1.1.2 异步处理消息

1.1.3 肖锋限流

2.RabbitMQ

2.1 安装RabbitMQ

2.1.1 安装Erlang程序运行环境

2.1.2 安装RabbitMQ服务器

2.1.3 使用RabbitMQ控制台 

2.2 RabbitMQ队列的工作模式 

2.2.1 Simple简单模式

2.2.2 work工作模式

2.2.3 发布订阅模式

2.2.4 Routing路由模式

2.2.5 Topic主题模式

3.SpringBoot整合MQ

3.1 开发生产端

3.1.1 配置spring配置

3.1.2 导入依赖

3.1.3 配置队列、交换机并绑定

3.1.4 发送消息

3.2 开发消费端

3.2.1 处理消息

3.3 生产者端消息确认和回退

3.3.1 yml配置

3.3.2 定义消息确认回调的方法

3.3.2.1 配置类添加配置

3.3.2.2 修改发送消息的方法，携带附加数据

3.3.3 消息回退

3.3.3.1 定义处理消息回退的方法

4.消息消费可靠性保障

4.1 生产者丢失消息

4.2 RabbitMQ丢失消息

4.3 消费者丢失消息

4.3.1 修改配置

4.3.2 然后在消费端代码中手动应答签收消息

4.3.3 使用 SpringBoot 提供的消息重试

4.5 消息重复消费（消息幂等性）

4.5.1 确保消费端只执行一次

4.5.2 允许消费端执行多次，保证数据不受影响

4.5.2.1 数据库唯一约束

4.5.2.2 数据库乐观锁*

5. 消息转换器

5.1 定义一个对象

5.2 创建生产者发送消息方法

5.3 发送消息

5.4 Spring的消息转换器

5.4.1 生产端配置消息转换器

5.4.2 消费端配置消息转换器

6.延迟队列

6.1 死信队列

6.2 TTL消息

6.2.1 生产端配置类配置TTL消息

6.2.2 创建生产者发送消息方法

6.2.3 发送消息

6.3 死信交换机

6.3.1 生产端配置类配置死信交换机

6.3.2 接收端处理消息

1.MQ介绍

        Message Queue （MQ）是一种跨进程的通信机制，用于在系统之间进行传递消息。MQ作为消息中间件，可以进行异步处理请求，从而减少请求响应时间和解耦

1.1 应用场景

1.1.1 系统之间通过MQ进行消息通信，异步解耦

        多应用间通过消息队列对同一消息进行处理，避免调用接口失败导致整个过程失败。

1.1.2 异步处理消息

        应用间并发处理消息，相比串行处理，减少处理时间；

        比如用户注册，系统需要发送注册邮件，验证短信。MQ作为消息中间件，可以进行异步处理请求，从而减少请求响应时间。

1.1.3 肖锋限流

        广泛应用于高并发系统中，避免流量过大导致处理请求阻塞的情况

2.RabbitMQ

        RabbitMQ是支持多种消息协议，易于部署和使用的开源消息代理服务器，用于在分布式系统中存储转发消息。

由以高性能、健壮以及可伸缩性出名的Erlang语言编写；
提供了成熟的高并发，高可用的解决方案 可以根据实际业务情况动态地扩展集群节点。
在集群中的机器上设置镜像，使得在部分节点出现问题的情况下仍然可用。
支持多种客户端语言，如：Python、Ruby、.NET、Java等，支持AJAX。
RabbitMQ提供了一个易用的用户界面，使得用户可以监控和管理消息、集群中的节点等。
MQ产品语言支持并发量消息回溯支持性能协议可靠性
RocketMQJava, C++较高支持良好自定义TCP良好
KafkaJava非常高不支持卓越自定义TCP良好
ActiveMQJava万级不支持一般JMS规范较差

1、Producer：生产者，即消息的提供者

2、Consumer：消费者，即消息的使用者

3、Message：消息，即进程之间进行通信的数据

4、Queue：队列，即消息存放的容器，消息以先进先出的方式进行存储

5、Vhost：虚拟主机，用于存储消息队列

  • 虚拟主机vhost是AMQP概念的基础，RabbitMQ包含了开箱即用的默认虚拟主机"/“

  • vhost之间是绝对隔离，保障了队列和交换机的安全性。  

2.1 安装RabbitMQ

2.1.1 安装Erlang程序运行环境

Downloads - Erlang/OTPThe official home of the Erlang Programming Languageicon-default.png?t=N7T8http://www.erlang.org/downloads

2.1.2 安装RabbitMQ服务器

RabbitMQ: easy to use, flexible messaging and streaming — RabbitMQicon-default.png?t=N7T8http://www.rabbitmq.com/

2.1.3 使用RabbitMQ控制台 

        建议直接使用启动

        在RabbitMQ的文件路径打开cmd，启动RabbitMQ 

net start RabbitMQ

         关闭RabbitMQ 

net stop RabbitMQ

        注意：第一次使用RabbitMQ需要激活

  1. #进入sbin目录下打开cmd
  2. #启用管理控制台
  3. rabbitmq-plugins.bat enable rabbitmq_management

        通过浏览器进行访问

  1. #管理控制台的端口是15672
  2. http://localhost:15672/
  3. #默认提供的管理员用户
  4. 用户名：guest   密码：guest

2.1.4 更改RabbitMQ配置

        新版本的配置好像都叫 advanced.config，位置如下

C:\Users\你的用户名\AppData\Roaming\RabbitMQ

2.1.4.1 更改RabbitMQ端口
  1. [
  2.   {rabbit, [
  3.     {tcp_listeners, [{"127.0.0.1", 5672}]},
  4.     {loopback_users, []}
  5.   ]},
  6.   {rabbitmq_management, [
  7.     {listener, [
  8.       {port, 15672},
  9.       {ip, "0.0.0.0"}
  10.     ]}
  11.   ]}
  12. ].
  • 0.0.0.0的ip可以让外部连接
  • 这里的配置是rabbitMQ的web端外部可以访问，连接只能本地连接 

2.1.5 默认guest账号只能在本地登录

        建议新创建一个管理员账号在非本地登录

2.2 RabbitMQ队列的工作模式 

2.2.1 Simple简单模式

        一个生产者对应一个消费者

2.2.2 work工作模式

        一对多，每个消费者得到的消息唯一（work模式下最大发挥每个消费者的性能）

2.2.3 发布订阅模式

        一对多，生产者将消息发送给交换机，交换机再将消息转发给与之绑定的队列，发布订阅模式对应的交换机类型是FANOUT

2.2.4 Routing路由模式

        一对多，可以根据指定的路由key，将消息发送给不同的队列，交换机类型是DIRECT

2.2.5 Topic主题模式

        一对多，在Routing路由模式的基础上，可以使用通配符定义路由key

                * 匹配单个字符

                # 匹配任意个字符

3.SpringBoot整合MQ

3.1 开发生产端

3.1.1 配置spring配置

  1. spring:
  2.   rabbitmq:
  3.     virtual-host: /myhost
  4.     port: 5672
  5.     host: localhost
  6.     username: admin
  7.     password: admin

3.1.2 导入依赖

  1.         <dependency>
  2.             <groupId>org.springframework.boot</groupId>
  3.             <artifactId>spring-boot-starter-amqp</artifactId>
  4.         </dependency>
  5.         <dependency>
  6.             <groupId>org.springframework.boot</groupId>
  7.             <artifactId>spring-boot-starter-test</artifactId>
  8.             <scope>test</scope>
  9.         </dependency>
  10.         <dependency>
  11.             <groupId>org.springframework.amqp</groupId>
  12.             <artifactId>spring-rabbit-test</artifactId>
  13.             <scope>test</scope>
  14.         </dependency>

3.1.3 配置队列、交换机并绑定

  1. @Configuration
  2. public class MqConfig {
  3.     public static final String MSG_QUEUE = "msg-queue";
  4.     public static final String MSG_EXCHANGE = "msg-exchange";
  5.     public static final String MSG_ROUTE_KEY = "msg.key";
  6.     /**
  7.      * 声明队列
  8.      */
  9.     @Bean
  10.     public Queue msgQueue() {
  11.         return new Queue(MSG_QUEUE,true,false,false);
  12.     }
  13.     /**
  14.      * 声明交换机
  15.      */
  16.     @Bean
  17.     public TopicExchange msgExchage() {
  18.         //参数2是否持久化，参数3是否自动删除
  19.         return new TopicExchange(MSG_EXCHANGE,true,false);
  20.     }
  21.     /**
  22.      * 将队列和交换机进行绑定
  23.      */
  24.     @Bean
  25.     public Binding bindMsgQueue() {
  26.         return BindingBuilder.bind(msgQueue()).to(msgExchage()).with(MSG_ROUTE_KEY);
  27.     }
  28. }

3.1.4 发送消息

  1. @Component
  2. public class MsgProducer {
  3.     @Autowired
  4.     private RabbitTemplate rabbitTemplate;
  5.     public void sendMsg(String message) {
  6.         rabbitTemplate.convertAndSend(MqConfig.MSG_EXCHANGE,MqConfig.MSG_ROUTE_KEY,message);
  7.     }
  8. }

3.2 开发消费端

        同样导入依赖、配置yml

3.2.1 处理消息

  1. @Component
  2. public class MsgConsumer {
  3.     public static final String MSG_QUEUE = "msg-queue";
  4.     public static final String MSG_EXCHANGE = "msg-exchange";
  5.     //消息处理的方法
  6.     @RabbitHandler
  7.     @RabbitListener(bindings = @QueueBinding(
  8.             value = @Queue(name=MSG_QUEUE,declare = "true",durable = "true",exclusive = "false",autoDelete = "false"),
  9.             exchange = @Exchange(name = MSG_EXCHANGE,type = ExchangeTypes.TOPIC),
  10.             key = "msg.*"
  11.     ))
  12.     //@Payload表示消费者处理的消息
  13.     //@Headers注解表示接收的消息头将会被绑定到`headers`参数上
  14.     public void receiveMsg(@Payload String msg, Channel channel, @Headers Map headers) {
  15.         System.out.println("消费者处理消息：" + msg);
  16.         Long tag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
  17.         try {
  18.             channel.basicAck(tag,false);
  19.         } catch (IOException e) {
  20.             throw new RuntimeException(e);
  21.         }
  22.     }
  23. }

@RabbitHandler注解：表示该方法是一个消息处理方法

        bindings属性用于指定队列和交换机的绑定关系。  

@RabbitListener注解：表示该类是一个消息监听器，用于监听指定的队列

        value属性用于指定队列的属性，包括队列的名称、是否需要声明、是否持久化、是否排他、是否自动删除等

        exchange属性用于指定交换机的名称和类型

        key属性用于指定消息的路由键

3.3 生产者端消息确认和回退

3.3.1 yml配置

  1. spring:
  2.   rabbitmq:
  3.     virtual-host: /myhost
  4.     port: 5672
  5.     host: localhost
  6.     username: admin
  7.     password: admin
  8.     publisher-confirm-type: correlated
  9.     publisher-returns: true
ConfirmType
NONE禁用发布确认模式，是默认值。
CORRELATED将消息成功发布到交换器后触发回调方法。
SIMPLE与CORRELATED相似，也会在将消息成功发布到交换器后触发回调方法。

3.3.2 定义消息确认回调的方法

3.3.2.1 配置类添加配置
  1.     RabbitTemplate.ConfirmCallback callback = new RabbitTemplate.ConfirmCallback() {
  2.         @Override
  3.         public void confirm(CorrelationData correlationData, boolean isAck, String cause) {
  4.             if (!isAck) {
  5.                 System.out.println("拒收的原因：" + cause);
  6.             } else {
  7.                 if (correlationData != null) {
  8.                     System.out.println("broker接收消息自定义ID：" + correlationData.getId());
  9.                 }
  10.             }
  11.         }
  12.     };
  13.     @Bean
  14.     public RabbitTemplate rabbitTemplate(ConnectionFactory factory) {
  15.         RabbitTemplate template = new RabbitTemplate();
  16.         template.setConnectionFactory(factory);
  17.         template.setConfirmCallback(callback);
  18.         return template;
  19.     }
3.3.2.2 修改发送消息的方法，携带附加数据
  1. @Component
  2. public class MsgProducer {
  3.     @Autowired
  4.     private RabbitTemplate rabbitTemplate;
  5.     public void sendMsg(String message) {
  6.         //自定义的附加数据
  7.         CorrelationData data = new CorrelationData();
  8.         data.setId("10001");
  9.         rabbitTemplate.convertAndSend(MqConfig.MSG_EXCHANGE,MqConfig.MSG_ROUTE_KEY,message,data);
  10.     }
  11. }

3.3.3 消息回退

        在配置文件中添加 publisher-returns: true 配置消息回退

3.3.3.1 定义处理消息回退的方法
  1.     RabbitTemplate.ReturnsCallback returnsCallback = new RabbitTemplate.ReturnsCallback() {
  2.         @Override
  3.         public void returnedMessage(ReturnedMessage msg) {
  4.             System.out.println("--------消息路由失败------------");
  5.             System.out.println("消息主体：" + msg.getMessage());
  6.             System.out.println("返回编码：" + msg.getReplyCode());
  7.             System.out.println("描述信息：" + msg.getReplyText());
  8.             System.out.println("交换机：" + msg.getExchange());
  9.             System.out.println("路由key：" + msg.getExchange());
  10.             System.out.println("------------------------------");
  11.         }
  12.     };
  13.     @Bean
  14.     public RabbitTemplate rabbitTemplate(ConnectionFactory factory) {
  15.         RabbitTemplate template = new RabbitTemplate();
  16.         template.setConnectionFactory(factory);
  17.         template.setConfirmCallback(callback);
  18.         template.setReturnsCallback(returnsCallback);
  19.         // true 表示消息通过交换机无法路由到队列时候，会把消息返回给生产者
  20.         // false 消息无法路由到队列就直接丢弃
  21.         template.setMandatory(true);
  22.         return template;
  23.     }

4.消息消费可靠性保障

常见情况
生产者消息没到交换机生产者丢失消息
交换机没有把消息路由到队列生产者丢失消息
RabbitMQ 宕机导致队列、队列中的消息丢失RabbitMQ 丢失消息
消费者消费出现异常，业务没执行消费者丢失消息

4.1 生产者丢失消息

        解决办法为上面的异步监听confirm-type、publisher-returns，3.3.2.1的代码

4.2 RabbitMQ丢失消息

        设置持久化将消息写出磁盘，否则RabbitMQ重启后所有队列和消息都会丢失

4.3 消费者丢失消息

        消费者丢数据一般是因为采用了自动确认消息模式。MQ收到确认消息后会删除消息，如果这时消费者异常了，那消息就没了

        使用ack机制，默认情况下自动应答，可以使用手动ack

4.3.1 修改配置

  1.     listener:
  2.       simple:
  3.         acknowledge-mode: manual #开启消费者手动确认模式 （channel.bacisAck）

4.3.2 然后在消费端代码中手动应答签收消息

        如果消息消费失败，不执行消息确认代码，用channel的basicNack方法拒收

  1.     public void receiveMsg(@Payload String msg, Channel channel, @Headers Map headers) {
  2.         System.out.println("消费者处理消息：" + msg);
  3.         Long tag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
  4.         try {
  5.             System.out.println(2/0);
  6.             //其中`tag`是消息的唯一标识，`false`表示只确认当前消息，不确认之前的所有未确认消息。
  7.             channel.basicAck(tag,false);
  8.         } catch (Exception e) {
  9.             System.out.println("签收失败");
  10.             try {
  11.                 //其中`tag`是消息的唯一标识，`false`表示只拒绝当前消息，`true`表示该消息将重新进入队列等待被消费。
  12.                 channel.basicNack(tag,false,true);
  13.             } catch (IOException ex) {
  14.                 System.out.println("拒收失败");
  15.             }
  16.         }
  17.     }

        通常的代码报错并不能因为重试而解决，可能会造成死循环

解决办法：

        当消费失败后将此消息存到 Redis，记录消费次数，如果消费了三次还是失败，就丢弃掉消息，记录日志落库保存
basicNack方法的参数3直接填 false ，不重回队列，记录日志、发送邮件等待开发手动处理
不启用手动 ack ，使用 SpringBoot 提供的消息重试

4.3.3 使用 SpringBoot 提供的消息重试

  1.     listener:
  2.       simple:
  3.         retry:
  4.           enabled: true
  5.           max-attempts: 3 #重试次数

        注意：要抛异常，因为SpringBoot 触发重试是根据方法中发生未捕捉的异常来决定的

  1.     public void receiveMsg(@Payload String msg, Channel channel, @Headers Map headers) {
  2.         System.out.println("消费者处理消息：" + msg);
  3.         Long tag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
  4.         try {
  5.             System.out.println(2/0);
  6.             //channel.basicAck(tag,false);
  7.         } catch (Exception e) {
  8.             System.out.println("签收失败");
  9.              //记录日志、发送邮件、保存消息到数据库，落库之前判断如果消息已经落库就不保存
  10.             throw new RuntimeException(e);
  11.         }
  12.     }

4.5 消息重复消费（消息幂等性）

        使用手动恢复MQ解决了消息在消费者端丢失的问题，但是如果消费者处理消息成功后，由于网络波动导致手动回复MQ失败，该条消息还保存在消息队列中，由于MQ消息的重发机制，该消息会被重复消费，造成不好的后果

4.5.1 确保消费端只执行一次

        使用 redis 将消费过的消息唯一标识存储起来，然后在消费端业务执行之前判断 redis 中是否已经存在这个标识

4.5.2 允许消费端执行多次，保证数据不受影响

4.5.2.1 数据库唯一约束

        如果消费端业务是新增操作，我们可以利用数据库的唯一键约束，比如优惠券流水表的优惠券编号，如果重复消费将会插入两条相同的优惠券编号记录，数据库会给我们报错，可以保证数据库数据不会插入两条

4.5.2.2 数据库乐观锁*

5. 消息转换器

        前面我们发送的消息都是字符串，如果想发送对象，就要用到消息转换器

        消息转换器（Message Converter）是用于将消息在生产者和消费者之间进行序列化和反序列化的组件。在消息传递过程中，生产者将消息对象转换为字节流发送到消息队列，而消费者则将接收到的字节流转换回消息对象进行处理。

5.1 定义一个对象

        对象需要序列化

5.2 创建生产者发送消息方法

  1. @Component
  2. public class OrderProducer {
  3.     @Autowired
  4.     private RabbitTemplate rabbitTemplate;
  5.     public void sendOrder(OrderDTO orderDTO) {
  6.         rabbitTemplate.convertAndSend(MqConfig.ORDER_EXCHANGE,MqConfig.ORDER_KEY,orderDTO);
  7.     }
  8. }

5.3 发送消息

  1. @Test
  2. void testSendOrder() {
  3.     OrderDTO orderDTO = new OrderDTO();
  4.     orderDTO.setOrderSn(UUID.randomUUID().toString());
  5.     orderDTO.setUsername("user");
  6.     orderDTO.setAmount(new BigDecimal("200"));
  7.     orderDTO.setCreateDate(new Date());
  8.     orderProducer.sendOrder(orderDTO);
  9. }

        控制台查看消息：

        上面采用的是JDK序列化方式，可以看出虽然获得了对象，但是得到的数据体积大，可读性差，为了解决这个问题，我们可以通过SpringAMQP的MessageConverter来处理

5.4 Spring的消息转换器

        Spring AMQP提供了多种消息转换器（Message Converter）这些消息转换器使得消息的发送和接收可以使用不同的消息格式，如JSON、XML等，从而更灵活地处理消息数据

5.4.1 生产端配置消息转换器

  1. @Bean
  2. public MessageConverter messageConverter() {
  3.     return new Jackson2JsonMessageConverter();
  4. }
  5. @Bean
  6. public RabbitTemplate rabbitTemplate(ConnectionFactory factory) {
  7.     RabbitTemplate template = new RabbitTemplate();
  8.     template.setConnectionFactory(factory);
  9.     template.setConfirmCallback(callback);
  10.     template.setReturnsCallback(returnsCallback);
  11.     // true 表示消息通过交换机无法路由到队列时候，会把消息返回给生产者
  12.     // false 消息无法路由到队列就直接丢弃
  13.     template.setMandatory(true);
  14.     template.setMessageConverter(messageConverter());
  15.     return template;
  16. }

5.4.2 消费端配置消息转换器

        添加依赖

  1. <dependency>
  2.      <groupId>org.springframework.boot</groupId>
  3.      <artifactId>spring-boot-starter-web</artifactId>
  4. </dependency>

        配置消息转换器

  1. @Configuration
  2. public class MqConfig implements RabbitListenerConfigurer {
  3.     @Resource
  4.     private ObjectMapper objectMapper;
  5.     //将消息转换为JSON格式
  6.     public MappingJackson2MessageConverter messageConverter(){
  7.         MappingJackson2MessageConverter converter = new MappingJackson2MessageConverter();
  8.         converter.setObjectMapper(objectMapper);
  9.         return converter;
  10.     }
  11.  
  12.     @Bean
  13.     public MessageHandlerMethodFactory messageHandlerMethodFactory(){
  14.         DefaultMessageHandlerMethodFactory factory = new DefaultMessageHandlerMethodFactory();
  15.         factory.setMessageConverter(messageConverter());
  16.         return factory;
  17.     }
  18.     @Override
  19.     public void configureRabbitListeners(RabbitListenerEndpointRegistrar rabbitListenerEndpointRegistrar) {
  20.         rabbitListenerEndpointRegistrar.setMessageHandlerMethodFactory(messageHandlerMethodFactory());
  21.     }
  22. }

        创建消费者处理消息方法

  1.     @RabbitHandler
  2.     @RabbitListener(queues = "msg-queue")
  3.     public void receiveOrder(@Payload OrderDTO orderDTO, Channel channel, @Headers Map map){
  4.         System.out.println("Order消息处理："+orderDTO);
  5.         Long tag = (Long)map.get(AmqpHeaders.DELIVERY_TAG);
  6.         try {
  7.             channel.basicAck(tag,false);
  8.         } catch (IOException e) {
  9.             throw new RuntimeException(e);
  10.         }
  11.     }

6.延迟队列

        延时队列就是用来存放需要在指定时间内被处理的消息的队列，是死信队列的一种

应用场景：

1、订单在十分钟之内未支付则自动取消

2、预定会议后，需要在预定的时间点前十分钟通知各个与会人员参加会议

3、用户发起退款，如果三天内没有得到处理则通知

4、用户注册成功后，如果三天内没有登陆则进行短信提醒

6.1 死信队列

        死信队列（Dead Letter Queue，简称DLQ）是一种用于处理消息处理失败或被拒绝的消息的特殊队列。当消息在队列中满足一定条件时，例如消息被消费者拒绝、消息过期、消息处理超时等，这些消息将被发送到死信队列而不是直接被丢弃或忽略

6.2 TTL消息

        TTL（Time To Live）指定消息在队列中存活的时间，超过指定的时间后如果消息还未被消费者消费，则该消息会被自动丢弃或转移到死信队列

6.2.1 生产端配置类配置TTL消息

  1. @Bean
  2. public Queue ttlQueue() {
  3.     Map map = new HashMap();
  4.     map.put("x-message-ttl",5000);
  5.     return new Queue("ttl-queue",false,false,false,map);
  6. }
  7. @Bean
  8. public TopicExchange ttlExchange() {
  9.     return new TopicExchange("ttl-exchange");
  10. }
  11. @Bean
  12. public Binding bindTllQueue() {
  13.     return BindingBuilder.bind(ttlQueue()).to(ttlExchange()).with("ttl.*");
  14. }

6.2.2 创建生产者发送消息方法

  1. public void sendMessage() {
  2.     rabbitTemplate.convertAndSend("ttl-exchange","ttl.msg","hello world");
  3. }

6.2.3 发送消息

  1. @Test
  2. void testSendMessage() {
  3.     msgProducer.sendMessage();;
  4. }

6.3 死信交换机

6.3.1 生产端配置类配置死信交换机

        配置死信队列、交换机并把二者绑定，修改TTL队列失效时放到死信交换机中进而存到死信队列中，而不是直接销毁

  1. @Bean
  2. public Queue ttlQueue(){
  3.     Map map = new HashMap<>();
  4.     map.put("x-message-ttl",5000);
  5.     map.put("x-dead-letter-exchange","dead-exchange"); //死信交换机
  6.     map.put("x-dead-letter-routing-key","dead.msg"); //发送消息时携带路由key
  7.     return new Queue("ttl-queue",false,false,false,map);
  8. }
  9. @Bean
  10. public TopicExchange deadExchange() {
  11.     return new TopicExchange("dead-exchange");
  12. }
  13. @Bean
  14. public Queue deadQueue() {
  15.     return new Queue("dead-queue");
  16. }
  17. @Bean
  18. public Binding bindDeadQueue() {
  19.     return BindingBuilder.bind(deadQueue()).to(deadExchange()).with("dead.#");
  20. }

6.3.2 接收端处理消息

  1. @RabbitHandler
  2. @RabbitListener(queues = {"dead-queue"})
  3. public void receiveDeadMsg(@Payload String msg, Channel channel, @Headers Map headers) {
  4.     Long tag = (Long)headers.get(AmqpHeaders.DELIVERY_TAG);
  5.     System.out.println("处理了已经超时的消息：" + msg);
  6.     try{
  7.         channel.basicAck(tag,false);
  8.     } catch (IOException e) {
  9.         throw new RuntimeException(e);
  10.     }
  11. }

7.延迟插件

        RabbitMQ实现延迟消息的方式有两种，一种是使用死信队列，另一种是使用延迟插件。 

        通过安装插件，自定义交换机，让交换机拥有延迟发送消息的能力，从而实现延迟消息，相较于死信队列延迟插件只需创建一个交换机和一个队列，使用起来简单

7.1 延迟插件下载安装

Releases · rabbitmq/rabbitmq-delayed-message-exchange · GitHubDelayed Messaging for RabbitMQ. Contribute to rabbitmq/rabbitmq-delayed-message-exchange development by creating an account on GitHub.icon-default.png?t=N7T8https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases        将插件文件复制到RabbitMQ安装目录的plugins目录下，然后进入RabbitMQ安装目录的sbin目录下，使用如下命令启用延迟插件；

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

7.2 SpringBoot中实现延迟插件

7.2.1 开发生产者端

7.2.1.1 配置交换机、队列和绑定关系
  1.     /**
  2.      * 订单延迟插件消息队列所绑定的交换机
  3.      */
  4.     @Bean
  5.     DirectExchange orderCancelExchange() {
  6.        return  ExchangeBuilder.directExchange("order-delay-exchange")
  7.                .delayed().durable(true)
  8.                 .build();
  9.     }
  10.     /**
  11.      * 订单延迟插件队列
  12.      */
  13.     @Bean
  14.     public Queue orderCancelQueue() {
  15.         return new Queue("order-delay-queue");
  16.     }
  17.     /**
  18.      * 将订单延迟插件队列绑定到交换机
  19.      */
  20.     @Bean
  21.     public Binding bindOrderCancelQueue() {
  22.         return BindingBuilder.bind(orderCancelQueue())
  23.                 .to(orderCancelExchange()).with("delay.order.key");
  24.     }
7.2.1.2 创建发送消息方法

        通过给消息设置x-delay头来设置消息从交换机发送到队列的延迟时间 

  1. @Component
  2. public class CancelOrderSender {
  3.     @Resource
  4.     private RabbitTemplate rabbitTemplate;
  5.     public void sendMessage(Long orderId,Long delayTime) {
  6.         rabbitTemplate.convertAndSend("order-delay-exchange", "delay.order.key", orderId, new MessagePostProcessor() {
  7.             @Override
  8.             public Message postProcessMessage(Message message) throws AmqpException {
  9.                 //给消息设置延迟毫秒值
  10.                 message.getMessageProperties().setHeader("x-delay",delayTime);
  11.                 return message;
  12.             }
  13.         });
  14.     }
  15. }
7.2.1.3  Service层
  1. @Service
  2. public class OrderServiceImpl {
  3.     @Resource
  4.     private CancelOrderSender cancelOrderSender;
  5.     public void createOrder() {
  6.         System.out.println("下单后生成订单ID");
  7.         Long orderId = 1001L;
  8.         sendDelayMessageCancelOrder(orderId);
  9.     }
  10.     public void sendDelayMessageCancelOrder(Long orderId) {
  11.         //获取订单超时时间，假设为5秒
  12.         long delayTimes = 5 * 1000;
  13.         cancelOrderSender.sendMessage(orderId,delayTimes);
  14.     }
  15. }
7.2.1.4 Controller层
  1. @RestController
  2. @RequestMapping("/order")
  3. public class OrderController {
  4.     @Resource
  5.     private OrderServiceImpl orderService;
  6.     @PostMapping
  7.     public String create() {
  8.         orderService.createOrder();
  9.         return "success";
  10.     }
  11. }

7.2.2 开发消费者端

7.2.2.1 Service层
  1. @Service
  2. public class OrderServiceImpl {
  3.     public void cancelOrder(Long orderId) {
  4.         System.out.println("查询订单编号为：" + orderId + "订单状态，如果是待支付状态，则更新为已失效");
  5.     }
  6. }
7.2.2.2 创建处理消息的方法
  1. @Component
  2. public class CancelOrderReceiver {
  3.     @Autowired
  4.     private OrderServiceImpl orderService;
  5.     @RabbitHandler
  6.     @RabbitListener(queues =  {"order-delay-queue"})
  7.     public void handle(@Payload Long orderId, Channel channel, @Headers Map headers) {
  8.         orderService.cancelOrder(orderId);
  9.         Long tag = (Long)headers.get(AmqpHeaders.DELIVERY_TAG);
  10.         try {
  11.             channel.basicAck(tag,false);
  12.         } catch (IOException e) {
  13.             throw new RuntimeException(e);
  14.         }
  15.     }
  16. }
声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/小丑西瓜9/article/detail/614606
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号