一、 基础环境

        1.canal ：我的是1.1.8的版本，只需要下载canal.deployer-1.1.8就可以，因为分库分表写入es是放在java代码里实现，canal下载canal地址 https://github.com/alibaba/canal/releases 。

        2.es：我的是7.10.2 版本，下载地址Elasticsearch 7.10.2 | Elastic  ，ik分词器已上传到文章里放在elasticsearch-7.10.2\plugins\并解压。

        3.jdk：推荐使用jdk11，我的版本是17所以启动的脚本都要去单独修改。

二、环境启动配置

        1.es：配置elasticsearch.yml，路径在 elasticsearch-7.10.2/config/elasticsearch.yml

# 集群名称，自定义你的 Elasticsearch 集群标识
cluster.name: my-es-pc-star

# 当前节点名称，可自定义命名以区分多个节点
node.name: node-1

# 数据保存目录（这里配置为 Windows 路径，请确保该路径存在并有写权限）
path.data: D:/java/Es/elasticsearch-7.10.2/data

# 日志保存目录（包含启动日志、GC 日志等）
path.logs: D:/java/Es/elasticsearch-7.10.2/logs

# 设置监听的主机地址（0.0.0.0 表示监听所有网卡，允许远程访问）
network.host: 0.0.0.0

# HTTP 服务端口（默认 9200）
http.port: 9209

# 设置 HTTP 接收请求体的最大大小，用于支持较大的文档上传（默认是 100mb）
http.max_content_length: 100mb

# 锁定内存，防止内存交换到磁盘，提升性能（需要系统配置支持，例如 Windows 需管理员身份启动）
bootstrap.memory_lock: true

# 防止误操作（如 delete *）删除所有索引，必须显式指定索引名
action.destructive_requires_name: false

# 单节点模式（不启用自动发现机制，适用于本地或单机环境）
discovery.type: single-node

# 允许Canal创建相关的所有索引
action.auto_create_index: "*"  

es启动脚本示例，路径在elasticsearch-7.10.2/bin/自行创建脚本：

#!/bin/bash

# =======================
# 自定义 Elasticsearch 启动脚本
# 放在 elasticsearch 目录下（与 bin/ 同级）
# chmod +x start-es.sh 后执行 ./start-es.sh -s
# =======================

ES_HOME=/opt/home/environment/es/elasticsearch-7.10.2
ES_BIN="$ES_HOME/bin/elasticsearch"
ES_JAVA_OPTS="-Xms2g -Xmx2g"
ES_PORT=9209

check_password=true
silent=false

for arg in "$@"; do
  case $arg in
    -s|--silent)
      silent=true
      ;;
    -h|--help|-V|--version)
      check_password=false
      ;;
  esac
done

# 模拟 keystore 密码输入（如果需要）
if [ "$check_password" = true ]; then
  echo -n "请输入 Elasticsearch keystore 密码: "
  read -s KEYSTORE_PASSWORD
  echo ""
  # 你可以在这里使用 KEYSTORE_PASSWORD 做进一步操作
fi

# 启动 ES（可加参数）
echo "Starting Elasticsearch..."
$ES_BIN -Ehttp.port=$ES_PORT -Epath.data="$ES_HOME/data" -Epath.logs="$ES_HOME/logs" &
PID=$!

# 等待启动
sleep 5
if ps -p $PID > /dev/null; then
  echo "Elasticsearch 启动成功，PID=$PID"
else
  echo "Elasticsearch 启动失败"
  if [ "$silent" = false ]; then
    read -p "按任意键退出..." pause
  fi
  exit 1
fi

        2.canal：配置instance.properties，路径在 canal.deployer-1.1.8\conf\example\instance.properties

注意：里面配置的regex改成自己业务需要导入es的分库 或者分表 

# 是否开启 GTID 模式，MySQL 如果启用了 GTID（推荐关闭除非你了解）
canal.instance.gtidon=false

# 每个 canal 实例在 MySQL 的唯一标识，不能重复
canal.instance.mysql.slaveId=1001

# MySQL 主库的地址（主机:端口）
canal.instance.master.address=127.0.0.1:4461

# 连接 MySQL 的账号密码（必须具备 REPLICATION 权限）
canal.instance.dbUsername=
canal.instance.dbPassword=

# 字符集，通常为 UTF-8
canal.instance.connectionCharset=UTF-8

# 启用 tsdb（增量订阅位点的版本管理）
canal.instance.tsdb.enable=false

# TSDB 数据存储目录（Windows 路径注意双反斜杠或使用 /）
canal.instance.tsdb.directory=/opt/home/environment/canal/canal-tsdb

