电子商务网站建设与开发选择题,西安市建设协会网站,低价的网站建设,南宁整合推广公司Lison dreamlison163.com, v1.0.0, 2023.06.23
RabbitMQ高级特性#xff08;消费端限流、利用限流实现不公平分发、消息存活时间、优先级队列 文章目录 RabbitMQ高级特性#xff08;消费端限流、利用限流实现不公平分发、消息存活时间、优先级队列消费端限流利用限流…Lison dreamlison163.com, v1.0.0, 2023.06.23
RabbitMQ高级特性消费端限流、利用限流实现不公平分发、消息存活时间、优先级队列 文章目录 RabbitMQ高级特性消费端限流、利用限流实现不公平分发、消息存活时间、优先级队列消费端限流利用限流实现不公平分发消息存活时间优先级队列 消费端限流 之前我们讲过MQ可以对请求进行“削峰填谷”即通过消费端限流的方式限制消息的拉取速度达到保护消费端的目的。
1、生产者批量发送消息
Test
public void testSendBatch() {// 发送十条消息for (int i 0; i 10; i) {rabbitTemplate.convertAndSend(my_topic_exchange, my_routing, send message...i);}
}2、消费端配置限流机制
spring:rabbitmq:host: 127.0.0.1port: 5672username: adminpassword: 1233456virtual-host: /listener:simple:# 限流机制必须开启手动签收acknowledge-mode: manual# 消费端最多拉取5条消息消费签收后不满5条才会继续拉取消息。prefetch: 53、消费者监听队列
Component
public class QosConsumer{RabbitListener(queues my_queue)public void listenMessage(Message message, Channel channel) throws IOException, InterruptedException {// 1.获取消息System.out.println(new String(message.getBody()));// 2.模拟业务处理Thread.sleep(3000);// 3.签收消息channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);}
}利用限流实现不公平分发
在RabbitMQ中多个消费者监听同一条队列则队列默认采用的轮询分发。但是在某种场景下这种策略并不是很好例如消费者1处 理任务的速度非常快而其他消费者处理速度却很慢。此时如果采用公平分发则消费者1有很大一部分时间处于空闲状态。此时可以 采用不公平分发即谁处理的快谁处理的消息多
1、生产者批量发送消息
Test
public void testSendBatch() {// 发送十条消息for (int i 0; i 10; i) {rabbitTemplate.convertAndSend(my_topic_exchange, my_routing, send message...i);}
}端配置不公平分发
spring:rabbitmq:host: 127.0.0.1port: 5672username: adminpassword: 1233456virtual-host: /listener:simple:# 限流机制必须开启手动签收acknowledge-mode: manual# 消费端最多拉取1条消息消费这样谁处理的快谁拉取下一条消息实现了不公平分发prefetch: 1编写两个消费者
Component
public class UnfairConsumer {// 消费者1RabbitListener(queues my_queue)public void listenMessage1(Message message, Channel channel) throws Exception{//1.获取消息System.out.println(消费者1:new String(message.getBody(),UTF-8));//2. 处理业务逻辑Thread.sleep(500); // 消费者1处理快//3. 手动签收channel.basicAck(message.getMessageProperties().getDeliveryTag(),true);}// 消费者2RabbitListener(queues my_queue)public void listenMessage2(Message message, Channel channel) throws Exception{//1.获取消息System.out.println(消费者2:new String(message.getBody(),UTF-8));//2. 处理业务逻辑Thread.sleep(3000);// 消费者2处理慢//3. 手动签收channel.basicAck(message.getMessageProperties().getDeliveryTag(),true);}
}消息存活时间
RabbitMQ可以设置消息的存活时间Time To Live简称TTL 当消息到达存活时间后还没有被消费会被移出队列。RabbitMQ 可以对队列的所有消息设置存活时间也可以对某条消息设置存活时间。
设置队列所有消息存活时间
1、在创建队列时设置其存活时间
Configuration
public class RabbitConfig2 {private final String EXCHANGE_NAMEmy_topic_exchange2;private final String QUEUE_NAMEmy_queue2;// 1.创建交换机Bean(bootExchange2)public Exchange getExchange2(){return ExchangeBuilder.topicExchange(EXCHANGE_NAME).durable(true).build();}// 2.创建队列Bean(bootQueue2)public Queue getMessageQueue2(){return QueueBuilder.durable(QUEUE_NAME).ttl(10000) //队列的每条消息存活10s.build();}// 3.将队列绑定到交换机Beanpublic Binding bindMessageQueue2(Qualifier(bootExchange2) Exchange exchange,Qualifier(bootQueue2) Queue queue){return BindingBuilder.bind(queue).to(exchange).with(my_routing).noargs();}
}2、生产者批量生产消息测试存活时间
Test
public void testSendBatch2() throws InterruptedException {// 发送十条消息for (int i 0; i 10; i) {rabbitTemplate.convertAndSend(my_topic_exchange2, my_routing, send message...i);}
}设置单条消息存活时间
Test
public void testSendMessage() {//设置消息属性MessageProperties messageProperties new MessageProperties();//设置存活时间messageProperties.setExpiration(10000);// 创建消息对象Message message new Message(send message....getBytes(StandardCharsets.UTF_8), messageProperties);// 发送消息rabbitTemplate.convertAndSend(my_topic_exchange, my_routing, message);
}注意 1 如果设置了单条消息的存活时间也设置了队列的存活时间以时间短的为准。 2 消息过期后并不会马上移除消息只有消息消费到队列顶端时才会移除该消息。 Test public void testSendMessage2() { for (int i 0; i 10; i) { if (i 5) { // 1.创建消息属性 MessageProperties messageProperties new MessageProperties(); // 2.设置存活时间 messageProperties.setExpiration(“10000”); // 3.创建消息对象 Message message new Message((“send message…” i).getBytes(),messageProperties); // 4.发送消息 rabbitTemplate.convertAndSend(“my_topic_exchange”, “my_routing”, message); } else { rabbitTemplate.convertAndSend(“my_topic_exchange”, “my_routing”, “sendmessage…” i); } } } 在以上案例中i5的消息才有过期时间10s后消息并没有 马上被移除但该消息已经不会被消费了当它到达队列顶 端时会被移除。 优先级队列
假设在电商系统中有一个订单催付的场景即客户在一段时间内未付款会给用户推送一条短信提醒但是系统中分为大型商家和小型商家。比如像苹果小米这样大商家一年能给我们创造很大的利润所以在订单量大时他们的订单必须得到优先处理此时就需要为不同的消息设置不同的优先级此时我们要使用优先级队列。
1、创建队列和交换机
Configuration
public class RabbitConfig3 {private final String EXCHANGE_NAMEpriority_exchange;private final String QUEUE_NAMEpriority_queue;// 1.创建交换机Bean(EXCHANGE_NAME)public Exchange priorityExchange(){return ExchangeBuilder.topicExchange(EXCHANGE_NAME).durable(true).build();}// 2.创建队列Bean(QUEUE_NAME)public Queue priorityQueue(){return QueueBuilder.durable(QUEUE_NAME)//设置队列的最大优先级最大可以设置到255官网推荐不要超过10,如果设置太高比较浪费资源.maxPriority(10).build();}// 3.将队列绑定到交换机Beanpublic Binding bindPriority(Qualifier(EXCHANGE_NAME) Exchange exchange, Qualifier(QUEUE_NAME) Queue queue){return BindingBuilder.bind(queue).to(exchange).with(my_routing).noargs();}
}2、编写生产者
Test
public void testPriority() {for (int i 0; i 10; i) {if (i 5) {// i为5时消息的优先级较高MessageProperties messageProperties new MessageProperties();messageProperties.setPriority(9);Message message new Message((send message... i).getBytes(StandardCharsets.UTF_8), messageProperties);rabbitTemplate.convertAndSend(priority_exchange, my_routing, message);} else {rabbitTemplate.convertAndSend(priority_exchange, my_routing, send message... i);}}
}3、编写消费者
Component
public class PriorityConsumer {RabbitListener(queues priority_queue)public void listenMessage(Message message, Channel channel) throws Exception{//获取消息System.out.println(new String(message.getBody()));//手动签收channel.basicAck(message.getMessageProperties().getDeliveryTag(),true);}
}
