Canal + Elasticsearch 实时同步实战（二）：部署 server 与 adapter，跑通全量 + 增量

系列文章

• 第一篇：选型、架构与前置准备
• 第二篇：部署 canal-server 与 adapter ← 本篇
• 第三篇：8 个踩坑记录与生产环境注意事项

前置条件：MySQL 已开 ROW binlog，canal 账号已建好，JDK 8 已安装（见第一篇）。 本篇所有配置文件里的 <MYSQL_HOST>、<CANAL_DB_PASSWORD> 等占位符，请替换成你自己的值。

---

目录

• 一、文件与目录规划
• 二、配置 canal-server
• 三、启动并验证 binlog 连接
• 四、部署 canal-adapter
• 五、全量同步与增量验证
• 六、生产化改造
• 七、运维命令速查
• 附录：完整配置文件

---

一、文件与目录规划

拿到两个 tar 包后，统一放在一个根目录下，后面所有操作都基于这个结构：

bash
mkdir -p /opt/canal && cd /opt/canal

# canal-server
tar -xzf canal.deployer-1.1.7.tar.gz -C server

# canal-adapter
tar -xzf canal.adapter-1.1.7.tar.gz -C adapter

最终结构：

/opt/canal/
├── server/                        # canal-server (deployer)
│   ├── bin/
│   │   ├── startup.sh
│   │   ├── stop.sh
│   │   └── restart.sh
│   ├── conf/
│   │   ├── canal.properties       # 全局配置
│   │   ├── logback.xml
│   │   └── example/
│   │       └── instance.properties  # 单实例配置（核心）
│   ├── lib/
│   └── logs/
│       ├── canal/canal.log        # 框架日志
│       └── example/example.log    # 实例日志 ← 排错主要看这个
│
└── adapter/                       # canal-adapter
    ├── bin/{startup.sh,stop.sh}
    ├── conf/
    │   ├── application.yml        # 主配置
    │   └── es7/                   # 映射文件目录
    ├── lib/
    ├── logs/adapter/adapter.log   # 同步日志
    └── plugin/                    # 各适配器 jar

conf/ 下还有 es6、es8、rdb、hbase 等目录，都是官方示例。我们只关心 es7。

---

二、配置 canal-server

2.1 canal.properties：只改两处

这个文件有一百多行，绝大部分用默认值就行，只需要改两个：

properties
# 实例列表，多个用逗号分隔。每个名字对应 conf/ 下的一个同名目录
canal.destinations = example

# 我们不用内嵌 H2 做位点存储，关掉避免额外问题
canal.instance.tsdb.enable = false

其他可能有用的默认值，了解一下即可：

配置	默认值	说明
canal.serverMode	tcp	消息投递模式，另有 kafka / rocketmq / rabbitmq
canal.port	11111	客户端订阅端口
canal.admin.port	11110	给 canal-admin 调用的 HTTP API（不是网页）
canal.metrics.pull.port	11112	Prometheus 指标端口
canal.instance.memory.buffer.size	16384	内存环形缓冲，单位是 memunit（默认 1KB）

canal.instance.tsdb.enable 关掉是必须的吗？不是。 它默认开启，用 H2 存表结构快照（用于解析 DDL）。单机小数据量没问题，但 H2 会额外占用内存和磁盘，且偶发锁表。 学习阶段关掉更省心，代价是 DDL 解析能力弱一些。

2.2 instance.properties：核心配置

路径 conf/example/instance.properties。官方文件带大量注释，下面是精简后的完整内容：

properties
# ---------- GTID ----------
canal.instance.gtidon = false

# ---------- 目标 MySQL ----------
canal.instance.master.address = <MYSQL_HOST>:3306

# 起点位点（留空 = 自动取当前位置）
canal.instance.master.journal.name = mysql-bin.000048
canal.instance.master.position = 4059448
canal.instance.master.timestamp =
canal.instance.master.gtid =

# ---------- 同步账号 ----------
canal.instance.dbUsername = canal
canal.instance.dbPassword = <CANAL_DB_PASSWORD>
canal.instance.connectionCharset = UTF-8
canal.instance.enableDruid = false

# ---------- 过滤规则 ----------
canal.instance.filter.regex = nfturbo\\..*
canal.instance.filter.black.regex =

# ---------- tsdb ----------
canal.instance.tsdb.enable = false

# ---------- MQ ----------
canal.mq.topic = example
canal.mq.partition = 0

2.3 位点（journal name / position）说明

