本文主要是介绍Pulsar集成Debezium监听MySQL日志，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！
此处为
docker部署方式
一.准备镜像和插件
pulsar镜像：这里使用最新版本apachepulsar/pulsar:latest，这里最新版是3.2.1，自己拉
插件：pulsar-io-debezium-mysql-3.2.1.nar，这里使用3.2.1版本的，也是最新版本的，保持一直。
pulsar-io-debezium-xxx-xxx.nar，这个是在Apache pulsar离线下载包里面找到，具体为：https://archive.apache.org/dist/pulsar/pulsar-3.2.1/connectors/，此处是pulsar与debezium集成的插件，具体的数据库找对应的。
二.准备文件
1.MySQL开启binlog（MySQL如果docker启动不用开启，不知道为啥）
subo vim /etc/my.cnf[mysqld] // 添加下面三行
server_id=1
log-bin=mysql-bin
binlog_format=row
2.配置source-config.yaml（文件名可自定义）
示例：
tenant: "public" # 租户，可自定义
namespace: "default" # 命名空间，可自定义
name: "debezium-mysql-source" # source名，可自定义
topicName: "debezium-mysql-topic" # topic名，可自定义
archive: "data/pulsar-io-debezium-mysql-2.9.1.nar" # 上传的nar文件的存放位置，data目录，名字可自定义，位置在plusar目录下第一层，与bin目录平齐parallelism: 1 # 不用改configs:database.hostname: "118.195.***.***" # 数据库ipdatabase.port: "3306" # 数据库portdatabase.serverTimezone: "GMT+8" # # 数据库市区，可以没有这一条database.user: "root" # 数据库用户名database.password: "123456" # 数据库密码database.server.id: "1" # 唯一值，可自定义database.server.name: "dbserver1" # 服务名，可自定义，对应查询topic的第一截database.include.list:"demo_esb" # 需要访问的数据库## database.exclude.list: "demo_dap" # 不需要访问的数据库，可以没有这一条table.include.list:"demo_esb.ack_sample,demo_esb.aac_sample" # 需要访问的数据表，可以没有这一条## table.exclude.list:"demo_esb.ack sample_test" # 不需要访问的数据表，可以没有这一条column.include.list:"demo_esb.ack_sample.s_ID,demo_esb.ack_sample.s_coDE" # 需要访问的列，可以没有这一条##column.exclude.list:"demo_esb.ack_sample.s TEsri # 不需要访问的列，可以没有这一条## include.schema.changes:true" # 包含schema的改变，可以没有这一条database.history: "org.apache.pulsar.io.debezium.PulsarDatabaseHistory" # 不用改database.history.pulsar.topic: "my-history-topic" # topic名，可自定义database.history.pulsar.service.url: "pulsar://118.195.***.***:6651,118.195.***.***:6652,118.195.***.***:6653" # pulsar集群服务地址，自己修改key.converter: "org.apache.kafka.connect.json.JsonConverter" # 不用改value.converter: "org.apache.kafka.connect.json.JsonConverter" # 不用改pulsar.service.url: "pulsar://118.195.***.***:6651,118.195.***.***:6652,118.195.***.***:6653" # pulsar集群服务地址，自己修改，同上offset.storage.topic: "offset-topic" # topic名，可自定义
实际使用：
tenant: "public"
namespace: "default"
name: "debezium-mysql-source"
topicName: "debezium-mysql-topic"
archive: "data/pulsar-io-debezium-mysql-3.2.1.nar" # 上传的nar文件的存放位置parallelism: 1configs:database.hostname: "192.168.8.112"database.port: "3306"database.serverTimezone: "GMT+8"database.user: "root"database.password: "root"database.server.id: "2"database.server.name: "b"database.include.list: "a"database.history: "org.apache.pulsar.io.debezium.PulsarDatabaseHistory"database.history.pulsar.topic: "my-history-topic"database.history.pulsar.service.url: "pulsar://localhost:6650"key.converter: "org.apache.kafka.connect.json.JsonConverter"value.converter: "org.apache.kafka.connect.json.JsonConverter"pulsar.service.url: "pulsar://localhost:6650"offset.storage.topic: "offset-mysql-topic"
source运行之后，监控的topic，共由三截构成：
第一截：database.server.name的值
第二截：database.include.list的值，即数据库名
第三截：该数据库中表名
即：topic：server_name.database_name.table_name
三、启动
1.MySQL启动
docker run --name mysql -d -p 3306:3306 -e MYSQL_ROOT_PASSWORD=root mysql:8.0.36
此处docker启动，数据卷自己挂载。外部服务（开启binlog）的话，自己启动即可。
2.pulsar启动
docker run --name pulsar -d -p 6650:6650 -p 8080:8080 apachepulsar/pulsar:latest bin/pulsar standalone
此处单机版启动，集群启动，自己配置。
3.debezium插件启动
a.pulsar-io-debezium文件放在data下（其他目录下也可以）
docker cp ./pulsar-io-debezium-mysql-3.2.1.nar 981c770b0f87:/pulsar/data/pulsar-io-debezium-mysql-3.2.1.nar
需要用数据卷挂载方式，自己整
b.source-config.yaml文件放在conf下（其他目录下也可以）
docker cp ./source-config.yaml 981c770b0f87:/pulsar/conf/source-config.yaml
需要用数据卷挂载方式，自己整
c.启动连接器
进入pulsar目录下
./bin/pulsar-admin source localrun --source-config-file conf/source-config.yaml
四、测试
1.查看 pulsar topic
./bin/pulsar-admin topics list public/default
2.消费设置的topic消息
./bin/pulsar-client consume -s "first-subscription" persistent://public/default/b.a.t_user -n 0
3.MySQL中增改删数据，观察消费输出
这篇关于Pulsar集成Debezium监听MySQL日志的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！
