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

devtools/2025/1/1 19:55:42/

目录

事务

消息分发?

应用场景

?1. 限流

2.负载均衡?

事务

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的事务

声明队列：

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

配置事务管理器：

    @Beanpublic RabbitTransactionManager rabbitTransactionManager(ConnectionFactory connectionFactory) {return new RabbitTransactionManager(connectionFactory);}@Bean("transRabbitTemple")public RabbitTemplate transRabbitTemple(ConnectionFactory connectionFactory) {RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);//开启事务rabbitTemplate.setChannelTransacted(true);return  rabbitTemplate;}

生产者代码编写：

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

测试：

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

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

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

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

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

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

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

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

此结果一切正常

消息分发

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）配置交换机,队列

package com.bite.extensions.config;import com.bite.extensions.constant.Constants;
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class QosConfig {@Bean("qosQueue")public Queue qosQueue() {return QueueBuilder.durable(Constants.QOS_QUEUE).build();}@Bean("qosExchange")public DirectExchange qosExchange() {return  ExchangeBuilder.directExchange(Constants.QOS_EXCHANGE).build();}@Bean("qosBinding")public Binding qosBinding(@Qualifier("qosQueue") Queue queue, @Qualifier("qosExchange") DirectExchange directExchange) {return BindingBuilder.bind(queue).to(directExchange).with("qos");}
}

  1. 生产者

     @RequestMapping("/qos")public String qos() {System.out.println("qos test...");for (int i = 0; i < 15; i++) {rabbitTemplate.convertAndSend(Constants.QOS_EXCHANGE, "qos", "qos test i..."+i);}return "消息发送成功";}

4）消费者

package com.bite.extensions.listener;import com.bite.extensions.constant.Constants;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;@Component
public class QosListener {@RabbitListener(queues = Constants.QOS_QUEUE)public void handleMessage(Message message, Channel channel) throws Exception {//消费者逻辑long deliverTag = message.getMessageProperties().getDeliveryTag();try {System.out.printf("[qos.queue]接收到信息: %s, deliveryTag: %d
",new String(message.getBody(),"UTF-8"),deliverTag);
/*            //业务逻辑处理System.out.println("业务逻辑处理！");//肯定确认channel.basicAck(deliverTag,false);*/} catch (Exception e) {//否定确认channel.basicNack(deliverTag,false,true);//requeue为false，则变成死信队列}}
}

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

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

测试2

把 prefetch: 5 注掉 再观看结果

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

2.负载均衡

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

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

消费者：

package com.bite.extensions.listener;import com.bite.extensions.constant.Constants;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;@Component
public class QosListener {@RabbitListener(queues = Constants.QOS_QUEUE)public void handleMessage(Message message, Channel channel) throws Exception {//消费者逻辑long deliverTag = message.getMessageProperties().getDeliveryTag();try {System.out.printf("第一个消费者 接收到信息: %s, deliveryTag: %d
",new String(message.getBody(),"UTF-8"),deliverTag);Thread.sleep(3000);channel.basicAck(deliverTag,false);} catch (Exception e) {//否定确认channel.basicNack(deliverTag,false,true);//requeue为false，则变成死信队列}}@RabbitListener(queues = Constants.QOS_QUEUE)public void handleMessage2(Message message, Channel channel) throws Exception {//消费者逻辑long deliverTag = message.getMessageProperties().getDeliveryTag();try {System.out.printf("第二个消费者 接收到信息: %s, deliveryTag: %d
",new String(message.getBody(),"UTF-8"),deliverTag);Thread.sleep(1000);channel.basicAck(deliverTag,false);} catch (Exception e) {//否定确认channel.basicNack(deliverTag,false,true);//requeue为false，则变成死信队列}}
}

结果：

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

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


http://www.ppmy.cn/devtools/146524.html

相关文章

什么是Spring Boot 应用开发？

什么是Spring Boot 应用开发？

一、引言 在当今的软件开发领域&#xff0c;Java 依然占据着重要的地位&#xff0c;而 Spring Boot 作为 Java 生态系统中极具影响力的框架&#xff0c;极大地简化了企业级应用的开发流程&#xff0c;提升了开发效率和应用的可维护性。它基于 Spring 框架构建&#xff0c;通过约…
阅读更多...
使用内网穿透工具，为树莓派配置公网地址实现远程ssh

使用内网穿透工具，为树莓派配置公网地址实现远程ssh

前言 在我们使用树莓派时&#xff0c;我们常使用ssh连接&#xff0c;但是每次连接都需要在局域网中才能连接&#xff0c;这并不是很方便。本文将介绍使用内网穿透工具&#xff0c;实现远程ssh; 开启树莓派ssh服务 具体可见专栏中的文章:https://blog.csdn.net/qq_64047342/ar…
阅读更多...
bash shell的条件语句

bash shell的条件语句

&#xff5e; script% touch if.sh &#xff5e; script% chmod 755 if.sh1.if-then-fi #!/usr/bin/env bashFOOD$1 if [ $FOOD"apple" ] thenecho The food is $FOOD fi exit 0~ script % ./if.sh apple The food is apple如果要将多条语句写在一行&#xff0c;可以…
阅读更多...
【PostgreSQL】入门篇——在不同操作系统上安装 PostgreSQL

【PostgreSQL】入门篇——在不同操作系统上安装 PostgreSQL

PostgreSQL在 Windows、macOS 和 Linux&#xff08;以 Ubuntu 为例&#xff09;的安装步骤&#xff0c;以及可能出现的问题和解决办法。 一、在 Windows 上安装 PostgreSQL 1. 下载 PostgreSQL 安装程序 访问 PostgreSQL 官方网站&#xff1a;PostgreSQL Downloads点击“Dow…
阅读更多...
在C语言基础上的C++第二章（类和对象）

在C语言基础上的C++第二章（类和对象）

1&#xff1a;面向对象的程序设计 我们学习过的C语言是一种面向过程的程序设计。思想是把问题分割成一个个函数&#xff0c;然后用主函数把它们串联起来。 而C是面向对象的程序设计。面向对象的程序设计&#xff08;Object - Oriented Programming&#xff0c;简称 OOP&#…
阅读更多...
Llama 3 后训练（三）

Llama 3 后训练（三）

目录 4. 后训练 4.1 建模 图表解读 4.1.1 聊天对话格式 4.1.2 奖励建模 4.1.3 监督微调&#xff08;Supervised Finetuning&#xff09; 4.1.4 直接偏好优化&#xff08;Direct Preference Optimization&#xff09; 4.1.5 模型平均&#xff08;Model Averaging&#x…
阅读更多...
3.微服务灰度发布落地实践(组件灰度增强)

3.微服务灰度发布落地实践(组件灰度增强)

文章目录 前言调用链示图dubbo服务之间的的调链cloud 服务之间的调用链 网关servlet容器: 标签续传1.定义插件2.实现灰度增强拦截 线程池: 标签续传1.拦截Runnable或Callable,接口增强实现标签续传;Callable 插件定义Runnable 插件定义拦载Callabl或Runnable构造(可共用)拦载ru…
阅读更多...
使用Optional类：Java编程中处理可能为空的值的一种优雅方式

使用Optional类：Java编程中处理可能为空的值的一种优雅方式

使用Optional类&#xff1a;Java编程中处理可能为空的值的一种优雅方式 引言 在Java开发过程中&#xff0c;null值一直是引发NullPointerException&#xff08;空指针异常&#xff09;的主要原因&#xff0c;这不仅会导致程序崩溃&#xff0c;还会增加调试和维护的成本。为了…
阅读更多...
最新文章

Copyright @ 2022~2023