这是最容易搞混的地方，单独说清楚：

三种情况：

配置	Canal 的行为
两个都留空	启动时执行 SHOW MASTER STATUS，从当前最新位点开始（存量数据不会同步）
两个都填上	直接从指定位点开始 dump，不执行 SHOW MASTER STATUS
只填 journal，不填 position	从该文件的起始位置开始

启动成功之后，Canal 会把位点持久化到 conf/example/meta.dat。 以后每次重启都从 meta.dat 恢复，配置文件里的位点只在第一次启动时生效。

这一点很重要：你改了配置文件里的位点，重启后发现没生效，是因为 meta.dat 覆盖了它。 真要重新定位点，得停服务、删 meta.dat、再改配置、再启动。

怎么查当前位点？

MySQL 8.0 及以前：

sql
SHOW MASTER STATUS;

MySQL 8.4（旧语句已移除）：

sql
SHOW BINARY LOG STATUS;

输出：

+----------------+----------+--------------+------------------+
| File           | Position | Binlog_Do_DB | Binlog_Ignore_DB |
+----------------+----------+--------------+------------------+
| bin-log.000048 |  4059448 |              |                  |
+----------------+----------+--------------+------------------+

本文为什么要写死位点？ 因为 MySQL 8.4 移除了 SHOW MASTER STATUS，而 Canal 1.1.7 里是硬编码调用它的，不改配置就会启动失败。 写死位点是绕过方案，详细原因和风险见第三篇的坑 1。 如果你的 MySQL 是 8.0 或 5.7，位点留空即可，不用写。

2.4 filter.regex 的写法

properties
# 匹配 nfturbo 库下所有表
canal.instance.filter.regex = nfturbo\\..*

两个注意点：

1. 必须是双反斜杠。Java Properties 文件会做一次转义，写 \\ 最终才是正则里的 \。
2. 多个规则用逗号分隔：nfturbo\\.order_.*,nfturbo\\.user

常见的几种写法：

properties
# 只同步某几张表
canal.instance.filter.regex = nfturbo\\.user,nfturbo\\.collection

# 排除某些表（黑名单）
canal.instance.filter.black.regex = nfturbo\\.undo_log,nfturbo\\.transaction_log

undo_log、transaction_log 这类中间表建议加进黑名单。它们写入频繁但没有同步价值，白白消耗 adapter 的处理能力。

---

三、启动并验证 binlog 连接

3.1 启动

bash
cd /opt/canal/server && ./bin/startup.sh

启动脚本会打印一长串 CLASSPATH，最后 cd 回来，不用管。

3.2 看实例日志（最重要的一步）

bash
tail -f /opt/canal/server/logs/example/example.log

正常输出应该是这样：

INFO  c.a.otter.canal.parse.inbound.mysql.dbsync.LogEventConvert - --> init table filter : ^nfturbo\..*$
INFO  c.a.otter.canal.instance.core.AbstractCanalInstance - start successful....
WARN  ...RdsBinlogEventParserProxy - ---> find start position successfully,
      EntryPosition[included=false,journalName=bin-log.000048,position=4059448,...],
      cost : 0ms , the next step is binlog dump

看到 start successful 和 find start position successfully 就说明起点找到了。

3.3 确认复制链路

光看 Canal 自己的日志不够，一定要去 MySQL 侧确认真的有一条复制连接：

sql
SHOW PROCESSLIST;

应该看到：

Id      User    Host                 Command       State
21355   canal   10.0.0.5:33544       Binlog Dump   Source has sent all binlog to replica; waiting for more updates

这一步极其重要，原因有两个：

• 只要出现 Binlog Dump 且状态是 "waiting for more updates"，说明复制链路已经通了，后面的问题一定是 adapter 的锅，可以大幅缩小排查范围。
• 如果没有这条线程，说明 Canal 虽然"启动成功"了，但根本没连上 MySQL——这时候去看 example.log 里的 ERROR，多半是账号权限或网络问题。

一个常见的假象：start successful 只代表 Canal 内部组件初始化完了，不代表 binlog 连接建立成功。 我见过不少人看到这行就以为完事了，结果数据一直不同步。

---

四、部署 canal-adapter

4.1 换掉过老的 JDBC 驱动

这是必须做的一步，否则连 MySQL 8 会认证失败。

adapter 自带的是 mysql-connector-java-5.1.48.jar（2015 年的版本），不支持 caching_sha2_password。

bash
cd /opt/canal/adapter/lib