# binlog 正则过滤规则（支持正则表达式）
# 匹配所有以 tg_ 开头但不包含 tg_mob_ 的数据库中的所有表
canal.instance.filter.regex=(tg_(?!mob_)[^\\.]+\\..*|ry-star\\.star_user)

# 是否启用 Druid 数据源（一般不启用）
canal.instance.enableDruid=false

canal启动脚本改造示例，路径在canal.deployer-1.1.8/bin/restart.sh：

#!/bin/bash

# Echo 控制（Linux 下默认开启）
# echo 开启可用 set -x，调试用

# 取当前脚本所在目录
ENV_PATH=$(cd "$(dirname "$0")"; pwd)
CONF_DIR="$ENV_PATH/../conf"

# 默认配置文件
CANAL_CONF="$CONF_DIR/canal.properties"
if [ "$1" == "local" ]; then
  CANAL_CONF="$CONF_DIR/canal_local.properties"
fi

LOGBACK_CONFIGURATION_FILE="$CONF_DIR/logback.xml"

# classpath 配置，所有 jar 包目录
CLASSPATH="$CONF_DIR:$ENV_PATH/../lib/*"

# JVM 参数配置
JAVA_MEM_OPTS="-Xms8g -Xmx8g"
JAVA_OPTS_EXT="-Djava.awt.headless=true -Djava.net.preferIPv4Stack=true -Dapplication.codeset=UTF-8 -Dfile.encoding=UTF-8"
JAVA_DEBUG_OPT="-server -Xdebug -Xnoagent -Djava.compiler=NONE -Xrunjdwp:transport=dt_socket,address=9099,server=y,suspend=n"
CANAL_OPTS="-DappName=otter-canal -Dlogback.configurationFile=$LOGBACK_CONFIGURATION_FILE -Dcanal.conf=$CANAL_CONF"

JAVA_OPTS="$JAVA_MEM_OPTS $JAVA_OPTS_EXT $JAVA_DEBUG_OPT $CANAL_OPTS"

CMD_STR="java $JAVA_OPTS -classpath \"$CLASSPATH\" com.alibaba.otter.canal.deployer.CanalLauncher"
echo "start cmd : $CMD_STR"

# 启动 Canal
exec java $JAVA_OPTS -classpath "$CLASSPATH" com.alibaba.otter.canal.deployer.CanalLauncher

注意在canal.deployer-1.1.8同级目录下创建canal-tsdb 数据存储目录

然后先启动es 在启动canal

三、配置es模版

        因为我的是分库分表，所以要把所有相同的规则索引套用一样的索引模版，例如  数据库1、2、3下的同一张表chatlist表，创建索引就会自动使用同一个模版（根据索引名称去自动套用比如我在模版中配置"template": "chatlist_index_*"，那么只要创建的索引前缀和chatlist_index_匹配就会自动套用这套模版例如chatlist_index_1或者chatlist_index_1_1111都可以自动适配），下面是我的一个不同库，同一张表所使用的索引模版demo，根据需求自己设置模版

PUT http://127.0.0.1:9209/_template/chatlist_index
{
  "template": "chatlist_index_*",
  "settings": {
    "index.max_ngram_diff": 8,
    "number_of_shards": 1,
    "number_of_replicas": 1,
    "analysis": {
      "filter": {
        "lowercase_filter": {
          "type": "lowercase"
        }
      },
      "analyzer": {
        "custom_ngram_analyzer": {
          "tokenizer": "custom_ngram_tokenizer",
          "filter": [
            "lowercase_filter"
          ]
        },
        "keyword_search": {
          "tokenizer": "keyword",
          "filter": [
            "lowercase_filter"
          ]
        }
      },
      "tokenizer": {
        "custom_ngram_tokenizer": {
          "type": "ngram",
          "min_gram": 2,
          "max_gram": 10,
          "token_chars": [
            "letter",
            "digit"
          ]
        }
      }
    }
  },
  "mappings": {
    "properties": {
      "id": {
        "type": "long"
      },
      "user_id": {
        "type": "keyword"
      },
      "name": {
        "type": "text",
        "analyzer": "ik_max_word",
        "fields": {
          "keyword": {
            "type": "keyword"
          },
          "substring": {
            "type": "text",
            "analyzer": "custom_ngram_analyzer",
            "search_analyzer": "keyword_search"
          }
        }
      },
      "user_name": {
        "type": "text",
        "analyzer": "ik_max_word",
        "fields": {
          "keyword": {
            "type": "keyword"
          },
          "substring": {
            "type": "text",
            "analyzer": "custom_ngram_analyzer",
            "search_analyzer": "keyword_search"
          }
        }
      },
      "phone": {
        "type": "text",
        "analyzer": "ik_max_word",
        "fields": {
          "keyword": {
            "type": "keyword"
          },
          "substring": {
            "type": "text",
            "analyzer": "custom_ngram_analyzer",
            "search_analyzer": "keyword_search"
          }
        }
      },
      "type": {
        "type": "keyword"
      },
      "tgid": {
        "type": "text",
        "analyzer": "ik_max_word",
        "fields": {
          "keyword": {
            "type": "keyword"
          },
          "substring": {
            "type": "text",
            "analyzer": "custom_ngram_analyzer",
            "search_analyzer": "keyword_search"
          }
        }
      },
      "uuid": {
        "type": "keyword"
      },
      "file_id": {
        "type": "keyword"
      },
      "record": {
        "type": "text"
      },
      "last_time": {
        "type": "date",
        "format": "yyyy-MM-dd HH:mm:ss||strict_date_optional_time||epoch_millis"
      },
      "create_by": {
        "type": "keyword"
      },
      "create_time": {
        "type": "date",
        "format": "yyyy-MM-dd HH:mm:ss||strict_date_optional_time||epoch_millis"
      },
      "update_by": {
        "type": "keyword"
      },
      "update_time": {
        "type": "date",
        "format": "yyyy-MM-dd HH:mm:ss||strict_date_optional_time||epoch_millis"
      }
    }
  }
}

