From 5bde61f97a1f3ed33b6a96ed20b1f8fd38151d45 Mon Sep 17 00:00:00 2001 From: agapple Date: Tue, 17 May 2022 14:08:32 +0800 Subject: [PATCH] Updated Canal Kafka RocketMQ QuickStart (markdown) --- Canal-Kafka-RocketMQ-QuickStart.md | 63 +++++++++++++++++++++++++++++- 1 file changed, 62 insertions(+), 1 deletion(-) diff --git a/Canal-Kafka-RocketMQ-QuickStart.md b/Canal-Kafka-RocketMQ-QuickStart.md index 8e6c613..bb793e0 100644 --- a/Canal-Kafka-RocketMQ-QuickStart.md +++ b/Canal-Kafka-RocketMQ-QuickStart.md @@ -87,7 +87,7 @@ canal.mq.acks = all canal.mq.transaction = false ``` -## mq相关参数说明 +## mq相关参数说明 (<=1.1.4版本) | 参数名 | 参数说明 | 默认值 | |-------|-----------|-------| @@ -121,6 +121,58 @@ canal.mq.transaction = false | canal.mq.partitionsNum | 散列模式的分区数 | 无 | | canal.mq.partitionHash | 散列规则定义
库名.表名 : 唯一主键，比如mytest.person: id
1.1.3版本支持新语法，见下文 | 无 | +## mq相关参数说明 (>=1.1.5版本) + +在1.1.5版本开始，引入了[MQ Connector设计](https://github.com/alibaba/canal/pull/2562)，因此参数配置做了部分调整 + +| 参数名 | 参数说明 | 默认值 | +|------|-----------|-------| +| canal.aliyun.accessKey | 阿里云ak | 空值 | +| canal.aliyun.secretKey | 阿里云sk | 空值 | +| canal.aliyun.uid | 阿里云uid | 空值 | +| canal.mq.flatMessage | 是否为json格式
如果设置为false,对应MQ收到的消息为protobuf格式
需要通过CanalMessageDeserializer进行解码 | false | +| canal.mq.canalBatchSize | 获取canal数据的批次大小 | 50 | +| canal.mq.canalGetTimeout | 获取canal数据的超时时间 | 100 | +| canal.mq.accessChannel = local | 是否为阿里云模式，可选值local/cloud | local | +| canal.mq.database.hash | 是否开启database混淆hash，确保不同库的数据可以均匀分散，如果关闭可以确保只按照业务字段做MQ分区计算 | true | +| canal.mq.send.thread.size | MQ消息发送并行度 | 30 | +| canal.mq.build.thread.size | MQ消息构建并行度 | 8 | +|------|-----------|-------| +| kafka.bootstrap.servers | kafka服务端地址 | 127.0.0.1:9092 | +| kafka.acks| kafka为`ProducerConfig.ACKS_CONFIG` | all | +| kafka.compression.type | 压缩类型 | none | +| kafka.batch.size | kafka为`ProducerConfig.BATCH_SIZE_CONFIG` | 16384 | +| kafka.linger.ms | kafka为`ProducerConfig.LINGER_MS_CONFIG` , 如果是flatMessage格式建议将该值调大, 如: 200 | 1 | +| kafka.max.request.size | kafka为`ProducerConfig.MAX_REQUEST_SIZE_CONFIG` | 1048576 | +| kafka.buffer.memory | kafka为`ProducerConfig.BUFFER_MEMORY_CONFIG` | 33554432 | +| kafka.max.in.flight.requests.per.connection | kafka为`ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION` | 1 | +| kafka.retries | 发送失败重试次数 | 0 | +| kafka.kerberos.enable | kerberos认证 | false | +| kafka.kerberos.krb5.file | kerberos认证 | ../conf/kerberos/krb5.conf | +| kafka.kerberos.jaas.file | kerberos认证 | ../conf/kerberos/jaas.conf | +|------|-----------|-------| +| rocketmq.producer.group | rocketMQ为ProducerGroup名 | test | +| rocketmq.enable.message.trace | 是否开启message trace | false | +| rocketmq.customized.trace.topic | message trace的topic | 空值 | +| rocketmq.namespace | rocketmq的namespace | 空值 | +| rocketmq.namesrv.addr | rocketmq的namesrv地址 | 127.0.0.1:9876 | +| rocketmq.retry.times.when.send.failed | 重试次数 | 0 | +| rocketmq.vip.channel.enabled | rocketmq是否开启vip channel | false | +| rocketmq.tag | rocketmq的tag配置 | 空值 | +| --- | --- | --- | +| canal.mq.vhost= | rabbitMQ配置 | 无 | +| canal.mq.exchange= | rabbitMQ配置 | 无 | +| canal.mq.username= | rabbitMQ配置 | 无 | +| canal.mq.password= | rabbitMQ配置 | 无 | +| canal.mq.aliyunuid= | rabbitMQ配置 | 无 | +| --- | --- | --- | +| canal.mq.topic | mq里的topic名 | 无 | +| canal.mq.dynamicTopic | mq里的动态topic规则, 1.1.3版本支持 | 无 | +| canal.mq.partition | 单队列模式的分区下标， | 1 | +| canal.mq.partitionsNum | 散列模式的分区数 | 无 | +| canal.mq.partitionHash | 散列规则定义
库名.表名 : 唯一主键，比如mytest.person: id
1.1.3版本支持新语法，见下文 | 无 | + + ### canal.mq.dynamicTopic 表达式说明 canal 1.1.3版本之后, 支持配置格式：schema 或 schema.table，多个配置之间使用逗号或分号分隔 @@ -184,6 +236,15 @@ canal.mq.namespace = rocketmq实例id canal.mq.topic=（在后台创建） ``` +## kafka ssl配置参数 +``` +# canal.properties配置文件 + +kafka.kerberos.enable = true +kafka.kerberos.krb5.file = "../conf/kerberos/krb5.conf" +kafka.kerberos.jaas.file = "../conf/kerberos/jaas.conf" +``` + ### 3.4 启动 ``` cd /usr/local/canal/