# Java客户端代码示例
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.CanalEntry.*;

public class CanalClient {
public static void main(String[] args) {
CanalConnector connector = CanalConnectors.newSingleConnector(
new InetSocketAddress(“192.168.1.51”, 11111),
“fgedu_instance”, “”, “”);

try {
connector.connect();
connector.subscribe(“fgedu_db\\..*”);
connector.rollback();

while (true) {
Message message = connector.getWithoutAck(100);
long batchId = message.getId();
int size = message.getEntries().size();

if (batchId == -1 || size == 0) {
Thread.sleep(1000);
} else {
printEntry(message.getEntries());
}
connector.ack(batchId);
}
} finally {
connector.disconnect();
}
}
}

场景二：同步到Kafka

canal.serverMode = kafka
canal.mq.servers = 192.168.1.51:9092,192.168.1.52:9092,192.168.1.53:9092
canal.mq.retries = 3
canal.mq.batchSize = 16384
canal.mq.maxRequestSize = 1048576
canal.mq.lingerMs = 100
canal.mq.bufferMemory = 33554432
canal.mq.acks = all

# 修改instance.properties配置Topic
$ vi /fgeudb/canal/server/conf/fgedu_instance/instance.properties

canal.mq.topic = canal_fgedu
canal.mq.partition = 0
canal.mq.partitionsNum = 3
canal.mq.partitionHash = .*\\..*:$pk$

# 重启Canal Server
$ sh /fgeudb/canal/server/bin/stop.sh
$ sh /fgeudb/canal/server/bin/startup.sh

场景三：使用Canal Adapter同步到目标库

server:
port: 8081

canal.conf:
mode: tcp
canalServerHost: 192.168.1.51:11111
batchSize: 500
syncBatchSize: 1000

srcDataSources:
defaultDS:
url: jdbc:mysql://192.168.1.51:3306/fgedu_db?useSSL=false
username: canal
password: canal123

canalAdapters:
– instance: fgedu_instance
groups:
– groupId: g1
outerAdapters:
– name: logger
– name: rdb
key: mysql1
properties:
jdbc.driverClassName: com.mysql.cj.jdbc.Driver
jdbc.url: jdbc:mysql://192.168.1.52:3306/fgedu_db_target
jdbc.username: root
jdbc.password: root123

# 配置表映射
$ vi /fgeudb/canal/adapter/conf/rdb/fgedu_order.yml

dataSourceKey: defaultDS
destination: fgedu_instance
groupId: g1
outerAdapterKey: mysql1
concurrent: true

dbMapping:
database: fgedu_db
table: order_info
targetTable: order_info
targetPk:
id: id
mapAll: true
etlCondition: “where create_time >= ‘2026-01-01′”

# 启动Adapter
$ cd /fgeudb/canal/adapter
$ sh bin/startup.sh

# 查看日志
$ tail -f logs/adapter/adapter.log

输出示例如下：
2026-04-04 10:00:00.000 [main] INFO c.a.o.canal.adapter.launcher.CanalAdapterApplication – ## start the canal adapter.
2026-04-04 10:00:00.100 [main] INFO c.a.o.canal.adapter.launcher.CanalAdapterApplication – ## the canal adapter is running now ……

# 配置zoo.cfg
$ vi /fgeudb/zookeeper/conf/zoo.cfg

tickTime=2000
initLimit=10
syncLimit=5
dataDir=/fgeudb/zookeeper/data
clientPort=2181
server.1=192.168.1.51:2888:3888
server.2=192.168.1.52:2888:3888
server.3=192.168.1.53:2888:3888

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

# 启动ZooKeeper
$ /fgeudb/zookeeper/bin/zkServer.sh start

# 检查状态
$ /fgeudb/zookeeper/bin/zkServer.sh status

输出示例如下：
Mode: leader

步骤2：Canal Server HA配置

# 编辑canal.properties
$ vi /fgeudb/canal/server/conf/canal.properties

canal.id = 1 # 第二台设置为2
canal.ip = 192.168.1.51 # 绑定各自IP
canal.port = 11111
canal.zkServers = 192.168.1.51:2181,192.168.1.52:2181,192.168.1.53:2181
canal.instance.global.spring.xml = classpath:spring/default-instance.xml

# 启动两台Canal Server
$ sh /fgeudb/canal/server/bin/startup.sh

# 查看HA状态
$ tail -f /fgeudb/canal/server/logs/canal/canal.log

输出示例如下：
2026-04-04 10:00:00.000 [main] INFO c.a.o.canal.deployer.CanalController – ## start the canal server[192.168.1.51:11111]
2026-04-04 10:00:00.100 [main] INFO c.a.o.canal.deployer.CanalLauncher – ## the canal server is running now ……
2026-04-04 10:00:00.200 [main] INFO c.a.o.canal.common.zookeeper.ZkClientx – zookeeper state changed (SyncConnected)

步骤3：监控Canal运行状态

# 通过ZooKeeper查看状态
$ /fgeudb/zookeeper/bin/zkCli.sh

[zk: localhost:2181(CONNECTED) 0] ls /otter/canal/destinations
[fgedu_instance]

[zk: localhost:2181(CONNECTED) 1] ls /otter/canal/destinations/fgedu_instance
[cluster, running]

[zk: localhost:2181(CONNECTED) 2] get /otter/canal/destinations/fgedu_instance/running

输出示例如下：
{“active”:true,”address”:”192.168.1.51:11111″}