文章目录
- 前言
- 消息分发
- RabbitMQ 分发机制的应用场景
- 1. 限流
- 2. 负载均衡
前言
当 RabbitMQ 的队列绑定了多个消费者的时候，队列会把消息分发给不同的消费者，每条消息只会发送给订阅列表的一个消费者，但是呢，RabbitMQ 默认是以轮询的方式进行分发的，而不会管消费者是否已经消费并且已经确认了消息，这种方式其实是不合理的，因为每个消费者的消费能力是不同的，如果某个消费者的消费能力很低，那么就会导致其他的消费者已经消费完成所有消息了，但是这个消费者还有很多消息需要消费，这样就会导致消息积压。那么该任何解决这个问题呢？就是我们这篇文章需要讲到的消息分发。
消息分发
RabbitMQ 的消息分发机制主要分为两种：1. 轮询分发 2. 非公平分发
- 轮询分发
- 在默认情况下，RabbitMQ采用轮询的方式将队列中的消息分发给消费者。这意味着如果有多个消费者订阅了同一个队列，RabbitMQ会尝试公平地将消息依次分发给每个消费者。
- 轮询分发机制确保了消息在多个消费者之间的均衡分配，避免了某个消费者过载而其他消费者空闲的情况。
- 非公平分发
- 为了更好地控制消息的分发过程，RabbitMQ提供了非公平分发的机制。在这种机制下，消费者可以通过设置basic.qos方法并指定prefetch_count参数来限制RabbitMQ一次性发送给它的消息数量。
- 通过调整prefetch_count的值，消费者可以根据自己的处理能力来控制消息的分发速度，从而避免因为处理速度不同而导致的消息堆积或空闲。
RabbitMQ 分发机制的应用场景
消息分发的常见应用场景有两个：
- 限流
- 负载均衡
1. 限流
每逢双十一或者其他节日的时候，某些购物平台的订单量会激增，这样就会导致单个服务器接收的订单数量超过了能够承受的范围，所以为了保证我们的订单服务器能够正常运行不发生宕机故障，就需要对服务器接收的消息数量做出限制。
那么如何实现限流的功能呢？我们通过设置 prefetchCount 参数并且设置确认方式为手动确认，prefetchCount 就是控制消费者从队列中预取消息的数量，以此来实现限流和负载均衡。通过设置这个配置，就可以保证消费者中最多只能存在 prefetchCount 个未确认的消息。
spring:
rabbitmq:
listener:
simple:
acknowledge-mode: manual # 设置确认方式为手动确认
prefetch: 5 # 限制消费者只能接收5条消息
public static final String QOS_EXCHANGE = "qos.exchange";
public static final String QOS_QUEUE = "qos.queue";
声明交换机、队列和绑定关系：
@Configuration
public class QosConfig {
@Bean("qosExchange")
public Exchange qosExchange() {
return ExchangeBuilder.directExchange(Constants.QOS_EXCHANGE).build();
}
@Bean("qosQueue")
public Queue qosQueue() {
return QueueBuilder.durable(Constants.QOS_QUEUE).build();
}
@Bean("qosBinding")
public Binding qosBinding(@Qualifier("qosExchange") Exchange exchange, @Qualifier("qosQueue") Queue queue) {
return BindingBuilder.bind(queue).to(exchange).with("qos").noargs();
}
}
生产者：
@RequestMapping("/qos")
public String qos() {
for (int i = 0; i < 20; i++) {
rabbitTemplate.convertAndSend(Constants.QOS_EXCHANGE,"qos","rabbitmq qos" + i);
}
return "消息发送成功";
}
消费者：
@Component
public class QosListener {
@RabbitListener(queues = Constants.QOS_QUEUE)
public void listener1(Message message, Channel channel) throws IOException {
long deliveryTag = message.getMessageProperties().getDeliveryTag();
System.out.println("消费者1接收到消息:" + new String(message.getBody()) + ".deliveryTag:" + deliveryTag);
//channel.basicAck(deliveryTag,false); 手动确认消息，我们这里不确认，看看在没有确认的情况下，队列会向消费者投递多少条消息
}
@RabbitListener(queues = Constants.QOS_QUEUE)
public void listener2(Message message, Channel channel) throws IOException {
long deliveryTag = message.getMessageProperties().getDeliveryTag();
System.out.println("消费者2接收到消息:" + new String(message.getBody()) + ".deliveryTag:" + deliveryTag);
//channel.basicAck(deliveryTag,false); 手动确认消息，我们这里不确认，看看在没有确认的情况下，队列会向消费者投递多少条消息
}
}
可以看到通过设置 prefetchCount 可以实现限流的效果，而如果我们讲这个配置给注掉的话，那么队列中的消息会全部打给消费者：
2. 负载均衡
因为每个服务器的处理业务能力不同，有的服务器处理业务速度很快，而有的服务器处理业务的速度则很慢，如果按照轮询的方式分发消息的话，就会出现某些服务器很忙，有些服务器处理完成业务之后很闲的情况，对于这种情况，我们可以通过设置 prefetchCount 的值为 1 来实现负载均衡。
只有当消费者处理完成消息并且手动确认之后，队列才会继续向其发送下一条消息。
我们修改 prefetchCount 的值为 1，然后其他的代码不需要修改，只是通过 Thread.sleep() 方法来模拟出消费者消费速度不同的情况：
@Component
public class QosListener {
@RabbitListener(queues = Constants.QOS_QUEUE)
public void listener1(Message message, Channel channel) throws IOException, InterruptedException {
long deliveryTag = message.getMessageProperties().getDeliveryTag();
Thread.sleep(1000);
System.out.println("消费者1接收到消息:" + new String(message.getBody()) + ".deliveryTag:" + deliveryTag);
channel.basicAck(deliveryTag,false);
}
@RabbitListener(queues = Constants.QOS_QUEUE)
public void listener2(Message message, Channel channel) throws IOException, InterruptedException {
long deliveryTag = message.getMessageProperties().getDeliveryTag();
Thread.sleep(2000);
System.out.println("消费者2接收到消息:" + new String(message.getBody()) + ".deliveryTag:" + deliveryTag);
channel.basicAck(deliveryTag,false);
}
}
通过设置 prefetchCount 的值为 1，就可以实现出负载均衡的效果。
