# 检查主机名解析
# hostname -f
debezium01.fgedu.net.cn

# 配置/etc/hosts
# cat /etc/hosts
127.0.0.1 localhost
192.168.1.51 debezium01.fgedu.net.cn debezium01
192.168.1.52 debezium02.fgedu.net.cn debezium02
192.168.1.53 debezium03.fgedu.net.cn debezium03

# 关闭SELinux（如未关闭）
# vi /etc/selinux/config
SELINUX=disabled

# 检查防火墙状态
# systemctl status firewalld

# 关闭防火墙（生产环境建议开放特定端口）
# systemctl stop firewalld
# systemctl disable firewalld

# 或者开放所需端口
# firewall-cmd –permanent –add-port=2181/tcp
# firewall-cmd –permanent –add-port=2888/tcp
# firewall-cmd –permanent –add-port=3888/tcp
# firewall-cmd –permanent –add-port=9092/tcp
# firewall-cmd –permanent –add-port=8083/tcp
# firewall-cmd –reload

# 添加以下参数
fs.file-max = 6815744
vm.swappiness = 10
vm.max_map_count = 262144
net.core.somaxconn = 32768
net.ipv4.tcp_max_syn_backlog = 65536

# 使配置生效
# sysctl -p

# 配置文件描述符限制
# vi /etc/security/limits.conf

# 添加以下内容
* soft nofile 65536
* hard nofile 65536
* soft nproc 65536
* hard nproc 65536

# 验证配置
# ulimit -n
65536

# 验证Java版本
$ java -version
openjdk version “11.0.22” 2024-01-16 LTS
OpenJDK Runtime Environment (build 11.0.22+7-LTS)
OpenJDK 64-Bit Server VM (build 11.0.22+7-LTS, mixed mode, sharing)

# 配置JAVA_HOME环境变量
# vi /etc/profile.d/java.sh

export JAVA_HOME=/usr/lib/jvm/java-11-openjdk
export PATH=$JAVA_HOME/bin:$PATH

# 使环境变量生效
# source /etc/profile.d/java.sh

# 验证JAVA_HOME
$ echo $JAVA_HOME
/usr/lib/jvm/java-11-openjdk

# 解压安装
# tar -xzf kafka_2.13-3.6.1.tgz
# mv kafka_2.13-3.6.1 /data/kafka

# 配置ZooKeeper
# vi /data/kafka/config/zookeeper.properties

dataDir=/data/zookeeper/data
clientPort=2181
maxClientCnxns=0
admin.enableServer=false

# 集群配置
server.1=192.168.1.51:2888:3888
server.2=192.168.1.52:2888:3888
server.3=192.168.1.53:2888:3888

# 创建数据目录
# mkdir -p /data/zookeeper/data

# 创建myid文件
# echo “1” > /data/zookeeper/data/myid

# 启动ZooKeeper
# /data/kafka/bin/zookeeper-server-start.sh -daemon /data/kafka/config/zookeeper.properties

# 检查ZooKeeper状态
$ echo stat | nc localhost 2181

# 输出示例：
Zookeeper version: 3.8.3
Clients:
/127.0.0.1:54321[0](queued=0,recved=1,sent=0)
Latency min/avg/max: 0/1.5/5
Received: 10
Sent: 9
Connections: 1
Outstanding: 0
Zxid: 0x100000003
Mode: leader
Node count: 5

# Broker配置
broker.id=1
listeners=PLAINTEXT://192.168.1.51:9092
advertised.listeners=PLAINTEXT://192.168.1.51:9092
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600

# 日志配置
log.dirs=/data/kafka/logs
num.partitions=3
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=2

# ZooKeeper配置
zookeeper.connect=192.168.1.51:2181,192.168.1.52:2181,192.168.1.53:2181
zookeeper.connection.timeout.ms=18000

# 日志保留配置
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000

# 创建日志目录
# mkdir -p /data/kafka/logs

# 启动Kafka Broker
# /data/kafka/bin/kafka-server-start.sh -daemon /data/kafka/config/server.properties

# 检查Kafka进程
$ jps | grep Kafka

# 输出示例：
12345 Kafka

# 创建测试Topic
$ /data/kafka/bin/kafka-topics.sh –create –topic test-topic –bootstrap-server 192.168.1.51:9092 –partitions 3 –replication-factor 3

# 输出示例：
Created topic test-topic.

# 查看Topic列表
$ /data/kafka/bin/kafka-topics.sh –list –bootstrap-server 192.168.1.51:9092

# 输出示例：
test-topic

# 基础配置
bootstrap.servers=192.168.1.51:9092,192.168.1.52:9092,192.168.1.53:9092
group.id=connect-cluster
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=true
value.converter.schemas.enable=true

# Offset存储配置
offset.storage.topic=connect-offsets
offset.storage.replication.factor=3
offset.storage.partitions=3

# Config存储配置
config.storage.topic=connect-configs
config.storage.replication.factor=1

# Status存储配置
status.storage.topic=connect-status
status.storage.replication.factor=3
status.storage.partitions=3

# 连接器配置
offset.flush.interval.ms=10000
rest.port=8083
rest.advertised.host.name=192.168.1.51

# 插件路径
plugin.path=/data/kafka/plugins

# 创建插件目录
# mkdir -p /data/kafka/plugins

# 启动Kafka Connect
# /data/kafka/bin/connect-distributed.sh -daemon /data/kafka/config/connect-distributed.properties

# 检查Kafka Connect状态
$ curl -s http://192.168.1.51:8083/ | jq

# 输出示例：
{
“version”: “3.6.1”,
“commit”: “5e3c2b738d253ff5”,
“kafka_cluster_id”: “abc123-def456-ghi789”
}

