网站二维码弹窗,上海建立公司网站,网站开发实用技术pdf,浦口区网站建设售后保障目录
事务
消息分发
应用场景 1. 限流
2.负载均衡 事务 RabbitMQ是基于AMQP协议实现的,该协议实现了事务机制,因此RabbitMQ也支持事务机制.SpringAMQP也提供了对事务相关的操作.RabbitMQ事务允许开发者确保消息的发送和接收是原子性的,要么全部成功,要么全部失败. 何为原…目录
事务
消息分发
应用场景 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 带异常的发送 看看会发生什么 TransactionalRequestMapping(/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 不带异常的发送 看看会发生什么 TransactionalRequestMapping(/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);}
}3) 生产者 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\n,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只是在消息进入队列时分派消息.它不考虑消费者未确认消息的数量. 我们可以使用设置prefetch1的⽅式,告诉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\n,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\n,new String(message.getBody(),UTF-8),deliverTag);Thread.sleep(1000);channel.basicAck(deliverTag,false);} catch (Exception e) {//否定确认channel.basicNack(deliverTag,false,true);//requeue为false则变成死信队列}}
}结果 这里可以看出每个消费者以不同的速度完成某项任务 以防止一个消费者未完成等很久的情况 结语 写博客不仅仅是为了分享学习经历同时这也有利于我巩固知识点总结该知识点由于作者水平有限对文章有任何问题的还请指出接受大家的批评让我改进。同时也希望读者们不吝啬你们的点赞收藏关注你们的鼓励是我创作的最大动力
