不同业务共用同一 TOPIC 引发的问题

短时间大量请求 + 不同业务使用同一个TOPIC的问题

阿里云RocketMQ消息堆积, 影响仓配系统订货单生成

1.监控工具提醒接口平台消息堆积
2.运营人员反馈仓配系统单据未生成

1.通过观测云查看接口平台应用JVM
2.阿里云查看MQ堆栈信息
3.查看接口平台MQ接收记录表和日志系统

JVM -> MQ堆栈 -> 操作记录

系统架构图

说明: 接口平台自己发送消息自己消费消息
1.接口平台系统接收到OMS系统的HTTP请求之后, 将数据写入 platform_third_receive_record 表, 发送MQ消息, 返回HTTP响应, 接下来会通过MQ异步处理
2.MQ线程将业务数据组装之后通过HTTP请求发送给WMS系统, WMS系统接收到HTTP请求,将数据写入数据库, 也是通过MQ异步处理
3.WMS系统发送HTTP请求,告诉接口平台,第 2 步操作是否成功,接口平台接收到请求之后,发送MQ异步处理
4.WMS系统将第 2 步处理成功之后,发送HTTP请求,告诉接口平台,需要把业务数据推送给第三方, 接口平台接收到请求之后, 发送MQ异步处理

查看接口平台应用JVM, 一切正常

查看MQ堆栈信息, 每隔10秒,查看3次, 未发现耗时操作.

查看接口平台接收记录表

          
SELECT receive_type_name, COUNT(*) AS total
FROM open_platform_center.platform_third_receive_record
WHERE
    create_time >= '2024-11-22 10:44:00'
    AND create_time <= '2024-11-22 10:50:00'
    AND receive_platform_code = 'APP-00002' -- 接收来自WMS系统的HTTP请求
GROUP BY receive_type_name

-- 查询结果
门店/收获人 3706
门店创建完成 3923


SELECT receive_type_name, COUNT(*) AS total
FROM open_platform_center.platform_third_receive_record
WHERE
    create_time >= '2024-11-22 10:44:00'
    AND create_time <= '2024-11-22 10:50:00'
    AND receive_platform_code = 'APP-00006' -- 接收来自OMS系统的HTTP请求
GROUP BY receive_type_name

-- 查询结果
门店/收获人 3973

查看接口平台日志系统 [2024-11-21 10:30:00 2024-11-21 10:50:00] 时间段接口平台接收HTTP请求分布图

查看接口平台日志系统 [2024-11-22 10:30:00 2024-11-22 10:50:00] 时间段接口平台接收HTTP请求分布图

从接口平台的接收记录表和日志系统可知,在短时间内收到大量与门店业务相关的HTTP请求(品仓合作,需要将品牌下1300+家门店一次性同步到接口平台), 6分钟内大概约1万条MQ消息需要消费, 接口平台只有一台机器, 并且所有的业务都使用同一个主题, 造成订货单业务被门店业务影响.
无论是接口平台收到HTTP请求,自身发送MQ消息, 还是说由其他生产者发送的MQ消息, 问题根源在于短时间内有大量MQ消息需要被消费.

1.将基础业务数据(商品,门店等)单独使用一个主题, 单据(订货单,发货单等)使用另一个主题, 主题隔离
2.增加接口平台消费者机器数量

即便消费者订阅多个主题, 一个消费者端只有一个线程池. 该线程池中的线程(ConsumeMessageThread_N)会根据主题选择用户自定义的不同监听器进行处理消息.

          
com.aliyun.openservices.shade.com.alibaba.rocketmq.client.impl.consumer.ConsumeMessageConcurrentlyService

// 虽然是无界队列, 但因为有拉取限流, 因此不会导致OOM
this.consumeRequestQueue = new LinkedBlockingQueue<Runnable>();

this.consumeExecutor = new ThreadPoolExecutor(
    this.defaultMQPushConsumer.getConsumeThreadMin(), // 20
    this.defaultMQPushConsumer.getConsumeThreadMax(), // 64
    1000 * 60,
    TimeUnit.MILLISECONDS,
    this.consumeRequestQueue,
    new ThreadFactoryImpl("ConsumeMessageThread_"));

参考外链

RocketMQ中单消费者订阅多个Topic, 会阻塞消费吗