一、背景在上一篇文章中，我们使用CanalAdmin搭建了一个CanalServer集群。在这篇文章中，我们在上一篇文章的基础上，向kafka消息队列发送消息。2、需要修改的地方以下配置文件的修改都是在CanalAdmin上修改的。1.修改canal.properties配置文件1.修改canal.serverMode的值2.修改kafka配置2.修改instance.propertios配置文件3.Canal向mq发送消息性能优化影响性能的几个参数：canal。instance.memory.rawEntry=true（表示是否需要提前序列化，非flatMessage场景需要设置为true）canal.mq.flatMessage=false（false表示二进制协议，true表示使用json格式,二进制协议有更好的性能)canal.mq.dynamicTopic（动态topic配置定义，不同表可设置不同topic，flatMessage方式可提高并行效率）canal.mq.partitionsNum/canal.mq.partitionHash（分区配置，对写入性能有不良影响，但是可以提高消费端的吞吐量）参考链接：https://github.com/alibaba/canal/wiki/Canal-MQ-Performance3.Kafka接收消息1.canal发送的消息/***发送的消息通过canal**@authorhuan.fu2021/9/2-4:06p.m.*/@Getter@Setter@ToStringpublicclassCanalMessage{/***测试表明在同一个东西下进行多次修改，这个id的值是一样的。*/私有整数id;/***数据库或模式*/privateStringdatabase;/***表名*/privateStringtable;/***主键字段名*/privateListpkNames;/***是不是ddl语句*/privateBooleanisDdl;/***类型：INSERT/UPDATE/DELETE*/privateString类型；/***binlogexecuteTime，执行耗时*/privateLonges;/***dml构建时间戳，同步时间*/privateLongts;/***执行的sql，dmlsql为空*/privateStringsql;/***数据列表*/privateList>data;/***旧数据列表，用于更新，大小和数据大小一一对应*/privateList>old;}2.监听消息@Component@Slf4jpublicclassKafkaConsumer{@KafkaListener(topics="customer",groupId="canal-kafka-springboot-001",concurrency="5")publicvoidconsumer(ConsumerRecordrecord,Acknowledgmentack)抛出中断dException{log.info(Thread.currentThread().getName()+":"+LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-ddHH:mm:ss"))+"收到kafka消息，分区："+record.partition()+",offset:"+record.offset()+"value:"+record.value());CanalMessagecanalMessage=JSON.parseObject(record.value(),CanalMessage.class);log.info("\r===================================");日志。info("收到的原始canal消息为：{}",record.value());log.info("转成Java对象再转成Json:{}",JSON.toJSONString(canalMessage));确认.acknowledge();}}3.获取信息4.MQ配置相关参数参数名称参数说明默认值canal.mq.serverskafka是bootstrap.serversrocketMQ是nameserverlist127.0.0.1:6667canal.mq.retries发送失败重试次数0canal.mq。batchSizekafka为ProducerConfig.BATCH_SIZE_CONFIGrocketMQ无意义16384canal.mq.maxRequestSizekafka为ProducerConfig.MAX_REQUEST_SIZE_CONFIGrocketMQ无意义1048576canal.mq.lingerMskafka为ProducerConfig.LINGER_MS_CONFIG,如果是flatMessage格式建议将该值调大,如:200rocketMQmeaningless1canal.mq.bufferMemorykafka为ProducerConfig.BUFFER_MEMORY_CONFIGrocketMQ无意义33554432canal.mq.ackskafka为ProducerConfig.ACKS_CONFIGrocketMQ无意义allcanal.mq.kafka.kerberos.enablekafka为ProducerConfig.ACKS_CONFIGrocketMQ无意义falsecanal.mq.kafka.kerberos.krb5FilePathkafkakerberos认证rocketMQ无意义../conf/kerberos/krb5.confcanal.mq.kafka.kerberos.jaasFilePathkafkakerberos认证rocketMQ无意义../conf/kerberos/jaas.confcanal.mq.producerGroupkafka无意义rocketMQ为ProducerGroup名称Canal-Producercanal.mq.accessChannelkafka无意义rocketMQ是通道模式，如果是aliyun，则配置为cloudlocalcanal.mq.vhost=rabbitMQconfigurationnocanal.mq.exchange=rabbitMQconfigurationnocanal.mq.username=rabbitMQconfigurationnocanal.mq.password=rabbitMQ配置无canal.mq.aliyunuid=rabbitMQ配置无canal.mq.canalBatchSize获取canal数据批量大小50canal.mq.canalGetTimeout获取canal数据超时时间100canal.mq.parallelThreadSizemq数据转换并行处理并发度8canal.mq是否.flatMessageisinjsonformat如果设置为false，则MQ接收到的消息是protobuf格式，需要通过CanalMessageDeserializer解码falsecanal.mq.topicmq中的主题名在canal.mq.dynamicTopicmq中没有动态pic规则，1.1.3版本支持无canal.mq.partition单队列模式的分区下标，1canal.mq.partitionsNum哈希模式下的分区数无canal.mq.partitionHash哈希规则定义库名。表名：唯一主键，如mytest.person:id1.1.3版本支持新语法，参考文档如下：https://github.com/alibaba/canal/wiki/Canal-Kafka-RocketMQ-QuickStart5.MQ接收binlog代码https://gitee.com/huan1993/spring-cloud-parent/tree/master/canal/canal-kafka-consumer6.参考文章1.Canal将binlog发送给mq进行性能测试。2.Canal向kafka发送消息
