RocketMQ 消费写入 MySQL 问题

[chenfang]

公司需求是把项目中产生的数据,实时入库到 MySQL,这些数据分别属于不同的 table.

现在的方案是把产生的数据通过 tomcat -> RocketMQ 的一个 topic 中,然后启用一个 MQ 消费者组合并消息(如果不合并入库次数太多也会慢),然后 batch 入库,这个方案单表没有任何问题.

如果设计到多个表入库,因为 mq 消费是顺序读取 topic 中的消息,也就导致如果一个表需要入库的数据量大,那么入库时间就会长,这会导致整体消费变慢,从而导致实时表不实时的问题.

这就是我现在遇到的问题

1. 入库 MySQL 的时间长,解决时间长的问题(这个我目前想不到解决办法)
2. 如果解决不了入库时间长的问题,那么就让他不影响其他表入库

我自己想过解决方案,针对问题 2 的

方案 1

从一个消费者组 改为 按表名创建消费者组,但是可用性不高

按表名创建消费者组,使用过一段时间,因为表比较多,然后在一个 jvm(如果多个程序分别启动不同的消费者组也不好管理)中启动那么多消费者组,首先内存会占用很多(应该是每个消费者组都会缓存一定数量的消息),也就是内存的大小跟实时表的数量是等比上涨的.其次我查了一下说是消费者组数量也不是可以无限加的,目前我们是 110 多张表,也比较多了.考虑到之后还会加表,这个方案试行了一段时间就废掉了.

或者这个方案有没有优化空间?

方案 2

把入库时间长的表单独分一个消费者组,可用性比第一个还低

经过实践,不晓得哪个时间点,表里数据生产就会增多,所以这样也会导致可能突然就延迟了....很不可控

然后就是大家有什么思路么?万分感谢!

[burymme11]

要实时的去写入 mysql ？？？我猜下是不是有其他地方要实时的去读？
如果是这样就有别的曲线救国方案。

[burymme11]

在 mq 和 mysql 之间加一层

[lopssh]

没看懂需求，要实时的数据，为什么还走 MQ 呢？

[DinnyXu]

@burymme11 我觉得你说的对，如果不实时读，干嘛要实时写，实时写一般是为了应付统计之类的，50-80w 的数据要进行实时写，对 MySQL 来说有很大的压力。

如果要实时读，可以根据读的业务来缓存数据。

[DinnyXu]

@lopssh MQ 就是一个中转层，数据没法第一时间入库，只能通过 MQ 中转

[wqhui]

问题是多表的变更都在一个 topic 上串行处理，不同表的数据也要串行吗？如果只是单表串行，是不是可以分开消费线程跟工作线程，每个表开一个工作线程，消费线程直接把接收到的消息扔给对应表的工作线程写库。不过有个风险就是某一个表的数据特别多，对应工作线程处理不过来缓存被挤爆了，而且因为消费线程接收完就 ack ，实际接收到的消息还在工作线程排队写入，会丢数据。另外你的 mysql 扛的住吗，扛不住就分库

[zcxey2911]

和 mq 就没关系啊，OP 的问题瓶颈是 Mysql ，说白了就是数据量大，写库慢啊

实时写入 Mysql 数据应该采用 Binlog Load 机制实现。Binlog Load 提供了一种使 Doris 增量同步用户对 Mysql 数据库的数据更新操作的 CDC(Change Data Capture)功能

[raysonlu]

尝试解读一下 OP 的需求。不考虑 mysql 写入性能，从 OP 视角来说，项目的问题通过 MQ 对数据库写入进行 batch 入库处理就能解决了，毕竟 OP 看到现在项目是“单表单队列进行 batch 入库”起到优化作用。这种情况下，如果是多表写入想 batch ，能有什么好的方案？
队列的设计我比较喜欢用现实场景类比：景区入口可以设有 1 ～ 3 个入口队伍。站在门口的保安发现有旅行团来，就临时增设“旅行团通道”；如果了解到 xx 旅行团来了一大批人，临时增设“xx 旅行团专属通道”；如果收到通知景区与 xx 旅行团签订了长期合作协议，应该就可以长期保留一个“xx 旅行团专属 vip 通道”。