请教Canal订阅阿里云RDS(MySql)的案例

参考：https://github.com/alibaba/canal/wiki/FAQ

原回答者GitHub用户agapple

下面是一个Canal订阅阿里云RDS（MySQL）的案例。

1. 安装Canal

首先需要安装Canal，可以通过官网下载最新的Canal版本，然后解压到任意目录即可。

1. 配置Canal

Canal的配置分为两个部分，一个是Canal Server的配置，一个是Canal Client（即Canal Adapter）的配置。下面我们来分别配置这两个部分。

（1）Canal Server的配置

进入Canal目录，找到conf目录下的example目录，复制一份example目录，重命名为自己的配置文件名，例如：alibaba.properties。

然后打开该配置文件，修改以下参数：

canal.instance.master.address = {阿里云RDS的主库地址} canal.instance.master.journal.name = mysql-bin.000001 canal.instance.master.position = 4 canal.instance.dbUsername = {阿里云RDS的用户名} canal.instance.dbPassword = {阿里云RDS的密码} canal.instance.connectionCharset = UTF-8

其中，canal.instance.master.address是阿里云RDS的主库地址，canal.instance.master.journal.name是binlog的名称，canal.instance.master.position是binlog的偏移量。可以通过执行SHOW MASTER STATUS命令获取到这些信息。

（2）Canal Client的配置

Canal Client是指Canal Adapter，它负责从Canal Server中拉取binlog数据，并将其转换为目标数据源（例如Kafka、RocketMQ等）的格式。下面我们以Kafka为例，来配置Canal Client。

进入Canal目录，找到conf目录下的example目录，复制一份example目录，重命名为自己的配置文件名，例如：kafka.properties。

然后打开该配置文件，修改以下参数：

canal.instance.filter.regex = .\.. canal.instance.kafka.bootstrap.servers = {Kafka的地址} canal.instance.kafka.topic = {Kafka的Topic} canal.instance.kafka.producer.batch.size = 16384 canal.instance.kafka.producer.linger.ms = 10 canal.instance.kafka.producer.buffer.memory = 33554432 canal.instance.kafka.producer.retries = 3 canal.instance.kafka.producer.key.serializer = org.apache.kafka.common.serialization.StringSerializer canal.instance.kafka.producer.value.serializer = org.apache.kafka.common.serialization.StringSerializer

其中，canal.instance.kafka.bootstrap.servers是Kafka的地址，canal.instance.kafka.topic是Kafka的Topic。

1. 启动Canal

配置完成后，进入Canal目录，执行如下命令启动Canal：

./bin/startup.sh

1. 验证数据

Canal启动后，就可以在Kafka的Topic中看到订阅到的数据了。如果没有看到数据，可以检查Canal和Kafka的日志，看是否有报错信息。

。