消息队列线程池模型如何保证重启时消息不丢

本文转载自微信公众号「咖啡拿铁」，作者咖啡拿铁 。转载本文请联系咖啡拿铁公众号。

创新互联专注于兴隆台网站建设服务及定制，我们拥有丰富的企业做网站经验。 热诚为您提供兴隆台营销型网站建设，兴隆台网站制作、兴隆台网页设计、兴隆台网站官网定制、微信小程序开发服务，打造兴隆台网络公司原创品牌,更为您提供兴隆台网站排名全网营销落地服务。

背景

今天在脉脉上面看到了一个帖子，比较有意思：

这个帖子的意思是：在使用Kafka的时候，我们已经设置了多个分区，如何去提升消费能力?如果使用线程池的方式去提升如何保证重启时消息不丢。

这个题其实问了两个点，第一个是如何提升消费能力，第二个是如果选择线程池，我们如何做到消息不丢。

这里先解释一下这两个问题到底是怎么回事，在很多消息队列中都有一个概念叫partion，代表着分区，分区是我们提高消息队列消费的关键，我们的消费者消费的渠道就是从每个分区中来的，一个分区只能被一个消费者持有，如下图所示：

有点类似银行排队，队列的个数越多，排队的时间相对来说就会越少，当然也可以通过异步的方式去处理，比如线程池，把所有的消息都扔到线程池中去执行，这就引出了作者说的第二个问题，首先我们来看看同步消费为什么不会丢消息呢?

如果我们使用的是同步模型，当我们消费了之后会将offset ack回去，如果我们出现了重启，没有成功offset，那么这部分数据将会再次消费，如果是用线程池进行消费，那么我们如何进行ack呢，比如我们用线程池消费了 10，11，12 三条消息如果12先消费完，那么我们ack 13吗?如果这样做的话，这个时候重启，kafka就会认为你已经处理了10，11的消息，这个时候消息就会出现丢失，而发这个帖子的同学就是对于这一块是比较疑惑。

网友的回答

我们来看看网友的一些回答：

网友A:

这名网友的回答本质还是使用线程池，作者也回复了，并没有解决线程池的问题。

网友B:

这个方法类似银行排队，只要队列多，那么处理速度就会加快，的确是第一个问题的解决办法之一。

网友C:

这一类主要解决了第二个问题，通过外部维护offset,比如通过offset入库的方式，我们就能找到正确的应该消费的offset，这个相对来说比较复杂，使用一个MQ还得配套一个数据库，万一我使用MQ的服务根本都没有数据库，还得单独去申请。

网友D:

还有另外一种观点就是，代码写好一点，让消费的速度提高，那消费能力自然就上去了，这个的确是一个很重要的点，通常被其他人给忽略，有时候消费比较慢，很多人可能一上来就是考虑中间件应该怎么设置，往往会忽略自己的代码。

看了这么多帖子的一个回复，感觉没有真正能让我满意的答案，下面来说说我心中的一些思路。

我的想法

对于第一个问题的话，如何提升消费能力?这个问题其实可以总结为三个办法：

  1. 如果每台消费者机器消费线程是固定的，那么我们可以扩容消费机器和partion,类似银行排队增加排队窗口一样。
  2. 如果机器和partion是固定的，增加消费线程就是一个比较好的办法，但是如果是顺序消费，就不能通过增加线程数的方式来提升消费能力，因为顺序消费每个partion都是一个单独的线程，只能通过第一种方式去解决。
  3. 增加自身代码的消费能力，你想想如果银行办事，如果柜员的办事效率能提升的非常高，那么整个排队速度肯定也是很快的。

对于第二个问题，如果我们使用线程池模型,如何去解决消息丢失问题，这里我比较推荐的是RocketMQ中的做法，我们之前说了用数据库去保存offset比较复杂，性能还比较差，在RocketMQ中使用了一个TreeMap的结构做了我们上面提到的数据库的事:

 
 
 
 
    
  
  
  
  
  1. private final TreeMap msgTreeMap = new TreeMap(); 
    2.

这个TreeMap的key是每个message的offset，value就是这条消息的一些信息，TreeMap的底层是使用红黑树去实现的，我们可以很快获取其中的最小值和最大值，当我们每次处理完某一条消息的时候我们会将这条消息从msgTreeMap中移除，

 
 
 
 
    
  
  
  
  
  1. public long removeMessage(final List msgs) { 
    2.   
  
  
  
  2.         long result = -1; 
    3.   
  
  
  
  3.         final long now = System.currentTimeMillis(); 
    4.   
  
  
  
  4.         try { 
    5.   
  
  
  
  5.             this.lockTreeMap.writeLock().lockInterruptibly(); 
    6.   
  
  
  
  6.             this.lastConsumeTimestamp = now; 
    7.   
  
  
  
  7.             try { 
    8.   
  
  
  
  8.                 if (!msgTreeMap.isEmpty()) { 
    9.   
  
  
  
  9.                     result = this.queueOffsetMax + 1; 
    10.   
  
  
  
  10.                     int removedCnt = 0; 
    11.   
  
  
  
  11.                     for (MessageExt msg : msgs) { 
    12.   
  
  
  
  12.                         MessageExt prev = msgTreeMap.remove(msg.getQueueOffset()); 
    13.   
  
  
  
  13.                         if (prev != null) { 
    14.   
  
  
  
  14.                             removedCnt--; 
    15.   
  
  
  
  15.                             msgSize.addAndGet(0 - msg.getBody().length); 
    16.   
  
  
  
  16.                         } 
    17.   
  
  
  
  17.                     } 
    18.   
  
  
  
  18.                     msgCount.addAndGet(removedCnt); 
    19.   
  
  
  
  19.  
    20.   
  
  
  
  20.                     if (!msgTreeMap.isEmpty()) { 
    21.   
  
  
  
  21.                         result = msgTreeMap.firstKey(); 
    22.   
  
  
  
  22.                     } 
    23.   
  
  
  
  23.                 } 
    24.   
  
  
  
  24.             } finally { 
    25.   
  
  
  
  25.                 this.lockTreeMap.writeLock().unlock(); 
    26.   
  
  
  
  26.             } 
    27.   
  
  
  
  27.         } catch (Throwable t) { 
    28.   
  
  
  
  28.             log.error("removeMessage exception", t); 
    29.   
  
  
  
  29.         } 
    30.   
  
  
  
  30.         return result; 
    31.   
  
  
  
  31.     } 
    32.

removeMessage这个方法就是移除已经消费过的消息，并且返回当前最新的消费offset，这里返回的结果就是msgTreeMap.firstKey()，我们ack给消息队列server的值其实也是这个，回到我们这个问题上，如果我们发生重启，那么其实也不需要担心我们会出现消息丢失。

文章标题：消息队列线程池模型如何保证重启时消息不丢
文章起源：http://www.shufengxianlan.com/qtweb/news7/537807.html

网站建设、网络推广公司-创新互联，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等

广告

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 创新互联

猜你还喜欢下面的内容

虚拟主机知识

各行业网站

微信小程序开发　　 　VI设计　　 　响应式报价　　 　成都网站改版　　 　外贸网站设计方案　　 　成都网站推广　　 　网站解决方案　　 　专业网站建设　　 　名片印刷　　 　微信开发　　 　腾讯香港虚拟主机　　 　定制网站设计　　 　新津网站运维　　 　重庆网站制作　　 　界面设计　　 　成都包装盒设计　　 　德阳做网站　　 　成都二枢电信服务器托管　　 　高防服务器租用　　 　云南护栏打桩机