RabbitMQ 长期化与不公中分发

RabbitMQ 长期化与不公中分发

1. RabbitMQ 长期化 (Durability)

概念

长期化是指将消息或队列生存在磁盘上，以确保纵然 RabbitMQ 服务器宕机或重启，数据也不会丢失。
长期化的三个层面

• 队列长期化：
  队列长期化意味着纵然 RabbitMQ 重启后，队列依然存在，但它不会包管队列中的消息仍旧存在。
  
  
  
  • 队列长期化声明：
    1. channel.queueDeclare("queue_name", true, false, false, null);

    复制代码
    此中，true 表现队列长期化。
    
    
    
    
  
  
• 消息长期化：
  消息长期化是在生产者发送消息时指定的，确保消息在服务器重启后依然可以大概保存在队列中。
  
  
  
  • 消息长期化声明：
    1. channel.basicPublish("", "queue_name", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());

    复制代码
  留意：队列长期化和消息长期化是独立的。纵然队列是长期化的，消息也须要单独设置为长期化。
  
  
• 交换机（Exchange）长期化：
  交换机也可以设置为长期化，确保 RabbitMQ 重启后，交换机不会丢失。
  
  
  
  • 交换机长期化声明：
    1. channel.exchangeDeclare("exchange_name", "direct", true);

    复制代码
    此中，true 表现交换机长期化。
    
  
  

长期化的留意事项

• 长期化利用会稍微影响性能，由于须要将数据写入磁盘。
  
• 队列长期化并不会让未长期化的消息在 RabbitMQ 重启后规复，消息也须要设置为长期化。
  

---

2. 不公中分发 (Fair Dispatch)

概念

RabbitMQ 默认接纳轮询的方式来分发消息给斲丧者，即每个斲丧者会按照同等的序次吸收消息。这种机制在某些环境下会导致某些斲丧者的任务积存过多，而其他斲丧者的任务处理处罚过快，导致体系资源浪费。这时，我们可以利用不公中分发（也称为预取机制）来优化分发过程。
不公中分发实现

不公中分发的实现方式是通过设置 Qos（Quality of Service） 预取值，使得每个斲丧者一次只会吸收特定命量的消息，直到这些消息处理处罚完并发送应答后，才会吸收新的消息。
步调

• 设置 QoS：
  在斲丧者端利用 basicQos 方法来设置每次吸收的消息数量。通常设置为 1，表现斲丧者一次只会吸收一条消息，在处理处罚完并应答之后再吸收下一条消息。
  1. // 设置为每次处理一条消息
  2. channel.basicQos(1);

  复制代码
• 手动消息应答：
  RabbitMQ 的默认活动是主动应答，即斲丧者吸收到消息后立刻应答，岂论是否处理处罚完成。要实现不公中分发，须要手动确认消息，在消息处理处罚完之后再发送应答。
  
  
  
  • 手动应答代码示例：
    1. boolean autoAck = false;  // 关闭自动应答
    2. channel.basicConsume(queueName, autoAck, deliverCallback, consumerTag -> {});
    4. // 消息处理完成后，手动应答
    5. channel.basicAck(deliveryTag, false);

    复制代码
  
  

不公中分发的长处

• 制止斲丧者压力过大：某些斲丧者大概任务处理处罚较慢，通过 QoS 设置，可以制止一次性收到大量任务导致压力过大。
  
• 优化资源利用：确保任务更公道地分配到斲丧者，提拔团体的任务处理处罚服从。
  

不公中分发的留意事项

• basicQos(1) 的设置会稍微低落吞吐量，但可以确保消息公中分配给斲丧者。
  
• 必须确保消息的手动应答，否则未应答的消息大概会再次发送给其他斲丧者，导致重复斲丧。
  

---

3. RabbitMQ 长期化与不公中分发联合

在现实应用中，RabbitMQ 长期化与不公中分发可以联合利用，以确保消息的可靠性和斲丧者的负载均衡。

• 长期化 确保消息和队列在服务重启时不会丢失，包管了体系的可靠性。
  
• 不公中分发 则通过手动应答和 QoS 设置，确保斲丧者可以根据本身的处理处罚本事，公平地获取消息并举行处理处罚，防止某个斲丧者处理处罚任务过于痴钝。
  

代码示例

1. // 创建通道并声明持久化队列
2. Channel channel = RabbitMqUtils.getChannel();
3. channel.queueDeclare("task_queue", true, false, false, null);
5. // 生产者发送持久化消息
6. String message = "Hello World";
7. channel.basicPublish("", "task_queue", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
9. // 消费者设置 QoS 和手动应答
10. channel.basicQos(1);
11. boolean autoAck = false;  // 关闭自动应答
12. DeliverCallback deliverCallback = (consumerTag, delivery) -> {
13.     String receivedMessage = new String(delivery.getBody(), "UTF-8");
14.     System.out.println("Received: " + receivedMessage);
15.    
16.     // 模拟任务处理
17.     doWork(receivedMessage);
18.    
19.     // 手动应答
20.     channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
21. };
23. // 消费消息
24. channel.basicConsume("task_queue", autoAck, deliverCallback, consumerTag -> {});

复制代码