# 下载 8.x 驱动（国内用阿里云镜像快很多）
curl -fsSL -o mysql-connector-java-8.0.33.jar \
  https://maven.aliyun.com/repository/public/com/mysql/mysql-connector-j/8.0.33/mysql-connector-j-8.0.33.jar

# 把旧驱动移出 lib 目录
mv mysql-connector-java-5.1.48.jar /opt/canal/

两个细节，都踩过：

1. Maven 坐标在 8.0.31 之后变了：mysql:mysql-connector-java → com:mysql-connector-j。 按旧坐标下载 8.0.33 会 404，别怀疑网络，是路径不对。
2. 旧驱动要"移出去"，不能改名成 .bak。 startup.sh 用 lib/*.jar 拼 CLASSPATH，.bak、.disabled 都会被加进去，两个驱动共存会有类加载顺序问题。

验证：

bash
ls -l /opt/canal/adapter/lib/ | grep mysql
# 只应该剩 mysql-connector-java-8.0.33.jar

4.2 application.yml

路径 conf/application.yml。

官方模板很长，但不要图省事删键（第三篇坑 5 会解释为什么）。下面是可以直接用的版本：

yaml
server:
  port: 8081

spring:
  jackson:
    date-format: yyyy-MM-dd HH:mm:ss
    time-zone: GMT+8
    default-property-inclusion: non_null

canal.conf:
  mode: tcp
  flatMessage: true
  zookeeperHosts:          # ← 不能删
  syncBatchSize: 1000
  retries: -1
  timeout:                 # ← 不能删
  accessKey:               # ← 不能删
  secretKey:               # ← 不能删

  consumerProperties:
    canal.tcp.server.host: 127.0.0.1:11111
    canal.tcp.zookeeper.hosts:
    canal.tcp.batch.size: 500
    canal.tcp.username:
    canal.tcp.password:

  srcDataSources:
    defaultDS:
      url: jdbc:mysql://<MYSQL_HOST>:3306/nfturbo?useUnicode=true&characterEncoding=UTF-8&useSSL=false&serverTimezone=Asia/Shanghai&allowPublicKeyRetrieval=true
      username: canal
      password: '<CANAL_DB_PASSWORD>'

  canalAdapters:
  - instance: example
    groups:
    - groupId: g1
      outerAdapters:
      - name: es7
        hosts: http://127.0.0.1:9200    # ← 必须带 http://
        properties:
          mode: rest                     # ← 必须显式声明
          cluster.name: my-cluster
          security.auth: 'elastic:<ES_PASSWORD>'

四个"看着不起眼但必须写"的配置

配置	不写会怎样	简述
accessKey: / secretKey:	启动直接 NPE	Properties.put() 不接受 null 值
mode: rest	None of the configured nodes are available	es7 适配器默认走 TransportClient
hosts: http://...	Illegal character in scheme name	REST 模式要求完整 URL
allowPublicKeyRetrieval=true	Public Key Retrieval is not allowed	非 SSL 下取 RSA 公钥要用

前三个的详细分析在第三篇坑 3/4/5，这里先记住"这么写就对了"。

JDBC URL 参数说明

jdbc:mysql://<MYSQL_HOST>:3306/nfturbo
  ?useUnicode=true              # 支持中文
  &characterEncoding=UTF-8
  &useSSL=false                 # 内网不加密，避免证书问题
  &serverTimezone=Asia/Shanghai # 避免时间字段差 8 小时
  &allowPublicKeyRetrieval=true # caching_sha2 认证必需

serverTimezone 漏写的后果很隐蔽：数据能同步过去，但所有时间字段都偏移 8 小时。 因为 JDBC 驱动默认按 UTC 解析，而 MySQL 存的是东八区时间。

ES 安全认证

如果 ES 开了 xpack security（本文环境开了）：

yaml
        properties:
          security.auth: 'elastic:<ES_PASSWORD>'

密码里有特殊字符（*、! 等）时必须用单引号包住，否则 YAML 解析会出问题。

4.3 写映射文件

路径 conf/es7/*.yml，一个文件对应一个索引：

yaml
dataSourceKey: defaultDS
destination: example
groupId: g1
esMapping:
  _index: nfturbo_collection
  _id: _id
  sql: "select id as _id, gmt_create, gmt_modified, name, cover, class_id, price,
        quantity, detail, saleable_inventory, identifier, state, create_time,
        creator_id, version from collection"
  commitBatch: 1000

字段	作用
dataSourceKey	对应 application.yml 里 srcDataSources 的 key
destination	canal-server 里的实例名（canal.destinations）
groupId	消费组，多个 adapter 消费同一实例时用来区分
_index	ES 索引名
_id	文档 ID 取哪个字段（SQL 结果里必须有这个别名）
sql	回查数据的 SQL，可以 join
commitBatch	批量提交条数

_id 的设计要点

sql
select id as _id, ... from collection          -- 单表：直接用主键
select concat('order_', id) as _id, ... from t -- 分表：加前缀避免冲突
select concat(a.id, '_', b.id) as _id, ...     -- 多表 join：组合主键

_id 决定了 UPDATE / DELETE 能不能正确落到文档上。没有正确的 _id，同步就只会往 ES 里堆重复数据。

关于 join 的坑（提前说）

yaml
sql: "select a.id as _id, a.title, b.name as category_name
      from article a left join category b on b.id = a.category_id"

这条 SQL 能查到 category_name，但只有 article 表发生变化时才会触发同步。 如果只改了 category.name，ES 里的文章文档不会更新，会一直是旧名字。

解决办法：给 category 单独配一条映射，或者接受这个延迟（大部分博客系统都这么做）。

4.4 先建索引，再启动

adapter 不会自动创建索引，必须先手工建好，而且要带 mapping：

bash
curl -u 'elastic:<ES_PASSWORD>' -X PUT "http://127.0.0.1:9200/nfturbo_collection" \
  -H 'Content-Type: application/json' -d '{
  "settings": { "number_of_shards": 1, "number_of_replicas": 0 },
  "mappings": {
    "properties": {
      "name":        { "type": "text", "fields": { "keyword": { "type": "keyword", "ignore_above": 256 } } },
      "cover":       { "type": "keyword" },
      "price":       { "type": "double" },
      "quantity":    { "type": "long" },
      "state":       { "type": "keyword" },
      "creator_id":  { "type": "keyword" },
      "create_time": { "type": "date", "format": "yyyy-MM-dd HH:mm:ss||epoch_millis" },
      "gmt_create":  { "type": "date", "format": "yyyy-MM-dd HH:mm:ss||epoch_millis" }
    }
  }
}'

为什么必须是显式 mapping？

adapter 写入前会读索引的 mapping，按字段类型做值转换（比如把 JDBC 的 Timestamp 转成 ES 认识的日期字符串）。 空 mapping 里查不到字段类型，getEsType() 返回 null，直接 NPE。

字段类型对照表：

MySQL 类型	ES 类型	说明
bigint	long
int / tinyint	integer	tinyint(1) 作布尔用时可用 boolean
decimal / double	double	金额不建议用 float
varchar	keyword	精确匹配、聚合
varchar（要全文搜索）	text + keyword 子字段	一套数据两种用途
text	text
datetime / timestamp	date + format	见下
json	object 或 keyword	看是否需要拆字段查询

日期格式建议用双格式：

json
"format": "yyyy-MM-dd HH:mm:ss||epoch_millis"

这样字符串和毫秒时间戳都能吃，兼容性最好。

4.5 启动 adapter

bash
cd /opt/canal/adapter && ./bin/startup.sh

看日志 logs/adapter/adapter.log，出现这三行才算成功：

INFO  c.a.o.c.client.adapter.es.core.config.ESSyncConfigLoader - ## ES mapping config loaded
INFO  c.a.o.canal.adapter.launcher.loader.CanalAdapterLoader - Load canal adapter: es7 succeed
INFO  c.a.otter.canal.adapter.launcher.loader.AdapterProcessor - =====> Subscribe destination: example succeed <=====

三行的含义分别是：映射文件加载成功 → ES 适配器初始化成功 → 订阅 canal 实例成功。 任何一行没出现，都说明对应环节有问题。

启动比较慢（30~60 秒），因为它要初始化 ES 客户端、连 canal、加载所有映射。别急着判断失败，等一分钟再看。

---

五、全量同步与增量验证

5.1 全量同步存量数据

adapter 内置了一个 ETL 接口，POST 一下就把存量数据刷进 ES：

bash
curl -X POST "http://127.0.0.1:8081/etl/es7/nfturbo_collection.yml"

返回：

json
{"succeeded":true,"resultMessage":"导入ES 数据：8 条"}

URL 结构是 /etl/{适配器名}/{映射文件名}，文件名必须带 .yml。

5.2 验证数据

bash
curl -u 'elastic:<ES_PASSWORD>' "http://127.0.0.1:9200/nfturbo_collection/_count"
# {"count":8,"_shards":{"total":1,"successful":1,"skipped":0,"failed":0}}

如果 count 是 0，但 ETL 说导入了 8 条——不是没同步，是 ES 还没 refresh。

ES 默认 refresh_interval 是 1 秒，写入后立刻查可能查不到。三种办法：

bash
# 1. 强制刷新
curl -u 'elastic:<ES_PASSWORD>' -X POST "http://127.0.0.1:9200/nfturbo_collection/_refresh"

# 2. 查询时带 refresh 参数
curl -u 'elastic:<ES_PASSWORD>' "http://127.0.0.1:9200/nfturbo_collection/_search?refresh=true"

# 3. 简单粗暴，sleep 1 再查

看单条数据确认字段映射正确：

bash
curl -u 'elastic:<ES_PASSWORD>' 'http://127.0.0.1:9200/nfturbo_collection/_doc/10000?pretty'

json
{
  "_index": "nfturbo_collection",
  "_id": "10000",
  "_source": {
    "name": "Founders Pirate",
    "price": 1.09,
    "quantity": 100,
    "saleable_inventory": 99,
    "state": "SUCCEED"
  }
}

5.3 增量验证

增量不需要任何额外操作——adapter 一直订阅着 canal，MySQL 一有变更就自动推。

验证方法：改一条数据，然后立刻查 ES。

sql
-- 在业务库执行
UPDATE collection SET name = 'Founders Pirate 2' WHERE id = 10000;

bash
sleep 2
curl -u 'elastic:<ES_PASSWORD>' 'http://127.0.0.1:9200/nfturbo_collection/_doc/10000?pretty'

延迟通常在 1~2 秒：binlog 传输（几十毫秒） + adapter 处理 + ES refresh（1 秒）。

如果超过 10 秒还没变化，按这个顺序排查：

1. tail -f /opt/canal/server/logs/example/example.log 有没有新事件
2. tail -f /opt/canal/adapter/logs/adapter/adapter.log 有没有写 ES 的记录
3. ES 里 _refresh 一下再看

---

六、生产化改造

跑通只是第一步。startup.sh 用 nohup 起的进程，服务器一重启就没了。

6.1 调小 JVM 内存

Canal 默认 -Xms2g -Xmx3g，对小机器太奢侈（本文机器只有 7.4G，ES 还要分走一部分）。

改 bin/startup.sh 里的 JVM 参数：

bash
sed -i 's/-Xms2g -Xmx3g -Xmn1g/-Xms512m -Xmx1024m -Xmn256m/' \
  /opt/canal/server/bin/startup.sh /opt/canal/adapter/bin/startup.sh

sed -i 's/-Xms2g -Xmx3g -XX:+UseG1GC/-Xms512m -Xmx1024m -XX:+UseG1GC/' \
  /opt/canal/server/bin/startup.sh /opt/canal/adapter/bin/startup.sh

因为脚本里有两套分支（G1 和 CMS），所以两条 sed 都要执行。

验证：

bash
ps -ef | grep canal | grep -o '\-Xm[sx][^ ]*' | sort | uniq -c
# 2 -Xms512m
# 2 -Xmx1024m

升级 Canal 版本时 startup.sh 会被覆盖，记得重新改一次。

6.2 systemd 托管

/etc/systemd/system/canal-server.service：

ini
[Unit]
Description=Canal Deployer 1.1.7 (MySQL binlog parser)
After=network-online.target docker.service
Wants=network-online.target

[Service]
Type=forking
PIDFile=/opt/canal/server/bin/canal.pid
Environment=JAVA_HOME=/usr/local/java/jdk1.8.0_202
Environment=PATH=/usr/local/java/jdk1.8.0_202/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
ExecStart=/opt/canal/server/bin/startup.sh
ExecStop=/opt/canal/server/bin/stop.sh
Restart=on-failure
RestartSec=15
LimitNOFILE=65535
TimeoutStartSec=180

[Install]
WantedBy=multi-user.target

/etc/systemd/system/canal-adapter.service 同理，三处差异：

ini
[Unit]
After=network-online.target docker.service canal-server.service
Wants=network-online.target canal-server.service

[Service]
PIDFile=/opt/canal/adapter/bin/adapter.pid
ExecStart=/opt/canal/adapter/bin/startup.sh
ExecStop=/opt/canal/adapter/bin/stop.sh

启用：

bash
systemctl daemon-reload
systemctl enable --now canal-server canal-adapter
systemctl status canal-server canal-adapter

三个关键点：

1. Environment=JAVA_HOME 必须写。systemd 不读 /etc/profile.d/，不写就找不到 java。
2. Type=forking + PIDFile 要匹配脚本。Canal 的 startup.sh 是 java ... & 后台启动并写 bin/canal.pid，属于典型 fork 型。
3. adapter 要 After=canal-server。否则开机时 adapter 可能先起来，连不上 server 就一直重试。

---

七、运维命令速查

bash
# ---------- 服务管理 ----------
systemctl status canal-server canal-adapter
systemctl restart canal-adapter
systemctl stop canal-server
systemctl disable canal-adapter        # 取消开机自启

# ---------- 日志 ----------
tail -f /opt/canal/server/logs/example/example.log     # binlog 解析
tail -f /opt/canal/adapter/logs/adapter/adapter.log    # 同步写 ES
tail -f /opt/canal/server/logs/canal/canal.log         # 框架日志

# ---------- 检查 binlog 连接 ----------
mysql -h <MYSQL_HOST> -ucanal -p -e "SHOW PROCESSLIST;" | grep "Binlog Dump"

# ---------- 手动触发全量 ----------
curl -X POST "http://127.0.0.1:8081/etl/es7/<映射文件名>.yml"

# ---------- 查 ES 数据 ----------
curl -u 'elastic:<ES_PASSWORD>' "http://127.0.0.1:9200/<index>/_count"
curl -u 'elastic:<ES_PASSWORD>' "http://127.0.0.1:9200/<index>/_search?pretty&size=3"

# ---------- 强制刷新 ----------
curl -u 'elastic:<ES_PASSWORD>' -X POST "http://127.0.0.1:9200/<index>/_refresh"

# ---------- canal 指标（Prometheus 格式） ----------
curl -s http://127.0.0.1:11112/metrics | head -30

---

附录：完整配置文件

conf/example/instance.properties

properties
canal.instance.gtidon = false

canal.instance.master.address = <MYSQL_HOST>:3306
canal.instance.master.journal.name = mysql-bin.000048
canal.instance.master.position = 4059448
canal.instance.master.timestamp =
canal.instance.master.gtid =

canal.instance.tsdb.enable = false

canal.instance.dbUsername = canal
canal.instance.dbPassword = <CANAL_DB_PASSWORD>
canal.instance.connectionCharset = UTF-8
canal.instance.enableDruid = false

canal.instance.filter.regex = nfturbo\\..*
canal.instance.filter.black.regex =

canal.mq.topic = example
canal.mq.partition = 0

conf/application.yml

yaml
server:
  port: 8081

canal.conf:
  mode: tcp
  flatMessage: true
  zookeeperHosts:
  syncBatchSize: 1000
  retries: -1
  timeout:
  accessKey:
  secretKey:
  consumerProperties:
    canal.tcp.server.host: 127.0.0.1:11111
    canal.tcp.zookeeper.hosts:
    canal.tcp.batch.size: 500
    canal.tcp.username:
    canal.tcp.password:
  srcDataSources:
    defaultDS:
      url: jdbc:mysql://<MYSQL_HOST>:3306/nfturbo?useUnicode=true&characterEncoding=UTF-8&useSSL=false&serverTimezone=Asia/Shanghai&allowPublicKeyRetrieval=true
      username: canal
      password: '<CANAL_DB_PASSWORD>'
  canalAdapters:
  - instance: example
    groups:
    - groupId: g1
      outerAdapters:
      - name: es7
        hosts: http://127.0.0.1:9200
        properties:
          mode: rest
          cluster.name: <ES_CLUSTER_NAME>
          security.auth: 'elastic:<ES_PASSWORD>'

conf/es7/nfturbo_collection.yml

yaml
dataSourceKey: defaultDS
destination: example
groupId: g1
esMapping:
  _index: nfturbo_collection
  _id: _id
  sql: "select id as _id, gmt_create, gmt_modified, name, cover, class_id, price,
        quantity, detail, saleable_inventory, identifier, state, create_time,
        creator_id, version from collection"
  commitBatch: 1000

---

下一篇：8 个踩坑记录与生产环境注意事项

涉及：MySQL 8.4 的 SHOW MASTER STATUS、adapter 自带驱动太老、es7 居然默认走 TransportClient、application.yml 少一行就 NPE、空索引导致的怪异常——每个坑都有完整报错原文和排查过程。