Pulsar集成Debezium source connector的详情可参考：链接
以下数据库都可：
MySQL
PostgreSQL
MongoDB
此处为
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.***.***" # 数据库ip
database.port: "3306" # 数据库port
database.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: 1
configs:
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
