0
0

【Rabbitmq篇】高级特性----事务，消息分发

事务

RabbitMQ是基于AMQP协议实现的,该协议实现了事务机制,因此RabbitMQ也支持事务机制.SpringAMQP也提供了对事务相关的操作.RabbitMQ事务允许开发者确保消息的发送和接收是原子性的,要么全部成功,要么全部失败.

何为原子性（面试重点）？

例如： 当A向B转账1000元，会经历俩个步骤

1.A 向 B 转账 1000元 A的账号将会减去1000元

2.B将会收到1000元 B的账号将会增加1000元

可是，如果遇到极端情况，当A向B转账1000元时，A-1000元已完成，这个时候系统出现故障，导致A-1000 但是B却没有接收到 那么1000元将无缘无故丢失了 ，肯定不会允许这种事情发生，不然谁还敢转账。

此时就是将1操作和2操作绑定在一起，要么同时完成，要么一个都不执行

当出现1执行失败的时候，将1操作进行“回滚”，回到原来的状态，就当一切都没发生过

接下来实现rabbitmq的事务

声明队列：

  1.     //事务
  2.     public static final String TRANS_QUEUE = "trans_queue";
  4.     @Bean("transQueue")
  5.     public Queue transQueue() {
  6.         return QueueBuilder.durable(Constants.TRANS_QUEUE).build();
  7.     }

** 配置事务管理器：**

  1.     @Bean
  2.     public RabbitTransactionManager rabbitTransactionManager(ConnectionFactory connectionFactory) {
  3.         return new RabbitTransactionManager(connectionFactory);
  4.     }
  5.     @Bean("transRabbitTemple")
  6.     public RabbitTemplate transRabbitTemple(ConnectionFactory connectionFactory) {
  7.         RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
  8.         //开启事务
  9.         rabbitTemplate.setChannelTransacted(true);
  10.         return  rabbitTemplate;
  11.     }

生产者代码编写：

  1.     @RequestMapping("/trans")
  2.     public String trans() {
  3.         System.out.println("trans test...");
  4.         transRabbitTemplate1.convertAndSend(Constants.TRANS_EXCHANGE, "trans", "trans test 1...");
  5.         int num = 5/0;
  6.         transRabbitTemplate1.convertAndSend(Constants.TRANS_EXCHANGE, "trans", "trans test 2...");
  7.         return "消息发送成功";
  8.     }

测试：

1）不带 @Transactional 带异常的发送 看看会发生什么？

**此时只有发送的第一条消息，紧接着发生了异常导致第二条消息未发送成功 **

2) 带 @Transactional 带异常的发送 看看会发生什么？ 

  1.     @Transactional
  2.     @RequestMapping("/trans")
  3.     public String trans() {
  4.         System.out.println("trans test...");
  5.         transRabbitTemplate1.convertAndSend(Constants.TRANS_EXCHANGE, "trans", "trans test 1...");
  6.         int num = 5/0;
  7.         transRabbitTemplate1.convertAndSend(Constants.TRANS_EXCHANGE, "trans", "trans test 2...");
  8.         return "消息发送成功";
  9.     }

此时发生异常 本来发送了一条消息 但有异常，进行了回滚，当做没发生

也证明了我们事务的可靠性

3）带 @Transactional 不带异常的发送 看看会发生什么？

  1.     @Transactional
  2.     @RequestMapping("/trans")
  3.     public String trans() {
  4.         System.out.println("trans test...");
  5.         transRabbitTemplate1.convertAndSend(Constants.TRANS_EXCHANGE, "trans", "trans test 1...");
  6. //        int num = 5/0;
  7.         transRabbitTemplate1.convertAndSend(Constants.TRANS_EXCHANGE, "trans", "trans test 2...");
  8.         return "消息发送成功";
  9.     }

此结果一切正常

消息分发

RabbitMQ队列拥有多个消费者时,队列会把收到的消息分派给不同的消费者.每条消息只会发送给订阅列表里的⼀个消费者.这种方式⾮常适合扩展,如果现在负载加重,那么只需要创建更多的消费者来消费处理消息即可.

