canal监听mysql数据变化和表结构变更，发送变化日记到kafka接收

canal配置kafka 接收消息

第一步，从官网下载canal包并解压到自己需要放置的目录下，；

解压好了，你会在conf目录下看到两个配置文件，canal.properties,/conf/example/instance.properties

第二步，在canal.properties中配置

canal.serverMode = kafka

kafka.bootstrap.servers = 172.16.158.74:9092

这个IP 是kafka所在服务的ip

canal.instance.master.address= 数据库链接信息

canal.instance.dbUsername= 数据账号
canal.instance.dbPassword= 数据库密码
canal.instance.connectionCharset = UTF-8

kafka创建的topic:

canal.mq.topic=test_cmp

第三步，在instance.properties中配置链接监听的数据库；

安装kafka:

linux环境下:

下载：wget https://dlcdn.apache.org/kafka/3.3.1/kafka_2.13-3.3.1.tgz

解压：

$ tar -xzf kafka_2.13-3.3.1.tgz
$ cd kafka_2.13-3.3.1

a安装zk :

$ bin/zookeeper-server-start.sh config/zookeeper.properties

修改kafka的配置文件server.properties
中的

# 放开此IP地址便于其他服务访问
listeners=PLAINTEXT://172.16.158.74:9092

启动kafka:

$ bin/kafka-server-start.sh config/kraft/server.properties

创建kafka的topic

$ bin/kafka-topics.sh --create --topic quickstart-events --bootstrap-server localhost:9092
检查是否创建成功：

$ bin/kafka-topics.sh --describe --topic quickstart-events --bootstrap-server localhost:9092

启动kafka的生产者：

$ bin/kafka-console-producer.sh --topic quickstart-events --bootstrap-server localhost:9092
启动kafka消费者：

$ bin/kafka-console-consumer.sh --topic quickstart-events --from-beginning --bootstrap-server localhost:9092

以上配置信息 都已完成

此刻修改数据库任何信息都能监听到事件

原文地址：http://www.cnblogs.com/Generalwake/p/16828222.html