一、背景

在上一篇文章中，咱们应用 Canal Admin 搭建了Canal Server 集群，在这篇文章中，咱们应用上篇文章的根底，将音讯发送到kafka音讯队列中。

二、须要批改的中央

以下 配置文件的批改，都是在 Canal Admin 上批改的。

1、canal.properties 配置文件批改

1、批改canal.serverMode的值

2、批改kafka配置

2、批改 instance.propertios 配置文件

3、canal发消息到mq性能优化

影响性能的几个参数：

  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 (分区配置，对写入性能有副作用，不过能够晋升生产端的吞吐)

参考链接：https://github.com/alibaba/canal/wiki/Canal-MQ-Performance

三、kafka接管音讯

1、canal 发送过去的音讯

/** * canal 发送过去的音讯 * * @author huan.fu 2021/9/2 - 下午4:06 */@Getter@Setter@ToStringpublic class CanalMessage {    /**     * 测试得出 同一个事物下产生多个批改，这个id的值是一样的。     */    private Integer id;    /**     * 数据库或schema     */    private String database;    /**     * 表名     */    private String table;    /**     * 主键字段名     */    private List<String> pkNames;    /**     * 是否是ddl语句     */    private Boolean isDdl;    /**     * 类型:INSERT/UPDATE/DELETE     */    private String type;    /**     * binlog executeTime, 执行耗时     */    private Long es;    /**     * dml build timeStamp, 同步工夫     */    private Long ts;    /**     * 执行的sql,dml sql为空     */    private String sql;    /**     * 数据列表     */    private List<Map<String, Object>> data;    /**     * 旧数据列表,用于update,size和data的size一一对应     */    private List<Map<String, Object>> old;}

2、监听音讯

@Component@Slf4jpublic class KafkaConsumer {    @KafkaListener(topics = "customer", groupId = "canal-kafka-springboot-001", concurrency = "5")    public void consumer(ConsumerRecord<String, String> record, Acknowledgment ack) throws InterruptedException {        log.info(Thread.currentThread().getName() + ":" + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")) + "接管到kafka音讯,partition:" + record.partition() + ",offset:" + record.offset() + "value:" + record.value());        CanalMessage canalMessage = JSON.parseObject(record.value(), CanalMessage.class);        log.info("\r=================================");        log.info("接管到的原始 canal message为: {}", record.value());        log.info("转换成Java对象后转换成Json为 : {}", JSON.toJSONString(canalMessage));        ack.acknowledge();    }}

3、获取音讯

四、MQ配置相干的参数

参数名参数阐明默认值
canal.mq.serverskafka为bootstrap.servers rocketMQ中为nameserver列表127.0.0.1:6667
canal.mq.retries发送失败重试次数0
canal.mq.batchSizekafka为ProducerConfig.BATCH_SIZE_CONFIG rocketMQ无意义16384
canal.mq.maxRequestSizekafka为ProducerConfig.MAX_REQUEST_SIZE_CONFIG rocketMQ无意义1048576
canal.mq.lingerMskafka为ProducerConfig.LINGER_MS_CONFIG , 如果是flatMessage格局倡议将该值调大, 如: 200 rocketMQ无意义1
canal.mq.bufferMemorykafka为ProducerConfig.BUFFER_MEMORY_CONFIG rocketMQ无意义33554432
canal.mq.ackskafka为ProducerConfig.ACKS_CONFIG rocketMQ无意义all
canal.mq.kafka.kerberos.enablekafka为ProducerConfig.ACKS_CONFIG rocketMQ无意义false
canal.mq.kafka.kerberos.krb5FilePathkafka kerberos认证 rocketMQ无意义../conf/kerberos/krb5.conf
canal.mq.kafka.kerberos.jaasFilePathkafka kerberos认证 rocketMQ无意义../conf/kerberos/jaas.conf
canal.mq.producerGroupkafka无意义 rocketMQ为ProducerGroup名Canal-Producer
canal.mq.accessChannelkafka无意义 rocketMQ为channel模式，如果为aliyun则配置为cloudlocal
canal.mq.vhost=rabbitMQ配置无
canal.mq.exchange=rabbitMQ配置无
canal.mq.username=rabbitMQ配置无
canal.mq.password=rabbitMQ配置无
canal.mq.aliyunuid=rabbitMQ配置无
canal.mq.canalBatchSize获取canal数据的批次大小50
canal.mq.canalGetTimeout获取canal数据的超时工夫100
canal.mq.parallelThreadSizemq数据转换并行处理的并发度8
canal.mq.flatMessage是否为json格局 如果设置为false,对应MQ收到的音讯为protobuf格局 须要通过CanalMessageDeserializer进行解码false
canal.mq.topicmq里的topic名无
canal.mq.dynamicTopicmq里的动静topic规定, 1.1.3版本反对无
canal.mq.partition单队列模式的分区下标，1
canal.mq.partitionsNum散列模式的分区数无
canal.mq.partitionHash散列规定定义 库名.表名 : 惟一主键，比方mytest.person: id 1.1.3版本反对新语法，见下文

参考文档：https://github.com/alibaba/canal/wiki/Canal-Kafka-RocketMQ-QuickStart

五、MQ接管binlog代码

https://gitee.com/huan1993/spring-cloud-parent/tree/master/canal/canal-kafka-consumer

六、参考文章

1、canal 发送binlog到mq中性能测试.
2、canal发送音讯到kafka中