消息队列——数十万级消息的消费方案

配景：

​        卑鄙平台通过消息队列上报监控消息，但是消息量很大，在三分钟左右可以到达百万级别，而对于我的服务来说，我需要对这些消息进行一些业务处理，然后再存入es中。（为了简化场景，以下对于消息的处理只是单纯的存储到es中）

服务启动不到10s，es中写入的数据
青铜方案：

​        MQ只要收到消息，就直接调用es进行存储。
伪代码如下：

1. // 伪代码版本
2. public void processRequestMessage(MessageInfo info) {
3.     // 将接收到的信息对象复制为一个新对象（例如，监控数据对象）
4.      MonitorData monitorData = Util.copyProperties(info, MonitorData.class);
5.     // 将新对象的 JSON 字符串索引到 Elasticsearch 中
6.      elasticClient.index("monitor_index", info.getId(), convertToJson(monitorData), false);
7. }

复制代码

存在的问题：

​        不难发现，如许的实现方式，会导致消息消费速度非常慢，甚至导致消息积压和服务挂掉，因为这里对es的调用次数=消息条数，通过在本地的测试中也可以发现，纵然在关闭掉消息生产者后，还是需要很长一段时间才能将消息消费完全消费掉。
白银方案：

​        通过瓶颈，可以很自然的想到使用es的批量增加，那么只需要实现一个缓冲池，将消息暂存到缓冲池中，在到达一定大小的时候再同一在es存储
伪代码如下：

1. @Component
2. @Slf4j
3. public class ESOperationMonitorBuffer{
5.     private static final int BUFFER_SIZE = 100;  // 缓冲池大小
6.     private List<ElasticDoc> buffer;  // 用于存储消息的缓冲池
8.     @Autowired
9.     private ElasticClient elasticClient;
10.     private String indexName = EsConstans.NODE_MONITOR_INDEX;  // Elasticsearch 索引名称
12.     public ESOperationMonitorBuffer() {
13.         this.buffer = new ArrayList<>();
14.     }
16.     // 添加消息到缓冲池
17.     public void addMessage(WlwMessageShareInfo message)  {
18.         ElasticDoc elasticDoc = new ElasticDoc();
19.         elasticDoc.setIndex(indexName);
20.         MonitorESData monitorESData = BeanUtil.copyProperties(message, MonitorESData.class);
21.         elasticDoc.setDoc(JSONObject.toJSONString(monitorESData));
22.         buffer.add(elasticDoc);
23.         if(buffer.size() > BUFFER_SIZE){
24.                 flush();
25.         }
26.     }
28.     /**
29.      * 执行 flush 操作
30.      */
31.     private void flush()  {
32.        log.info("开始批量插入 Elasticsearch，共 {} 条数据", buffer.size());
33.             if (buffer.isEmpty()) {
34.                 return;  // 如果缓冲池为空，不执行操作
35.             }
36.             BulkResponse index = elasticClient.index(buffer, false);// 批量插入
37.             if (index.hasFailures()) {
38.                 log.error("批量插入 Elasticsearch 失败，失败原因：{}", index.buildFailureMessage());
39.             }
40.             // 清空缓冲池
41.             buffer.clear();
42.     }
44.     // 如果程序关闭前有剩余数据，执行 flush 操作
45.     public void close() throws IOException {
46.     }
48. }

复制代码

存在的问题：

• 若消息数目一直没到达阈值，就一直不会保存到es
  
• 存在并发问题，ConcurrentModificationException（并发修改非常），是基于java聚集中的 快速失败（fail-fast） 机制产生的，在使用迭代器遍历一个聚集对象时，假如遍历过程中对聚集对象的内容进行了增编削，就会抛出该非常。快速失败机制使得java的聚集类不能在多线程下并发修改，也不能在迭代过程中被修改。在上面场景中的体现就是在flush操作中时，又有消息进入到了buffer中。
  

黄金方案：

​        解决问题1：可以开启一个定时任务去执行flush方法
​        解决问题2：大概大家第一时间会想到对buffer加锁，但是如许又会导致在存入buffer的时候速度慢，所以不难想到可以对 flush() 方法加锁， 但是如许一来还是无法解决buffer存在的并发问题，怎么办呢？其实很简单，我们可以用两个buffer来分别给add()方法和flush()方法使用，如许一来，就可以制止并发问题，并且继续对flush()方法加锁，制止和定时任务同时执行，导致数据重复。
伪代码如下：

1. @Component
2. @Slf4j
3. public class ESOperationMonitorBuffer implements CommandLineRunner {
5.     private static final int BUFFER_SIZE = 100;  // 缓冲池大小
6.     private List<ElasticDoc> buffer;  // 用于存储消息的缓冲池
7.     private List<ElasticDoc> temBuffer;  // 用于存储临时消息的缓冲池
9.     @Autowired
10.     private ElasticClient elasticClient;
11.     private String indexName = EsConstans.NODE_MONITOR_INDEX;  // Elasticsearch 索引名称
12.     private Lock lock = new ReentrantLock();
13.    
14.     public ESOperationMonitorBuffer() {
15.         this.buffer = new ArrayList<>();
16.         this.temBuffer = new ArrayList<>();
17.     }
19.     // 添加消息到缓冲池
20.     public void addMessage(WlwMessageShareInfo message)  {
21.         ElasticDoc elasticDoc = new ElasticDoc();
22.         elasticDoc.setIndex(indexName);
23.         MonitorESData monitorESData = BeanUtil.copyProperties(message, MonitorESData.class);
24.         elasticDoc.setDoc(JSONObject.toJSONString(monitorESData));
25.         temBuffer.add(elasticDoc);
27.         // 当缓冲池达到设定大小时，批量插入到 Elasticsearch
28.         if (temBuffer.size() >= BUFFER_SIZE) {
29.             lock.lock();
30.             try{
31.                 buffer.addAll(temBuffer);
32.                 temBuffer.clear();
33.             }catch(Exception e){
34.                 log.error("添加消息到缓冲池失败",e);
35.             }finally {
36.                 lock.unlock();
37.             }
38.             flush();
39.         }
40.     }
42.     /**
43.      * 执行 flush 操作
44.      */
45.     private void flush()  {
46.         lock.lock();
47.         try{
48.             log.info("开始批量插入 Elasticsearch，共 {} 条数据", buffer.size());
49.             if (buffer.isEmpty()) {
50.                 return;  // 如果缓冲池为空，不执行操作
51.             }
52.             BulkResponse index = elasticClient.index(buffer, false);// 批量插入
53.             if (index.hasFailures()) {
54.                 log.error("批量插入 Elasticsearch 失败，失败原因：{}", index.buildFailureMessage());
55.             }
56.             // 清空缓冲池
57.             buffer.clear();
58.         }catch (Exception e){
59.             log.info("批量插入 Elasticsearch 失败",e);
60.         }finally {
61.             lock.unlock();
62.         }
63.     }
65.     // 如果程序关闭前有剩余数据，执行 flush 操作
66.     public void close() throws IOException {
67.     }
69.     @Override
70.     public void run(String... args) throws Exception {
71.         log.info("启动 ESOperationMonitorBuffer 缓冲池，开启线程池定时执行flush操作");
72.         // 定时执行 flush 操作
73.         ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
74.         scheduledExecutorService.scheduleAtFixedRate(()->{
75.             try{
76.                 flush();
77.             }catch (Exception e){
78.                 log.error("定时执行 flush 操作失败",e);
79.             }
80.         },1,5, TimeUnit.SECONDS);
81.     }
82. }

复制代码

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