【RabbitMQ】RabbitMQ堆积100万，1000万条消息对应的计谋总结 ...

  
更多相干内容可查看
增加消费者数量

假设当前利用的是 RabbitMQ，且只有一个消费者在处置惩罚消息，该消费者处置惩罚一条消息平均耗时 100 毫秒（即每秒处置惩罚 10 条消息）。那么增加到 10 个消费者后，如果每个消费者的处置惩罚能力相同且不存在资源竞争等题目，理论上处置惩罚速度可提升到每秒 100 条消息。
案例：在一个电商订单处置惩罚体系中，订单创建成功后会发送一条消息到 MQ，本来单个消费者负责处置惩罚订单消息并更新库存、天生物流订单等操作。当消息大量堆积时，可启动多个消费者实例，每个实例都能独立地从 MQ 中获取订单消息并进行处置惩罚。但在实际情况中，若消费者必要对共享资源（如数据库连接池）进行操作，就必要公道配置资源，制止出现连接数过多导致数据库性能下降的题目。比如，利用连接池技术时，根据消费者数量和数据库服务器的负载能力，调整连接池的最大连接数和最小连接数，确保每个消费者都能获取到可用连接且不会过分消耗数据库资源。
示例代码，同时考虑了利用连接池技术来公道管理数据库连接：

1. public class RabbitMQConsumer {
2.     private static final String QUEUE_NAME = "order_queue";
3.     private static final String RABBITMQ_HOST = "localhost";
4.     private static final int RABBITMQ_PORT = 5672;
5.     private static final String RABBITMQ_USERNAME = "guest";
6.     private static final String RABBITMQ_PASSWORD = "guest";
9.     // 数据库连接池配置
10.     private static final String JDBC_URL = "jdbc:mysql://localhost:3306/your_database";
11.     private static final String JDBC_USERNAME = "root";
12.     private static final String JDBC_PASSWORD = "password";
13.     private static final int MAX_POOL_SIZE = 20;
14.     private static final int MIN_POOL_SIZE = 5;
15.     private static final DataSource dataSource;
18.     static {
19.         BasicDataSource ds = new BasicDataSource();
20.         ds.setUrl(JDBC_URL);
21.         ds.setUsername(JDBC_USERNAME);
22.         ds.setPassword(JDBC_PASSWORD);
23.         ds.setInitialSize(MIN_POOL_SIZE);
24.         ds.setMaxTotal(MAX_POOL_SIZE);
25.         dataSource = ds;
26.     }
29.     public static void main(String[] args) {
30.         int numConsumers = 10; // 增加到 10 个消费者
31.         //使用 `ExecutorService` 来创建一个固定大小的线程池，大小为 `numConsumers`（这里是 10），用于启动多个消费者线程
32.         ExecutorService executorService = Executors.newFixedThreadPool(numConsumers);
34.                 //ConnectionFactory用于创建 RabbitMQ 的连接。我们设置了 RabbitMQ 的主机、端口、用户名和密码
35.         try (ConnectionFactory factory = new ConnectionFactory()) {
36.             factory.setHost(RABBITMQ_HOST);
37.             factory.setPort(RABBITMQ_PORT);
38.             factory.setUsername(RABBITMQ_USERNAME);
39.             factory.setPassword(RABBITMQ_PASSWORD);
42.             Connection connection = factory.newConnection();
43.             Channel channel = connection.createChannel();
45.                         //声明了一个持久化的队列，确保队列在 RabbitMQ 服务器重启后仍然存在。
46.             channel.queueDeclare(QUEUE_NAME, true, false, false, null);
49.             // 设置了每个消费者的预取数量为 10，这意味着每个消费者一次最多接收 10 条消息，防止消费者接收过多消息而无法及时处理
50.             channel.basicQos(10);
53.             for (int i = 0; i < numConsumers; i++) {
54.                 executorService.execute(() -> {
55.                     try {
56.                         // 每个消费者创建一个新的 channel 来接收消息
57.                         Channel consumerChannel = connection.createChannel();
58.                         consumerChannel.basicConsume(QUEUE_NAME, false, new DeliverCallback() {
59.                             @Override
60.                             public void handle(String consumerTag, Delivery delivery) throws IOException {
61.                                 String message = new String(delivery.getBody(), "UTF-8");
62.                                 System.out.println("Received: " + message);
65.                                 // 从连接池获取数据库连接
66.                                 try (Connection dbConnection = dataSource.getConnection()) {
67.                                     // 处理订单消息，更新库存、生成物流订单等操作
68.                                     processOrderMessage(dbConnection, message);
71.                                     // 手动确认消息
72.                                     consumerChannel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
73.                                 } catch (SQLException e) {
74.                                     e.printStackTrace();
75.                                     // 消息处理失败，不确认消息，将重新放回队列
76.                                 }
77.                             }
78.                         }, consumerTag -> {
79.                         });
80.                     } catch (IOException | TimeoutException e) {
81.                         e.printStackTrace();
82.                     }
83.                 });
84.             }
87.             // 等待一段时间后关闭资源
88.             Thread.sleep(60000);
89.             executorService.shutdown();
90.             channel.close();
91.             connection.close();
92.         } catch (IOException | TimeoutException | InterruptedException | SQLException e) {
93.             e.printStackTrace();
94.         }
95.     }
98.     private static void processOrderMessage(Connection connection, String message) throws SQLException {
99.         String sql = "INSERT INTO order_processing_log (message) VALUES (?)";
100.         try (PreparedStatement preparedStatement = connection.prepareStatement(sql)) {
101.             preparedStatement.setString(1, message);
102.             preparedStatement.executeUpdate();
103.         }
104.     }
105. }