默认情况下,RabbitMQ是以轮询的方法进行分发的,而不管消费者是否已经消费并已经确认了消息.这种方式是不太合理的,试想⼀下,如果某些消费者消费速度慢,而某些消费者消费速度快,就可能会导致某些消费者消息积压,某些消费者空闲,进而应用整体的吞吐量下降.

这样A都做完了10个任务，B还在写第一个任务，这样将会大大影响效率，从而导致整个的效率下降

如何处理呢我们可以使用前面章节讲到的channel.basicQos(intprefetchCount)方法,来限制当前信道上的消费者所能保持的最大未确认消息的数量

比如:消费端调用了channelbasicQos(1),

此时A接收1条信息，并且消费1条 B同时也接收1条信息 但是它效率比较慢 所有它还在消费 而A处理完1条消息又接着处理第二条消息，属于多劳多得，并不会因为B影响整体的效率

应用场景

1. 限流

如下使用场景:
订单系统每秒最多处理5000请求,正常情况下,订单系统可以正常满足需求
但是在秒杀时间点,请求瞬间增多,每秒1万个请求,如果这些请求全部通过MQ发送到订单系统,无疑会把订单系统压垮.

RabbitMQ提供了限流机制,可以控制消费端⼀次只拉取N个请求
通过设置prefetchCount参数,同时也必须要设置消息应答方式为手动应答
prefetchCount:控制消费者从队列中预取(prefetch)消息的数量,以此来实现流控制和负载均衡.

**1） 配置prefetch参数,设置应答方式为手动应答 **

** 2） 配置交换机,队列**

  1. package com.bite.extensions.config;
  3. import com.bite.extensions.constant.Constants;
  4. import org.springframework.amqp.core.*;
  5. import org.springframework.beans.factory.annotation.Qualifier;
  6. import org.springframework.context.annotation.Bean;
  7. import org.springframework.context.annotation.Configuration;
  9. @Configuration
  10. public class QosConfig {
  12.     @Bean("qosQueue")
  13.     public Queue qosQueue() {
  14.         return QueueBuilder.durable(Constants.QOS_QUEUE).build();
  15.     }
  16.     @Bean("qosExchange")
  17.     public DirectExchange qosExchange() {
  18.         return  ExchangeBuilder.directExchange(Constants.QOS_EXCHANGE).build();
  19.     }
  20.     @Bean("qosBinding")
  21.     public Binding qosBinding(@Qualifier("qosQueue") Queue queue, @Qualifier("qosExchange") DirectExchange directExchange) {
  22.         return BindingBuilder.bind(queue).to(directExchange).with("qos");
  23.     }
  24. }
  1. 生产者
  1.     @RequestMapping("/qos")
  2.     public String qos() {
  3.         System.out.println("qos test...");
  4.         for (int i = 0; i < 15; i++) {
  5.             rabbitTemplate.convertAndSend(Constants.QOS_EXCHANGE, "qos", "qos test i..."+i);
  6.         }
  7.         return "消息发送成功";
  8.     }

4）消费者

  1. package com.bite.extensions.listener;
  3. import com.bite.extensions.constant.Constants;
  4. import com.rabbitmq.client.Channel;
  5. import org.springframework.amqp.core.Message;
  6. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  7. import org.springframework.stereotype.Component;
  9. @Component
  10. public class QosListener {
  11.     @RabbitListener(queues = Constants.QOS_QUEUE)
  12.     public void handleMessage(Message message, Channel channel) throws Exception {
  13.         //消费者逻辑
  14.         long deliverTag = message.getMessageProperties().getDeliveryTag();
  15.         try {
  16.             System.out.printf("[qos.queue]接收到信息: %s, deliveryTag: %d\n",new String(message.getBody(),"UTF-8"),deliverTag);
  17. /*            //业务逻辑处理
  18.             System.out.println("业务逻辑处理！");
  19.             //肯定确认
  20.             channel.basicAck(deliverTag,false);*/
  21.         } catch (Exception e) {
  22.             //否定确认
  23.             channel.basicNack(deliverTag,false,true);//requeue为false，则变成死信队列
  24.         }
  25.     }
  26. }

