2.3. 在 Red Hat Enterprise Linux 上使用 AMQ Streams 部署 Debezium

流程

1. 从 Red Hat Integration 下载站点下载 您要使用的 Debezium 连接器或连接器。例如，要将 Debezium 与 MySQL 数据库搭配使用，请下载 Debezium 1.9.5 MySQL Connector。

2. 在您部署了 AMQ Streams 的 Red Hat Enterprise Linux 主机上，打开终端窗口并在 /opt/kafka 中创建一个 连接器 目录，如果它尚不存在：

   $ sudo mkdir /opt/kafka/connector-plugins

   Copy to Clipboard Toggle word wrap

3. 输入以下命令来提取您下载到 /opt/kafka/connector-plugins 目录的 Debezium 连接器存档的内容。

   $ sudo unzip debezium-connector-mysql-1.9.5.Final.zip -d /opt/kafka/connector-plugins

   Copy to Clipboard Toggle word wrap
4. 对于您要安装的每个连接器，重复步骤 1 -3。

5. 在终端窗口中以 kafka 用户身份登录：

   $ su - kafka
   $ Password:

   Copy to Clipboard Toggle word wrap

6. 如果 Kafka Connect 进程正在运行，则停止它。

   1. 输入以下命令检查 Kafka Connect 是否在分布式模式下运行：

      $ jcmd | grep ConnectDistributed

      Copy to Clipboard Toggle word wrap

      如果进程正在运行，命令会返回进程 ID，例如：

      18514 org.apache.kafka.connect.cli.ConnectDistributed /opt/kafka/config/connect-distributed.properties

      Copy to Clipboard Toggle word wrap

   2. 使用进程 ID 输入 kill 命令停止进程，例如：

      $ kill 18514

      Copy to Clipboard Toggle word wrap

7. 编辑 /opt/kafka/config/ 中的 connect-distributed.properties 文件，并将 plugin.path 的值设置为 Debezium 连接器插件的父目录位置：

   plugin.path=/opt/kafka/connector-plugins

   Copy to Clipboard Toggle word wrap

8. 在分布式模式下启动 Kafka 连接。

   $ /opt/kafka/bin/connect-distributed.sh /opt/kafka/config/connect-distributed.properties

   Copy to Clipboard Toggle word wrap

9. 在 Kafka Connect 运行后，使用 Kafka Connect API 注册连接器。
   输入 curl 命令，提交 POST 请求，将您在 第 2.2 节 “规划 Debezium 连接器配置” 中指定的连接器配置 JSON 发送到 localhost:8083/connectors 的 Kafka Connect REST API 端点。
   例如：

   curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" localhost:8083/connectors/ \
   -d '{"name": "inventory-connector", "config": \
   { "connector.class": "io.debezium.connector.mysql.MySqlConnector", \
   "tasks.max": "1", \
   "database.hostname": "mysql", \
   "database.port": "3306", \
   "database.user": "debezium", \
   "database.password": "dbz", \
   "database.server.id": "184054", \
   "database.server.name": "dbserver1", \
   "database.include.list": "public.inventory", \
   "database.history.kafka.bootstrap.servers": "kafka:9092", \
   "database.history.kafka.topic": "dbhistory.inventory" } }'

   Copy to Clipboard Toggle word wrap

   要注册多个连接器，请为每个连接器提交一个单独的请求。

10. 重启 Kafka Connect 以实施您的更改。

    当 Kafka Connect 启动时，它会从 connector-plugins 目录载入配置的 Debezium 连接器。

    完成配置后，部署的连接器连接到源数据库，并为每个插入、更新或删除的行或文档生成事件。

11. 为每个 Kafka 连接 worker 节点重复步骤 5-10。