复制代码

优化消费者处置惩罚逻辑

• 批量处置惩罚
  例如在一个数据收集体系中，MQ 中堆积了大量的传感器数据消息，每个消息包罗单个传感器的一次数据收罗结果。本来消费者是收到一条消息就将数据插入数据库。优化后，可以每收集 100 条消息，将这 100 个传感器数据组合成一个批量插入语句，一次性插入数据库。如许可以大大减少数据库事件的开启和关闭次数，进步处置惩罚服从。比如，利用 MySQL 数据库时，本来逐条插入 100 万条数据可能必要数小时，而接纳批量插入（假设每次批量插入 100 条）可能将时间收缩到几非常钟。
  

示例：

1. public class BatchProcessingConsumer {
2.     private static final String QUEUE_NAME = "sensor_data_queue";
3.     private static final String RABBITMQ_HOST = "localhost";
4.     private static final int RABBITMQ_PORT = 5672;
5.     private static final String RABBITMQ_USERNAME = "guest";
6.     private static final String RABBITMQ_PASSWORD = "guest";
7.     private static final String JDBC_URL = "jdbc:mysql://localhost:3306/your_database";
8.     private static final String JDBC_USERNAME = "root";
9.     private static final String JDBC_PASSWORD = "password";
10.     private static final int BATCH_SIZE = 100;
13.     public static void main(String[] args) {
14.         Connection connection = null;
15.         Channel channel = null;
16.         List<String> batch = new ArrayList<>();
19.         try {
20.             ConnectionFactory factory = new ConnectionFactory();
21.             factory.setHost(RABBITMQ_HOST);
22.             factory.setPort(RABBITMQ_PORT);
23.             factory.setUsername(RABBITMQ_USERNAME);
24.             factory.setPassword(RABBITMQ_PASSWORD);
27.             connection = DriverManager.getConnection(JDBC_URL, JDBC_USERNAME, JDBC_PASSWORD);
28.             channel = factory.newConnection().createChannel();
31.             channel.queueDeclare(QUEUE_NAME, true, false, false, null);
34.             channel.basicConsume(QUEUE_NAME, true, new DeliverCallback() {
35.                 @Override
36.                 public void handle(String consumerTag, Delivery delivery) throws IOException {
37.                     String message = new String(delivery.getBody(), "UTF-8");
38.                     batch.add(message);
41.                     if (batch.size() >= BATCH_SIZE) {
42.                         insertBatch(connection, batch);
43.                         batch.clear();
44.                     }
45.                 }
46.             }, consumerTag -> {
47.             });
50.         } catch (IOException | SQLException e) {
51.             e.printStackTrace();
52.         } finally {
53.             try {
54.                 if (channel!= null) {
55.                     channel.close();
56.                 }
57.                 if (connection!= null) {
58.                     connection.close();
59.                 }
60.             } catch (IOException | SQLException e) {
61.                 e.printStackTrace();
62.             }
63.         }
64.     }
67.     private static void insertBatch(Connection connection, List<String> batch) {
68.         String sql = "INSERT INTO sensor_data (data) VALUES (?)";
69.         try (PreparedStatement preparedStatement = connection.prepareStatement(sql)) {
70.             for (String message : batch) {
71.                 preparedStatement.setString(1, message);
72.                 preparedStatement.addBatch();
73.             }
74.             preparedStatement.executeBatch();
75.         } catch (SQLException e) {
76.             e.printStackTrace();
77.         }
78.     }
79. }