四、java代码写入ES 并设置索引名称为了套用到索引模版

package com.ruoyi.star.listener;


import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry.*;
import com.alibaba.otter.canal.protocol.Message;
import com.ruoyi.star.config.CanalProperties;
import lombok.RequiredArgsConstructor;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Controller;

import javax.annotation.PostConstruct;
import java.net.InetSocketAddress;
import java.time.Instant;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.*;


@Component
@RequiredArgsConstructor
public class CanalListener {
    @Autowired
    private CanalProperties canalProperties;
    private final RestHighLevelClient esClient;

    @PostConstruct
    public void start() {
        System.out.println(">>> CanalListener 已启动监听线程");
        System.out.println(">>> CanalConnector 已连接：" + canalProperties.getHost() + ":" + canalProperties.getPort());
        new Thread(this::runListener).start();
    }

    public void runListener() {
        CanalConnector connector = CanalConnectors.newSingleConnector(
                new InetSocketAddress(canalProperties.getHost(), canalProperties.getPort()),
                canalProperties.getDestination(), "", ""
        );

        connector.connect();
        connector.subscribe(canalProperties.getSubscribe());
        connector.rollback();

        while (true) {
            Message message = connector.getWithoutAck(canalProperties.getBatchSize());
            long batchId = message.getId();
            List<Entry> entries = message.getEntries();
            if (entries.isEmpty()) {
                sleep(1000);
                continue;
            }

            try {
                for (Entry entry : entries) {
                    if (entry.getEntryType() != EntryType.ROWDATA) continue;
                    RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
                    EventType eventType = rowChange.getEventType();
                    String tableName = entry.getHeader().getTableName();
                    for (RowData rowData : rowChange.getRowDatasList()) {
                        switch (eventType) {
                            case INSERT, UPDATE -> handleInsertOrUpdate(tableName, rowData.getAfterColumnsList());
                            case DELETE -> handleDelete(tableName, rowData.getBeforeColumnsList());
                        }
                    }
                }
                connector.ack(batchId);
            } catch (Exception e) {
                e.printStackTrace();
                connector.rollback(batchId);
            }
        }
    }

    private void handleInsertOrUpdate(String tableName, List<Column> columns) throws Exception {
        // 将列数据转换成 Map
        Map<String, Object> data = toMap(columns);
        // 根据表名和数据生成 ES 索引名
        String index = getIndexName(tableName, columns);
        // 根据表名和数据生成文档 ID
        String docId = getDocumentId(tableName, columns);

        // 定义所有可能是日期类型的字段名称集合
        Set<String> dateFields = Set.of("create_time", "update_time", "last_active", "created", "date", "last_time", "data_updating_time");

        // 移除所有日期字段中值为 null、空字符串或无效时间字符串 "0000-00-00 00:00:00" 的字段，避免 ES 日期格式解析失败
        data.entrySet().removeIf(entry -> {
            if (!dateFields.contains(entry.getKey())) {
                return false; // 非日期字段不处理
            }
            Object val = entry.getValue();
            // 过滤条件：值为 null，空字符串，或 "0000-00-00 00:00:00"
            return val == null || "".equals(val) || "0000-00-00 00:00:00".equals(val);
        });

        // 构造索引请求，写入或更新 ES 文档
        IndexRequest request = new IndexRequest(index).id(docId).source(data, XContentType.JSON);
        esClient.index(request, RequestOptions.DEFAULT);
    }