** 5）测试1 未设置肯定确认情况**

此时将会只接收到5条，并且会阻塞住，达到一个限流的状态

**测试2 **

把 prefetch: 5 注掉 再观看结果

此时将会一次性把队列的消息全部发送，并且全部消费

2.负载均衡

如下图,在有两个消费者的情况下，⼀个消费者处理任务非常快,另⼀个非常慢,就会造成⼀个消费者会⼀直很忙,而另⼀个消费者很闲.这是因为RabbitMQ只是在消息进入队列时分派消息.它不考虑消费者未确认消息的数量.

**我们可以使用设置prefetch=1的⽅式,告诉RabbitMQ⼀次只给⼀个消费者⼀条消息,也就是说,在处理并确认前⼀条消息之前,不要向该消费者发送新消息.相反,它会将它分派给下⼀个不忙的消费者. **

消费者： 

  1. package com.bite.extensions.listener;
  3. import com.bite.extensions.constant.Constants;
  4. import com.rabbitmq.client.Channel;
  5. import org.springframework.amqp.core.Message;
  6. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  7. import org.springframework.stereotype.Component;
  9. @Component
  10. public class QosListener {
  11.     @RabbitListener(queues = Constants.QOS_QUEUE)
  12.     public void handleMessage(Message message, Channel channel) throws Exception {
  13.         //消费者逻辑
  14.         long deliverTag = message.getMessageProperties().getDeliveryTag();
  15.         try {
  16.             System.out.printf("第一个消费者 接收到信息: %s, deliveryTag: %d\n",new String(message.getBody(),"UTF-8"),deliverTag);
  17.             Thread.sleep(3000);
  18.             channel.basicAck(deliverTag,false);
  19.         } catch (Exception e) {
  20.             //否定确认
  21.             channel.basicNack(deliverTag,false,true);//requeue为false，则变成死信队列
  22.         }
  23.     }
  24.     @RabbitListener(queues = Constants.QOS_QUEUE)
  25.     public void handleMessage2(Message message, Channel channel) throws Exception {
  26.         //消费者逻辑
  27.         long deliverTag = message.getMessageProperties().getDeliveryTag();
  28.         try {
  29.             System.out.printf("第二个消费者 接收到信息: %s, deliveryTag: %d\n",new String(message.getBody(),"UTF-8"),deliverTag);
  30.             Thread.sleep(1000);
  31.             channel.basicAck(deliverTag,false);
  32.         } catch (Exception e) {
  33.             //否定确认
  34.             channel.basicNack(deliverTag,false,true);//requeue为false，则变成死信队列
  35.         }
  36.     }
  37. }

** 结果：**

这里可以看出每个消费者以不同的速度完成某项任务 以防止一个消费者未完成等很久的情况

结语： 写博客不仅仅是为了分享学习经历，同时这也有利于我巩固知识点，总结该知识点，由于作者水平有限，对文章有任何问题的还请指出，接受大家的批评，让我改进。同时也希望读者们不吝啬你们的点赞+收藏+关注，你们的鼓励是我创作的最大动力！

标签： rabbitmq 分布式 spring
本文转载自: https://blog.csdn.net/chaodddddd/article/details/144080620
版权归原作者 呼啦啦啦啦啦啦啦啦 所有， 如有侵权，请联系我们删除。
登录后发布评论

“【Rabbitmq篇】高级特性----事务，消息分发”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

网络安全法-网络运行安全

使用selenium/drissionpage时如何阻止chrome自动跳转http到https

docker desktop 里部署的Open WebUI 管理员密码忘记了的处理方法

在ubuntu20.04中搭建onsite比赛运行环境

利用开源的低代码表单设计器FcDesigner高效管理和渲染复杂表单结构

Kafka学习笔记

【前端】浏览器输入url到页面呈现发生了什么？

文章导航
事务消息分发