MQ消息发送性能优化

[agapple]

目前canal原生支持了kafka/rocketmq/rabbitmq的消息发送，目前收到比较多的用户反馈发送MQ消息，存在部分的性能瓶颈，近期会更新一下相关的优化思路供大家参考

ps. pulsar官方也采用了canal作为默认的connector

[agapple]

初步的性能测试 (发送kafka)

--
机器环境：2台24core的物理机，1台部署kafka，1台部署canal (额外部署了kafka-manager/promethues来做监控)

目前测试只对一个目标表做单topic写入，batchSize 50 + timeout 100

构造的数据场景：(混合2张表的insert、update、delete)，包含10个bantch的insert，单条update和随机单条delete

场景	1个topic + 单分区	1个topic+3分区	2个topic+1分区	2个topic+3分区
不开启flatMessage	20k tps (7.50k rps)	15k tps (5.7k rps)	11.5k tps (4.56k rps)	10.8k tps (4.01k rps)
开启flatMessage	0.6k tps (0.2k rps)	0.7k tps (0.2k rps)	0.8k tps (0.2k rps)	0.9k tps (0.2k rps)

优化的性能测试 (2019-09-30)

混合的DML场景测试：(canal server的cpu消耗大概在3个core)

场景	1个topic + 单分区	1个topic+3分区	2个topic+1分区	2个topic+3分区
不开启flatMessage	29.6k tps (9.71k rps)	17.54k tps (6.53k rps)	21.6k tps (7.9k rps)	16.8k tps (5.71k rps)
开启flatMessage	11.79k tps (4.36k rps)	15.97 tps (5.94k rps)	11.91k tps (4.45k rps)	16.96k tps (6.26k rps)

单表的batch insert场景测试: (canal server的cpu消耗大概在4个core)

场景	1个topic + 单分区	1个topic+3分区
不开启flatMessage	59.6k tps	45.1k tps
开启flatMessage	51.3k tps	49.6k tps

优化思路：发送MQ消息存在顺序性考虑，会有串行的代码路径，需要尽可能减少串行路径上的代价，比如序列化、分区散列、flatMessage对象转化等

优化的内容：

1. 针对不开启flatMessage，默认设置canal.instance.memory.rawEntry=true，确保在发送kafka消息阶段不需要做bytes序列化。针对单队列的情况，可以有50%的性能提升
2. 针对开启flatMessage，引入多线程进行protobuf对象的转化到flat对象的过程，减少串行路径，提升会比较明显，可以有近10倍
3. 针对多topic的发送，引入多线程发送机制，减少串行路径，随着topic数量分布可以有倍数上的提升
4. 减少kafka发送端的flush频率，一个getBatchSize在最后进行一次flush阻塞 (之前在flatMessage上每条消息都做了flush)

几个小tips:

1. 单topic单分区 > 多topic单分区，估计是大batch拆分为了小batch，同时有了proto对象的反序列化开销
2. flatMessage模式，性能普遍比不开启flatMessage模式慢，主要就是proto对象的反序列化开销和小batch的发送成本，后续可以通过更多的topic+分区追回

优化小结：

1. 混合DML场景，同步效率在15k ~ 30k tps (每秒的事务数大概在5k~10k的rps，一个事务会包含至少3条binlog)
2. batch操作场景，同步效率在50k ~ 60k tps

---

影响性能的几个参数：

1. canal.instance.memory.rawEntry = true (表示是否需要提前做序列化，非flatMessage场景需要设置为true)
2. canal.mq.flatMessage = false (false代表二进制协议，true代表使用json格式，二进制协议有更好的性能)
3. canal.mq.dynamicTopic (动态topic配置定义，可以针对不同表设置不同的topic，在flatMessage摸是下可以提升并行效率)
4. canal.mq.partitionsNum/canal.mq.partitionHash (分区配置，对写入性能有反作用，不过可以提升消费端的吞吐)

[724686158]

flatMessage支持一下包含GTID好不好。

[724686158]

现在如果ringbuffer设置太大，消息生产太快，kafkaproducer会OOM。

[rewerma]

之前有issue提到多个topic下producer是串行执行的，可以改成并行执行

[changwenwen]

目前写kafka是单producer，写入没毛病但是consumer在poll的时候就非常少，每次1条1条的。
改为kafka多producer写入就OK了

[agapple]

rocketmq 优化测试

混合的DML场景测试：(canal server的cpu消耗大概在3个core)

场景	1个topic + 单分区	1个topic+3分区	2个topic+1分区	2个topic+3分区
不开启flatMessage	29.6k tps (10.71k rps)	23.3k tps (8.59k rps)	26.7k tps (9.46k rps)	21.7k tps (7.66k rps)
开启flatMessage	16.75k tps (6.17k rps)	14.96k tps (5.55k rps)	17.83k tps (6.63k rps)	16.93k tps (6.26k rps)

单表的batch insert场景测试: (canal server的cpu消耗大概在3个core)

场景	1个topic + 单分区	1个topic+3分区
不开启flatMessage	81.2k tps	51.3k tps
开启flatMessage	62.6k tps	57.9k tps

主要优化点：

1. 复用kafka的并行解析protobuf
2. 针对flatMessage，利用batch发送的能力提升吞吐，优化之前为1k，接近有10倍的性能提升

整体的测试来看，canal对接rocketmq的吞吐性能要优于kafka

[rewerma]

同样是开启flatMessage为什么kafka和rocketmq差了一倍多？

[agapple]

@rewerma 应该是后面优化的代码，我重新测试对齐了，差距没那么大，但rocketmq还是有一些优势.
我的判断，rocketmq的性能优势主要在于他是一个可控的batch和并行机制，有直接的batch send接口 以及针对指定分区的并行处理

1. rocketmq相比于kafka，大概有30%的性能提升
2. 非flatMessage模式相比于flatMessage模式，大概有30%~60%的性能提升 (调整了默认参数为flatMessage=false)

[agapple]

文档可参见：https://github.com/alibaba/canal/wiki/Canal-MQ-Performance

[luxuebao]

之前有issue提到多个topic下producer是串行执行的，可以改成并行执行

请问如何将Producer改成并行的