复制代码

• 异步处置惩罚
  以一个用户注册体系为例，当用户注册成功后，会向 MQ 发送一条注册成功消息，消费者必要记录注册日志、发送注册成功邮件以及更新用户统计信息。此中，记录注册日志相对不那么紧急，可以将其异步处置惩罚。比如，消费者在收到注册成功消息后，立即在主线程中发送注册成功邮件和更新用户统计信息，而将记录注册日志的操作交给一个异步线程池处置惩罚。如许，消费者就可以更快地处置惩罚下一条注册消息，进步团体处置惩罚速度。
  

1. public class AsynchronousProcessingConsumer {
2.     private static final String QUEUE_NAME = "user_registration_queue";
3.     private static final String RABBITMQ_HOST = "localhost";
4.     private static final int RABBITMQ_PORT = 5672;
5.     private static final String RABBITMQ_USERNAME = "guest";
6.     private static final String RABBITMQ_PASSWORD = "guest";
9.     public static void main(String[] args) {
10.         ExecutorService executorService = Executors.newFixedThreadPool(2);
13.         try (ConnectionFactory factory = new ConnectionFactory()) {
14.             factory.setHost(RABBITMQ_HOST);
15.             factory.setPort(RABBITMQ_PORT);
16.             factory.setUsername(RABBITMQ_USERNAME);
17.             factory.setPassword(RABBITMQ_PASSWORD);
20.             try (Connection connection = factory.newConnection();
21.                  Channel channel = connection.createChannel()) {
22.                 channel.queueDeclare(QUEUE_NAME, true, false, false, null);
25.                 channel.basicConsume(QUEUE_NAME, false, new DeliverCallback() {
26.                     @Override
27.                     public void handle(String consumerTag, Delivery delivery) throws IOException {
28.                         String message = new String(delivery.getBody(), "UTF-8");
29.                         System.out.println("Received: " + message);
32.                         // 处理注册消息
33.                         processRegistrationMessage(message, executorService);
36.                         // 手动确认消息
37.                         channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
38.                     }
39.                 }, consumerTag -> {
40.                 });
43.                 // 让程序保持运行
44.                 Thread.sleep(Long.MAX_VALUE);
45.             } catch (InterruptedException | TimeoutException e) {
46.                 e.printStackTrace();
47.             }
48.         } catch (IOException e) {
49.             e.printStackTrace();
50.         } finally {
51.             executorService.shutdown();
52.         }
53.     }
56.     private static void processRegistrationMessage(String message, ExecutorService executorService) {
57.         // 发送注册成功邮件和更新用户统计信息在主线程
58.         sendRegistrationEmail(message);
59.         updateUserStats(message);
62.         // 异步处理记录注册日志
63.         executorService.execute(() -> logRegistration(message));
64.     }
67.     private static void sendRegistrationEmail(String message) {
68.         // 实际的发送邮件逻辑，这里仅作示例
69.         System.out.println("Sending registration email for: " + message);
70.     }
73.     private static void updateUserStats(String message) {
74.         // 实际的更新用户统计信息逻辑，这里仅作示例
75.         System.out.println("Updating user stats for: " + message);
76.     }
79.     private static void logRegistration(String message) {
80.         // 实际的日志记录逻辑，这里仅作示例
81.         System.out.println("Logging registration for: " + message);
82.     }
83. }