# 输出示例：
–2024-04-05 10:00:00– https://repo1.maven.org/maven2/io/debezium/debezium-connector-mysql/2.7.0.Final/debezium-connector-mysql-2.7.0.Final-plugin.tar.gz
Resolving repo1.maven.org… 199.232.196.209
Connecting to repo1.maven.org|199.232.196.209|:443… connected.
HTTP request sent, awaiting response… 200 OK
Length: 45678901 (44M) [application/x-gzip]
Saving to: ‘debezium-connector-mysql-2.7.0.Final-plugin.tar.gz’

debezium-connector-mysql-2.7.0.Final-plugin.tar.gz 100%[=================================================>] 43.56M 15.6MB/s in 2.8s

2024-04-05 10:00:03 (15.6 MB/s) – ‘debezium-connector-mysql-2.7.0.Final-plugin.tar.gz’ saved [45678901/45678901]

# 解压到插件目录
# tar -xzf debezium-connector-mysql-2.7.0.Final-plugin.tar.gz -C /data/kafka/plugins/

# 下载PostgreSQL连接器
# wget https://repo1.maven.org/maven2/io/debezium/debezium-connector-postgres/2.7.0.Final/debezium-connector-postgres-2.7.0.Final-plugin.tar.gz
# tar -xzf debezium-connector-postgres-2.7.0.Final-plugin.tar.gz -C /data/kafka/plugins/

# 验证安装
$ ls -l /data/kafka/plugins/

# 输出示例：
total 0
drwxr-xr-x 2 root root 4096 Apr 5 10:00 debezium-connector-mysql
drwxr-xr-x 2 root root 4096 Apr 5 10:00 debezium-connector-postgres

# 查看连接器文件
$ ls -lh /data/kafka/plugins/debezium-connector-mysql/

# 输出示例：
total 45M
-rw-r–r– 1 root root 12M Apr 5 10:00 debezium-connector-mysql-2.7.0.Final.jar
-rw-r–r– 1 root root 15M Apr 5 10:00 debezium-core-2.7.0.Final.jar
-rw-r–r– 1 root root 2.5M Apr 5 10:00 mysql-connector-j-8.2.0.jar
…

# 或者直接kill进程
# kill $(pgrep -f ConnectDistributed)

# 重新启动Kafka Connect
# /data/kafka/bin/connect-distributed.sh -daemon /data/kafka/config/connect-distributed.properties

# 检查连接器插件
$ curl -s http://192.168.1.51:8083/connector-plugins | jq

# 输出示例：
[
{
“class”: “io.debezium.connector.mysql.MySqlConnector”,
“type”: “source”,
“version”: “2.7.0.Final”
},
{
“class”: “io.debezium.connector.postgresql.PostgresConnector”,
“type”: “source”,
“version”: “2.7.0.Final”
}
]

[mysqld]
server-id = 1
log-bin = mysql-bin
binlog_format = ROW
binlog_row_image = FULL
gtid_mode = ON
enforce_gtid_consistency = ON

# 重启MySQL
# systemctl restart mysqld

# 创建Debezium用户
$ mysql -u root -p

mysql> CREATE USER ‘debezium’@’%’ IDENTIFIED BY ‘fgedu_debezium_2024’;
mysql> GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO ‘debezium’@’%’;
mysql> FLUSH PRIVILEGES;

# 验证binlog配置
mysql> SHOW VARIABLES LIKE ‘binlog_format’;

# 输出示例：
+—————+——-+
| Variable_name | Value |
+—————+——-+
| binlog_format | ROW |
+—————+——-+

# 创建测试数据库和表
mysql> CREATE DATABASE fgedudb;
mysql> USE fgedudb;
mysql> CREATE TABLE fgedu_employees (
id INT PRIMARY KEY AUTO_INCREMENT,
name VARCHAR(100),
department VARCHAR(50),
salary DECIMAL(10,2),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
mysql> INSERT INTO fgedu_employees (name, department, salary) VALUES
(‘张三’, ‘技术部’, 15000.00),
(‘李四’, ‘销售部’, 12000.00),
(‘王五’, ‘财务部’, 13000.00);

{
“name”: “fgedudb-mysql-connector”,
“config”: {
“connector.class”: “io.debezium.connector.mysql.MySqlConnector”,
“database.hostname”: “192.168.1.51”,
“database.port”: “3306”,
“database.user”: “debezium”,
“database.password”: “fgedu_debezium_2024”,
“database.server.id”: “5400”,
“database.server.name”: “fgedudb_server”,
“database.include.list”: “fgedudb”,
“table.include.list”: “fgedudb.fgedu_employees,fgedudb.fgedu_departments”,
“database.history.kafka.bootstrap.servers”: “192.168.1.51:9092”,
“database.history.kafka.topic”: “schema-changes.fgedudb”,
“include.schema.changes”: “true”,
“snapshot.mode”: “initial”,
“snapshot.locking.mode”: “minimal”,
“snapshot.fetch.size”: “2000”,
“max.queue.size”: “8192”,
“max.batch.size”: “2048”,
“poll.interval.ms”: “500”,
“heartbeat.interval.ms”: “5000”,
“decimal.handling.mode”: “string”,
“time.precision.mode”: “connect”,
“tombstones.on.delete”: “true”,
“database.connectionTimeZone”: “Asia/Shanghai”
}
}

# 注册连接器
$ curl -X POST -H “Content-Type: application/json” –data @/data/kafka/config/mysql-connector.json http://192.168.1.51:8083/connectors

# 输出示例：
{
“name”: “fgedudb-mysql-connector”,
“config”: {
“connector.class”: “io.debezium.connector.mysql.MySqlConnector”,
“database.hostname”: “192.168.1.51”,
…
},
“tasks”: [],
“type”: “source”
}