    private void handleDelete(String tableName, List<Column> columns) throws Exception {
        String index = getIndexName(tableName, columns);
        String docId = getDocumentId(tableName, columns);

        DeleteRequest request = new DeleteRequest(index, docId);
        esClient.delete(request, RequestOptions.DEFAULT);
    }

    private Map<String, Object> toMap(List<Column> columns) {
        Map<String, Object> map = new HashMap<>();
        for (Column col : columns) {
            map.put(col.getName(), col.getValue());
        }
        return map;
    }

    private String getIndexName(String tableName, List<Column> columns) {
        String suffix = "unknown";
        String createTime = null;

        for (Column col : columns) {
            if ("create_time".equalsIgnoreCase(col.getName())) {
                createTime = col.getValue();
                break;
            }
        }
        if (useYearMonth(tableName)) {
            suffix = extractYearMonthAndWeekOfMonth(createTime);
        } else {
            suffix = extractYear(createTime);
        }
        return tableName.toLowerCase() + "_index_" + suffix;
    }

    // 判断是否是需要按 yyyyMM wN的表
    private boolean useYearMonth(String tableName) {
        Set<String> monthlyTables = Set.of(
                "chatlist", "contacts", "file", "location", "media", "message", "sessions", "textentities"
        );
        return monthlyTables.contains(tableName.toLowerCase());
    }

    /**
     * 提取日期字符串或时间戳对应的“年月+月内第几周”后缀，格式为 yyyyMM_wN。
     * 其中：
     *  - yyyyMM 表示年和月，例如202508
     *  - wN 表示该日期所在月份的第N周，N从1开始计算，周一到周日为一周
     * 支持输入为：
     *  - 10位或13位时间戳字符串（秒或毫秒）
     *  - 常见日期字符串格式，如 "yyyy-MM-dd HH:mm:ss", "yyyy/MM/dd HH:mm:ss", "yyyy-MM-dd", "yyyy/MM/dd"
     * 如果解析失败，返回 "unknown"。
     *
     * @param createTime 日期字符串或时间戳字符串
     * @return 格式化后的年月+月内周字符串，例如 "202508_w3"
     */
    private String extractYearMonthAndWeekOfMonth(String createTime) {
        if (createTime == null || createTime.isEmpty()) return "unknown";

        try {
            // 尝试将字符串解析为时间戳（秒或毫秒）
            long ts = Long.parseLong(createTime);
            if (ts < 10000000000L) ts *= 1000;  // 秒转毫秒
            Instant instant = Instant.ofEpochMilli(ts);
            LocalDate dt = LocalDateTime.ofInstant(instant, ZoneId.systemDefault()).toLocalDate();

            return formatYearMonthWeek(dt);
        } catch (NumberFormatException e) {
            // 不是纯数字，尝试用日期格式解析
            DateTimeFormatter[] fmts = {
                    DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"),
                    DateTimeFormatter.ofPattern("yyyy/MM/dd HH:mm:ss"),
                    DateTimeFormatter.ofPattern("yyyy-MM-dd"),
                    DateTimeFormatter.ofPattern("yyyy/MM/dd")
            };
            for (DateTimeFormatter fmt : fmts) {
                try {
                    LocalDate dt = LocalDate.parse(createTime, fmt);
                    return formatYearMonthWeek(dt);
                } catch (Exception ignore) {
                    // 格式不匹配，继续尝试下一个格式
                }
            }
        }
        // 解析失败返回默认值
        return "unknown";
    }
    /**
     * 格式化年月和月内第几周，格式为 yyyyMM_wN
     * @param dt LocalDate日期对象
     * @return 格式化字符串，例如 "202508_w3"
     */
    private String formatYearMonthWeek(LocalDate dt) {
        String yearMonth = String.format("%04d%02d", dt.getYear(), dt.getMonthValue());

        // 计算“月内第几周”
        int dayOfMonth = dt.getDayOfMonth();
        LocalDate firstDayOfMonth = dt.withDayOfMonth(1);
        // Java中 getDayOfWeek() 返回 MONDAY=1 ... SUNDAY=7
        int firstWeekDay = firstDayOfMonth.getDayOfWeek().getValue();

        // 计算周数： (当前日 + 第一天是周几 - 1) / 7 向上取整
        int weekOfMonth = (int) Math.ceil((dayOfMonth + firstWeekDay - 1) / 7.0);

        return yearMonth + "_w" + weekOfMonth;
    }
    // 提取年 yyyy（支持时间戳或字符串）
    private String extractYear(String createTime) {
        if (createTime == null || createTime.isEmpty()) return "unknown";
        try {
            long ts = Long.parseLong(createTime);
            if (ts < 10000000000L) ts *= 1000;
            return LocalDate.ofEpochDay(ts / 86400000).getYear() + "";
        } catch (NumberFormatException e) {
            DateTimeFormatter[] fmts = {
                    DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"),
                    DateTimeFormatter.ofPattern("yyyy/MM/dd HH:mm:ss"),
                    DateTimeFormatter.ofPattern("yyyy-MM-dd"),
                    DateTimeFormatter.ofPattern("yyyy/MM/dd")
            };
            for (DateTimeFormatter fmt : fmts) {
                try {
                    return LocalDate.parse(createTime, fmt).getYear() + "";
                } catch (Exception ignore) {
                }
            }
        }
        return "unknown";
    }