复制代码

• 优化业务逻辑复杂度
  

假设在一个商品推荐体系中，消费者从 MQ 中获取用户的欣赏历史消息，然后根据这些消息计算推荐商品。原始的计算逻辑可能是先查询数据库获取用户的全部历史欣赏商品类别，再对每个类别进行复杂的统计分析，最后根据分析结果查询商品库获取推荐商品。优化时，可以先对数据进行预处置惩罚，例如在用户欣赏商品时就将欣赏数据进行分类统计并存储在缓存中，消费者只需从缓存中获取统计结果并进行简单计算即可得到推荐商品，大大简化了业务逻辑，进步了处置惩罚速度。

1. import com.rabbitmq.client.*;
2. import java.io.IOException;
3. import java.util.HashMap;
4. import java.util.Map;
7. public class OptimizedBusinessLogicConsumer {
8.     private static final String QUEUE_NAME = "browsing_history_queue";
9.     private static final String RABBITMQ_HOST = "localhost";
10.     private static final int RABBITMQ_PORT = 5672;
11.     private static final String RABBITMQ_USERNAME = "guest";
12.     private static final String RABBITMQ_PASSWORD = "guest";
13.     private static final Map<String, Map<String, Integer>> userBrowsingStatsCache = new HashMap<>();
16.     public static void main(String[] args) {
17.         try (ConnectionFactory factory = new ConnectionFactory()) {
18.             factory.setHost(RABBITMQ_HOST);
19.             factory.setPort(RABBITMQ_PORT);
20.             factory.setUsername(RABBITMQ_USERNAME);
21.             factory.setPassword(RABBITMQ_PASSWORD);
24.             try (Connection connection = factory.newConnection();
25.                  Channel channel = connection.createChannel()) {
26.                 channel.queueDeclare(QUEUE_NAME, true, false, false, null);
29.                 channel.basicConsume(QUEUE_NAME, true, new DeliverCallback() {
30.                     @Override
31.                     public void handle(String consumerTag, Delivery delivery) throws IOException {
32.                         String message = new String(delivery.getBody(), "UTF-8");
33.                         System.out.println("Received: " + message);
36.                         // 处理浏览历史消息
37.                         processBrowsingHistory(message);
40.                         // 计算推荐商品
41.                         recommendProducts(message);
42.                     }
43.                 }, consumerTag -> {
44.                 });
47.                 // 让程序保持运行
48.                 Thread.sleep(Long.MAX_VALUE);
49.             } catch (InterruptedException | TimeoutException e) {
50.                 e.printStackTrace();
51.             }
52.         } catch (IOException e) {
53.             e.printStackTrace();
54.         }
55.     }
58.     private static void processBrowsingHistory(String message) {
59.         // 假设 message 是 JSON 格式，包含用户 ID 和浏览的商品类别
60.         // 这里仅作示例，实际可能需要解析 JSON
61.         String userId = "user123";
62.         String category = "electronics";
65.         // 先更新缓存中的浏览统计信息
66.         userBrowsingStatsCache.computeIfAbsent(userId, k -> new HashMap<>())
67.                .compute(category, (k, v) -> v == null? 1 : v + 1);
70.         // 实际中可以将缓存更新到持久存储，这里暂不展示
71.     }
74.     private static void recommendProducts(String message) {
75.         // 假设 message 是 JSON 格式，包含用户 ID
76.         // 这里仅作示例，实际可能需要解析 JSON
77.         String userId = "user123";
80.         // 从缓存中获取用户的浏览统计信息
81.         Map<String, Integer> userStats = userBrowsingStatsCache.get(userId);
84.         if (userStats!= null) {
85.             // 根据缓存中的统计信息进行简单推荐
86.             System.out.println("Recommending products based on user stats in cache for user: " + userId);
87.         } else {
88.             System.out.println("No browsing history found for user: " + userId);
89.         }
90.     }
91. }

