Kafka消息积压案例分析

悠悠 2022-09-03 00:24 46阅读 0赞

## 案例 ##

一个微服务同一个分组消费同一个topic的kafka消息，不通业务通过key值区分，由于其中一个业务消息量大，偶尔会出现消费滞后的情况，导致当前微服务消费组出现大量消息积压情况，影响业务。

简单的说，即一个单线程消费，一个消息处理完毕，才会处理下一个。

Created with Raphaël 2.3.0开始消费者组消息处理消费成功？yes

## 分析 ##

存在消息积压的场景：消息消费

@Override
        @Consumer(topic = Const.KAFKA_TOPIC, key = Const.TOPIC_KEY)
        public boolean handle(String msg) throws Exception { 
            OriginInfo msg = JSON.parseObject(msg, OriginInfo.class);
            return service.process(msg);
        }

其中， service.process(msg)方法流程比较复杂，存在多个查询和入库、更新操作，有时候由于数据库性能，这个过程会变慢，导致整体消息消费跟不上。

## 建议 ##

持久化方案：消息先入库，再消费，去掉中间处理流程。

Created with Raphaël 2.3.0开始消费者组消息入库消费成功？yes

消息入库：

消息处理：可以构建一个单线程池，定时创建一个新线程丢到线程池，已确保消息顺序消费处理。

/** * kafka消息异步处理 * @author loongshawn */
    public class MsgService implements Runnable { 
    
        private static final Logger logger = LoggerFactory.getLogger(MsgService.class);
    
        private AService aService;
    
        public MsgService() { 
            this.aService = (AService) SpringContextUtil.getBean("aService");
        }
    
        /** * 消息处理入口 */
        void process() { 
            List<OriginInfo> list = new ArrayList<>();
            try { 
                // 依据入库顺序读出消息记录，顺序消费
                list = aService.get();
                if (CollectionUtils.isEmpty(list)) { 
                    return;
                }
            } catch (Exception e) { 
                logger.error("exception {}", e.toString());
            }
    
            for (OriginInfo item : list) { 
                try { 
                    // 顺序消费
                    aService.process(item);
                    // 消费完毕，更新消息标记
                    aService.update(item);
                } catch (Exception e) { 
                    logger.error("exception {}", e.toString());
                }
            }
        }
    
        @Override
        public void run() { 
            try { 
                this.process();
            } catch (Exception e) { 
                logger.error("exception {}", e.toString());
            }
        }
    }

## 优化 ##

由于此样例中呈现的是消息写入库，比如MySQL、Oracle等等，消息量几十万还能应对，但如果应对大规模比如数千万消息要写入库，上述方案就不行了。需要考虑引入缓存方式，降低数据库IO成本。

Created with Raphaël 2.3.0开始消费者组消息缓存消费成功？yes

三步走方案：

*  kafka消息---->写入redis缓存---->消费结束
 *  读redis缓存---->持久化至数据库
 *  读数据库---->处理消息