    private String getDocumentId(String tableName, List<Column> columns) {
        Map<String, String> map = new HashMap<>();
        for (Column c : columns) {
            map.put(c.getName(), c.getValue());
        }

        return switch (tableName.toLowerCase()) {
            case "star_user" -> map.getOrDefault("tgid", "null") + "_" + map.getOrDefault("id", "null");
            case "sessions" ->
                    map.getOrDefault("tgid", "null") + "_" + map.getOrDefault("uuid", "null") + "_" + map.getOrDefault("id", "null");
            case "textentities" ->
                    map.getOrDefault("chatlist_id", "null") + "_" + map.getOrDefault("message_id", "null") + "_" + map.getOrDefault("id", "null");
            case "chatlist" ->
                    map.getOrDefault("tgid", "null") + "_" + map.getOrDefault("uuid", "null") + "_" + map.getOrDefault("id", "null");
            case "contacts" ->
                    map.getOrDefault("tgid", "null") + "_" + map.getOrDefault("uuid", "null") + "_" + map.getOrDefault("id", "null");
            case "media" ->
                    map.getOrDefault("chatlist_id", "null") + "_" + map.getOrDefault("message_id", "null") + "_" + map.getOrDefault("id", "null");
            case "location" ->
                    map.getOrDefault("chatlist_id", "null") + "_" + map.getOrDefault("message_id", "null") + "_" + map.getOrDefault("id", "null");
            case "file" -> map.getOrDefault("file_id", "null") + "_" + map.getOrDefault("id", "null");
            case "message" ->
                    map.getOrDefault("chatlist_id", "null") + "_" + map.getOrDefault("message_id", "null") + "_" + map.getOrDefault("id", "null");
            default -> map.getOrDefault("id", "null");
        };
    }

    private void sleep(long ms) {
        try {
            Thread.sleep(ms);
        } catch (InterruptedException ignored) {
        }
    }
}

package com.ruoyi.star.config;

import lombok.Data;
import org.apache.http.HttpHost;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.util.Map;

@Data
@Component
/**
 * 获取到nacos 中yml的es配置
 */
public class ElasticsearchConfig {
    @Value("${spring.elasticsearch.uris}")
    private String elasticsearchUris;

    @Bean
    public RestHighLevelClient restHighLevelClient() {
        return new RestHighLevelClient(
                RestClient.builder(HttpHost.create(elasticsearchUris))
        );
    }
}

package com.ruoyi.star.config;

import lombok.Data;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

@Data
@Component
/**
 * 获取到nacos 中yml的canal配置
 */
public class CanalProperties {

    @Value("${spring.canal.host}")
    private String host;
    @Value("${spring.canal.port}")
    private int port;
    @Value("${spring.canal.destination}")
    private String destination;
    @Value("${spring.canal.subscribe}")
    private String subscribe;
    @Value("${spring.canal.batchSize}")
    private int batchSize;


    @Value("${spring.canal.jdbcurl}")
    private String jdbcurl;
    @Value("${spring.canal.username}")
    private String username;
    @Value("${spring.canal.password}")
    private String password;
}

  canal:
    host: 127.0.0.1
    port: 11111
    destination: example
    subscribe: "(tg_(?!mob_)[^\\.]+\\..*|ry-star\\.star_user)"
    batchSize: 100
    jdbcurl: jdbc:mysql://127.0.0.1:3306/ry-star?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=true&serverTimezone=GMT%2B8
    username: root
    password: 123456

其中

@Value("${spring.elasticsearch.uris}")
private String elasticsearchUris;

@Autowired
private CanalProperties canalProperties;

里面配置了es 端口 之类的配置项 可以自行配置

都是我再nacos的canal和es配置端口之类的 可以自行配置