复制代码

调整消息队列参数

• 增加队列容量和消息预取数量
  -对于 RabbitMQ，如果当前队列容量设置为只能容纳 10 万条消息，而消息堆积量达到 100 万条，就必要调整队列容量。例如，将队列的最大长度参数（x-max-length）设置为 200 万或更大，以防止消息因队列满而被扬弃。同时，调整消费者的消息预取数量（basicQos）。
  

代码示例：

1. import com.rabbitmq.client.*;
2. import java.io.IOException;
3. import java.util.concurrent.TimeoutException;
6. public class RabbitMQConsumerAdjustments {
7.     private static final String QUEUE_NAME = "promotion_queue";
8.     private static final String RABBITMQ_HOST = "localhost";
9.     private static final int RABBITMQ_PORT = 5672;
10.     private static final String RABBITMQ_USERNAME = "guest";
11.     private static final String RABBITMQ_PASSWORD = "guest";
14.     public static void main(String[] args) {
15.         try (ConnectionFactory factory = new ConnectionFactory()) {
16.             factory.setHost(RABBITMQ_HOST);
17.             factory.setPort(RABBITMQ_PORT);
18.             factory.setUsername(RABBITMQ_USERNAME);
19.             factory.setPassword(RABBITMQ_PASSWORD);
22.             try (Connection connection = factory.newConnection();
23.                  Channel channel = connection.createChannel()) {
25.                 // 声明队列，并调整队列的最大长度（x-max-length）为 200 万
26.                 Map<String, Object> argsMap = new HashMap<>();
27.                 argsMap.put("x-max-length", 2000000);
28.                 channel.queueDeclare(QUEUE_NAME, true, false, false, argsMap);
31.                 // 调整消费者的消息预取数量为 100
32.                 channel.basicQos(100);
35.                 channel.basicConsume(QUEUE_NAME, false, new DeliverCallback() {
36.                     @Override
37.                     public void handle(String consumerTag, Delivery delivery) throws IOException {
38.                         String message = new String(delivery.getBody(), "UTF-8");
39.                         System.out.println("Received: " + message);
42.                         // 这里添加处理消息的逻辑，例如打印消息内容
43.                         try {
44.                             Thread.sleep(100); // 模拟消息处理时间
45.                         } catch (InterruptedException e) {
46.                             e.printStackTrace();
47.                         }
50.                         // 手动确认消息
51.                         channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
52.                     }
53.                 }, consumerTag -> {
54.                 });
57.                 // 等待一段时间，让消费者持续消费消息
58.                 Thread.sleep(60000);
59.             } catch (InterruptedException e) {
60.                 e.printStackTrace();
61.             }
62.         } catch (IOException | TimeoutException e) {
63.             e.printStackTrace();
64.         }
65.     }
66. }

复制代码

• 调整消息过期时间和死信队列设置
  假设在一个限时促销活动的消息处置惩罚场景中，MQ 中有许多促销活动相干的消息，如优惠券发放消息等。本来优惠券发放消息的过期时间设置为 1 小时，由于消息堆积导致许多消息还未被处置惩罚就过期了。可以将过期时间调整为 2 小时，给消费者更多时间处置惩罚。同时，设置公道的死信队列。例如，当消费者尝试处置惩罚一条优惠券发放消息 5 次都失败（可能是因为接收用户不存在等原因），就将该消息转移到死信队列。如许可以制止这些无法处置惩罚的消息不停堆积在原队列中影响正常消息的处置惩罚，而且可以对死信队列中的消息进行后续分析，找出处置惩罚失败的原因并进行修复。
  

1. import com.rabbitmq.client.*;
2. import java.io.IOException;
3. import java.util.HashMap;
4. import java.util.Map;
5. import java.util.concurrent.TimeoutException;
8. public class RabbitMQDeadLetterQueue {
9.     private static final String QUEUE_NAME = "promotion_queue";
10.     private static final String DEAD_LETTER_QUEUE_NAME = "dead_letter_queue";
11.     private static final String RABBITMQ_HOST = "localhost";
12.     private static final int RABBITMQ_PORT = 5672;
13.     private static final String RABBITMQ_USERNAME = "guest";
14.     private static final String RABBITMQ_PASSWORD = "guest";
17.     public static void main(String[] args) {
18.         try (ConnectionFactory factory = new ConnectionFactory()) {
19.             factory.setHost(RABBITMQ_HOST);
20.             factory.setPort(RABBITMQ_PORT);
21.             factory.setUsername(RABBITMQ_USERNAME);
22.             factory.setPassword(RABBITMQ_PASSWORD);
25.             try (Connection connection = factory.newConnection();
26.                  Channel channel = connection.createChannel()) {
29.                 // 声明死信队列
30.                 channel.queueDeclare(DEAD_LETTER_QUEUE_NAME, true, false, false, null);
33.                 // 声明主队列，并设置消息过期时间和死信队列参数
34.                 Map<String, Object> argsMap = new HashMap<>();
35.                 argsMap.put("x-message-ttl", 7200000); // 消息过期时间为 2 小时（2 * 60 * 60 * 1000 毫秒）
36.                 argsMap.put("x-dead-letter-exchange", "");
37.                 argsMap.put("x-dead-letter-routing-key", DEAD_LETTER_QUEUE_NAME);
40.                 channel.queueDeclare(QUEUE_NAME, true, false, false, argsMap);
43.                 // 调整消费者的消息预取数量为 10
44.                 channel.basicQos(10);
47.                 channel.basicConsume(QUEUE_NAME, false, new DeliverCallback() {
48.                     int retryCount = 0;
51.                     @Override
52.                     public void handle(String consumerTag, Delivery delivery) throws IOException {
53.                         String message = new String(delivery.getBody(), "UTF-8");
54.                         System.out.println("Received: " + message);
57.                         try {
58.                             // 这里添加处理消息的逻辑，例如打印消息内容
59.                             if (retryCount < 5) {
60.                                 // 模拟消息处理失败，仅在重试次数小于 5 次时重试
61.                                 retryCount++;
62.                                 throw new RuntimeException("Message processing failed");
63.                             } else {
64.                                 System.out.println("Message processed successfully");
65.                                 // 手动确认消息
66.                                 channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
67.                             }
68.                         } catch (RuntimeException e) {
69.                             System.out.println("Message processing failed, retrying...");
70.                             // 不确认消息，让消息重新排队或进入死信队列
71.                         }
72.                     }
73.                 }, consumerTag -> {
74.                 });
77.                 // 等待一段时间，让消费者持续消费消息
78.                 Thread.sleep(60000);
79.             } catch (InterruptedException e) {
80.                 e.printStackTrace();
81.             }
82.         } catch (IOException | TimeoutException e) {
83.             e.printStackTrace();
84.         }
85.     }
86. }

复制代码

数据分流与持久化计谋调整

• 数据分流
  
  
  
  • 在一个大型企业的消息处置惩罚体系中，MQ 中有多种范例的消息，如业务订单消息、体系日志消息、员工通知消息等。当出现 100 万消息堆积时，可以根据消息范例进行分流。例如，将业务订单消息分配给一组高性能的消费者，这些消费者运行在性能较好的服务器上，而且接纳更优化的处置惩罚逻辑（如批量处置惩罚、异步处置惩罚等）；将体系日志消息分配给另一组消费者，由于日志消息对处置惩罚实时性要求相对较低，可以接纳较低优先级的处置惩罚方式；员工通知消息则分配给专门的通知服务消费者进行处置惩罚。如许可以制止差别范例的消息相互干扰，进步团体处置惩罚服从。
    
  
  
• 持久化计谋调整
  
  
  
  • 例如在一个数据备份相对美满的文件处置惩罚体系中，MQ 中有大量的文件处置惩罚使命消息。本来消息接纳磁盘持久化，以确保在任何情况下消息都不会丢失。但在消息大量堆积且处置惩罚速度成为关键题目时，可以考虑将部分可重新天生的文件处置惩罚使命消息的持久化计谋调整为内存持久化（如果消息队列支持）。比如，对于一些可以从源文件重新解析天生处置惩罚使命的消息，接纳内存持久化可以加快消费者获取和处置惩罚消息的速度。不外，必要在体系中增加额外的监控和规复机制，一旦出现体系故障导致内存中的消息丢失，可以从源文件重新天生使命消息并发送到 MQ 中进行处置惩罚，以降低消息丢失的风险。
    
  
  

监控与预警体系优化

• 实时监控消息堆积情况
  
  
  
  • 利用 Prometheus 联合 RabbitMQ 的 Exporter 插件来监控 RabbitMQ 的各项指标。例如，通过配置 Exporter 可以获取 RabbitMQ 队列中的消息数量（rabbitmq_queue_messages指标）、消费者数量（rabbitmq_queue_consumers指标）、消息入队速率（rabbitmq_queue_messages_publish_rate指标）和消息出队速率（rabbitmq_queue_messages_deliver_rate指标）等。在 Grafana 中创建仪表盘，直观地展示这些指标的变化趋势。设置告警规则，当消息堆积数量（即rabbitmq_queue_messages指标）超过 50 万条时，通过邮件、短信或者即时通讯工具（如 Slack）向运维人员发送告警信息，以便及时采取步伐。
    
  
  
• 分析堆积原因
  
  
  
  • 除了监控指标外，还必要对消息堆积的原因进行深入分析。可以通过查看消息队列的日志文件来获取更多信息。例如，在 RabbitMQ 中，日志文件可能记录了生产者发送消息的速率变化、消费者连接断开的时间和原因等。如果发现生产者发送消息的速率忽然大幅增加，可能是因为业务高峰期或者某个上游体系出现故障导致大量消息积存发送；如果是消费者连接断开频繁，可能是消费者所在的服务器资源不足（如内存溢出、网络故障等）导致消费者无法正常工作。通太过析这些原因，针对性地采取办理方案，如在业务高峰期增加资源或者优化上游体系的消息发送逻辑，修复消费者所在服务器的资源题目或网络故障等，从根本上办理消息堆积题目，防止再次发生。
    
  
  

简答总结：
1.增加消费者的数量，多线程异步去消费
2.如果有io操作，批量处置惩罚
3.优化业务逻辑，做一下预处置惩罚，放到缓存种，消费的时候直接去从缓存取
4.增加队列容积跟消息预取数量qos
5.增加消息过期时间，制止因为消费过慢产生消息丢失。
6.将消费失败的消息，进行重试，一定重试次数后还失败的扔到死信队列中
7.数据分流，如果时效性要求没那么高的比如日志的消息，可以交给异步线程池去处置惩罚，要求时效性高的消息，扔给高效的消费者组去消费
8.将磁盘持久化切换为内存持久化（如果队列支持），加快消费者的获取和处置惩罚消息的速度
9.上述文章末尾的监控本领及参数，并在达到一定堆积阈值，触发告警提示，比如邮件，短信等
10.根据消息队列的日志文件进行堆积原因分析

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。