做数据同步的应该都遇到过这种场面：业务方跑过来说，这张表同步过去之后手机号得脱敏，不然下游报表一拉全是明文；过两天又来说，得带上来源标记，不然出了问题没人知道数据是哪个库写进来的；再过一周说，某些状态的行别同步了，看着心烦。这类需求用 Otter 里那套字段映射、过滤条件的配置化能力基本没戏，绕来绕去最后还是得写代码。Otter 给的口子就是Event Processor——一个允许你在 ETL 链路上插手、对每一条同步事件做二次加工的扩展点。
这篇不聊概念，只聊怎么写、怎么挂、怎么保证它别把整条同步链路带崩。Event Processor 自定义写法的核心就三件事：搞清楚它在链路里的位置、写对接口实现、用正确的方式注册和部署。但真正让人栽跟头的是第四件事——它在异常、幂等、并发这三块上的行为，和大多数人想的不一样。适合正在用 Otter 做库到库同步、被原生配置能力卡住、准备写第一个自定义处理器的同学看，也适合已经写了但发现线上数据不对劲的同学回头对一遍。
1. 先搞清楚 Otter 的 ETL 链路，再谈扩展点
1.1 一条 binlog 在 Otter 里要走多远
Otter 的 Node 端干的事就是把源库的增量日志抽出来、转一道、再写进去。整个流程拆成三段：Extract负责从源库拉 binlog 并解析成内部事件对象，Transform负责按照你在管理端配的映射规则做字段映射、表名映射、简单的行过滤，Load负责把加工后的事件翻译成目标库能执行的 SQL 并落库。
这三段里，Extract 是纯搬运，Transform 是配置驱动，Load 是执行。Event Processor 在最后一段的入口插进去，也就是 Transform 已经把字段映射做完、数据即将被翻译成 INSERT/UPDATE/DELETE 的那一刻。这个位置很关键，因为它意味着你拿到的Event对象里的列名已经是目标表的名字，不是源表的；字段值也已经是映射后的结果。
理解这一点能省掉很多无谓的调试。我见过有人在处理器里按源表字段名user_mobile去找列，结果死活找不到，因为映射规则里目标字段叫mobile。处理器看到的是"即将写入目标库的样子"，不是"从源库读出来的样子"。
还有个容易忽略的点：一个 Event 里可能不止一行数据。Otter 是按批次搬运的，一次事件里的rowDatas可能包含几十上百条变更行。所以处理器里必须是循环遍历，不能想当然按单行处理。刚上手时写个event.getRowDatas().get(0)就觉得完事了，批量场景下直接漏数据，而且漏得悄无声息。
1.2 Event Processor 能做什么，边界在哪
它的能力边界其实挺清晰的：你可以任意读写事件对象里的内容，但不能改变事件在链路中的走向。具体来说能干的事包括改字段值、加字段、删字段、删除某些行数据、调整事件类型（有风险，后面细说），不能干的事包括改同步位点、跳过整个 pipeline、动态改映射规则。
这个"能改数据不能改流程"的设定其实是对的。数据加工是业务需求，流程控制是同步系统自己的事，混在一起迟早出事。所以如果你的需求是"这个表干脆别同步了"，应该去管理端配过滤规则，而不是在处理器里把行全删了——后者虽然也能实现效果，但每条数据都要走一遍 ETL 再到你这里被扔掉，白白消耗资源。
我一般把 Event Processor 的适用场景归成四类：字段级加工（脱敏、加密、枚举字典转换、时间格式统一、金额单位换算）、行级标记（补来源实例、补同步时间、补批次号、补租户 ID）、行级过滤（按业务条件丢弃部分变更，比如只同步有效状态的数据）、跨表分发（同一份变更同时写到多个目标表，或者按分片键路由到不同目标表）。
第三类和第四类要谨慎，尤其第四类。跨表分发意味着你要复制事件对象，而事件对象里的RowData、Column都是可变对象，浅拷贝会导致两边互相污染，改一边另一边跟着变。要复制就得深拷贝，这个成本不低，而且容易漏字段。
1.3 和 RowDataFilter、LoadDataHandler 的选型对比
Otter 的扩展点不止 Event Processor 一个，选错了后面很难受。我按实际项目里常见的几种需求做了个对照：
| 扩展方式 | 介入时机 | 能力范围 | 适合场景 | 主要限制 |
|---|---|---|---|---|
| 管理端映射与过滤配置 | Transform 阶段 | 字段映射、表名映射、条件过滤 | 简单的字段改名、整表过滤 | 无法做值计算，无法加常量以外的字段 |
| Event Processor | Load 之前的入口 | 任意读写事件对象 | 脱敏、打标、复杂过滤、行复制 | 单例、同步执行、无事务 |
| 自定义 Load Handler | Load 执行阶段 | 干预 SQL 生成与执行 | 特殊 SQL 方言适配、写入拦截 | 侵入更深，升级容易冲突 |
| 目标端触发器/存储过程 | 落库之后 | 库内任意逻辑 | 简单的审计、统计 | 难维护，排查链路长，性能差 |
绝大多数场景 Event Processor 就够了。它的优势是位置靠后但不越界：数据已经映射好，你只管加工，不用担心影响映射逻辑；同时它又在 SQL 生成之前，改完就能生效，不需要碰更底层的执行环节。
只有一种情况我会推荐往下走一层：目标库的写入方式本身要变，比如需要把多条变更合并成一条 upsert，或者需要按特定顺序批量提交。这种属于写入策略问题，塞进 Event Processor 里做会很难看。
2. EventProcessor 接口的骨架与自定义实现
2.1 先把接口签名核对清楚
Otter 的版本迭代过程中，扩展点接口的包路径和方法签名有过调整，直接照抄网上的代码大概率编译不过。动手之前，花两分钟从你自己的 jar 里把接口定义挖出来：
# 先定位 node 安装目录下的 jar ls /data/otter/node/lib | grep -i otter # 从 jar 里反编译看接口签名 javap -classpath /data/otter/node/lib/otter-node-*.jar \ com.alibaba.otter.node.extend.processor.EventProcessor如果包路径不对，javap会报找不到类，这时候用unzip -l把 jar 里的类列表打出来搜一下：
unzip -l /data/otter/node/lib/otter-node-*.jar | grep -i processor接口大致长这样，不同版本可能是返回boolean也可能是void，方法参数可能多一个上下文对象：
public interface EventProcessor { /** * 处理一条同步事件 * @return true 表示继续后续流程，false 表示丢弃该事件 */ boolean process(Event event); }我的建议是不要依赖返回值的语义。返回值在不同版本里含义不一致，有的版本干脆是void。想做行过滤就直接操作rowDatas列表，把不要的行remove掉或者重新set一个新列表，这个做法在所有版本上行为一致。
2.2 一个最小可用的实现：给变更行打来源标记
先写个能跑通、能验证链路的最小版本。目标很简单：给每一条同步过去的行补上两个字段——同步时间戳和来源实例标识。这类需求在线多库汇聚的场景里几乎是标配，出了问题靠这两个字段能快速定位是哪个源。
package com.example.otter.processor; import com.alibaba.otter.node.extend.processor.EventProcessor; import com.alibaba.otter.shared.etl.model.Column; import com.alibaba.otter.shared.etl.model.Event; import com.alibaba.otter.shared.etl.model.EventType; import com.alibaba.otter.shared.etl.model.RowData; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.ArrayList; import java.util.Date; import java.util.List; public class SourceTagEventProcessor implements EventProcessor { private static final Logger logger = LoggerFactory.getLogger(SourceTagEventProcessor.class); /** 来源标识，通过配置注入，不要写死 */ private String sourceTag = "unknown"; @Override public boolean process(Event event) { if (event == null || event.getRowDatas() == null || event.getRowDatas().isEmpty()) { return true; } EventType type = event.getEventType(); // 只处理 DML，DDL 直接放过 if (type != EventType.INSERT && type != EventType.UPDATE && type != EventType.DELETE) { return true; } for (RowData rowData : event.getRowDatas()) { // INSERT / UPDATE 走 after，DELETE 只有 before List<Column> target = rowData.getAfterColumns(); if (target == null || target.isEmpty()) { target = rowData.getBeforeColumns(); } if (target == null) { continue; } putColumn(target, "sync_time", new Date()); putColumn(target, "sync_source", sourceTag); } logger.debug("tagged event, table={}, rows={}", event.getTableName(), event.getRowDatas().size()); return true; } private void putColumn(List<Column> columns, String name, Object value) { for (Column c : columns) { if (name.equalsIgnoreCase(c.getName())) { c.setValue(value); return; } } Column col = new Column(); col.setName(name); col.setValue(value); col.setIsNull(false); columns.add(col); } public void setSourceTag(String sourceTag) { this.sourceTag = sourceTag; } }两个细节值得单独说。第一，DELETE 事件必须用beforeColumns。删除操作在 binlog 里只有变更前的镜像，afterColumns是空的。新手最常犯的错就是统一取afterColumns，结果删除同步全丢，而且日志里看不出来——因为代码没报错，只是拿了个空列表然后continue过去了。
第二，新增字段要检查目标表里有没有这一列。putColumn往列表里塞一个新Column，最终生成的 SQL 里就会带上这个字段名。目标表如果没有sync_time这一列，落库直接报Unknown column。所以要么先改表结构，要么把putColumn里的新增逻辑注释掉，只做已有字段的赋值。
2.3 进阶：字段级加工的正确姿势
脱敏是最高频的需求。写起来简单，但有三个坑：判空、长度校验、类型对齐。
private static final Pattern MOBILE = Pattern.compile("^1\\d{10}$"); private void maskSensitive(List<Column> columns) { for (Column c : columns) { if (c.getValue() == null) { continue; } String name = c.getName() == null ? "" : c.getName().toLowerCase(); if ("mobile".equals(name) || "phone".equals(name)) { String v = String.valueOf(c.getValue()); if (MOBILE.matcher(v).matches()) { c.setValue(v.substring(0, 3) + "****" + v.substring(7)); } } else if ("id_card".equals(name)) { String v = String.valueOf(c.getValue()); if (v.length() == 18) { c.setValue(v.substring(0, 6) + "**********" + v.substring(16)); } } } }Pattern提到静态常量里预编译，别在循环里String.matches()。一个批次几百行、每秒几十个批次，正则编译的开销很快就上来了。这个优化点单次看不出来，压测的时候能差出好几毫秒的 P99。
再说类型对齐。Column.setValue()接受Object，但最终写库时是按列的 JDBC 类型走的。你把一个String塞进原本是Integer的列，有的驱动会隐式转换，有的直接抛ClassCastException。稳妥做法是保持原类型：原来是Integer的就传Integer，要改成别的值就用Integer.valueOf()转换，别图省事直接传字符串。
还有一种更隐蔽的类型问题：源库datetime字段在事件里可能是java.util.Date，也可能是java.sql.Timestamp，取决于驱动版本。如果你的加工逻辑要比较时间，先统一转成long时间戳再比，别直接equals。
2.4 用策略模式管住膨胀的处理器
第一个处理器写完很清爽，三个月后它会变成一坨三百行的 if-else：按表名分流、按字段名判断、各种业务特判堆在一起。改一处提心吊胆，因为不知道会影响哪条链路。
我的做法是从第二个需求开始就上策略：一个总入口负责按表名/事件类型分发，具体的加工逻辑拆到独立的处理器类里。这样每加一个需求只新增一个类，不动存量代码。
public class DispatchingProcessor implements EventProcessor { private Map<String, EventProcessor> tableProcessorMap = new HashMap<>(); private EventProcessor defaultProcessor; @Override public boolean process(Event event) { EventProcessor target = tableProcessorMap.get(event.getTableName()); if (target == null) { target = defaultProcessor; } if (target == null) { return true; } return target.process(event); } public void setTableProcessorMap(Map<String, EventProcessor> map) { this.tableProcessorMap = map; } public void setDefaultProcessor(EventProcessor p) { this.defaultProcessor = p; } }映射关系放在 Spring 配置里维护，改分流规则不用重新编译，重启节点就行。表名注意大小写，Otter 里表名的大小写取决于源库配置和 binlog 里的原始大小写，做映射前先打日志确认一下实际值，别想当然按小写处理。
3. 把实现挂进 Otter Node：配置、打包与验证
3.1 找到注入点再动手改配置
Node 端的配置目录一般是conf/，里面有几个 XML 文件。不同发行包的文件名和结构有差异，别硬记，直接搜：
cd /data/otter/node grep -rn "EventProcessor" conf/ | head -20搜出来的一般是一个列表型的 bean 定义，或者某个 bean 上有个eventProcessors属性。这就是你的注入点。找到之后按现有格式追加：
<!-- 你自己的处理器 --> <bean id="sourceTagProcessor" class="com.example.otter.processor.SourceTagEventProcessor"> <property name="sourceTag" value="order-db-shanghai-01"/> </bean> <bean id="maskProcessor" class="com.example.otter.processor.MaskEventProcessor"/>然后把它塞进框架认的那个列表里。如果原来的定义长这样：
<property name="eventProcessors"> <list> <!-- 原有内容 --> </list> </property>就在<list>里追加<ref bean="sourceTagProcessor"/>。顺序在这里是有意义的，列表里靠前的先执行，后面的能看到前面改过的结果。脱敏和打标的顺序通常无所谓，但如果两个处理器都改同一个字段，顺序就决定了谁赢。
改配置之前先备份。这不是客套话，我见过把整个 bean 定义改坏、Node 起不来、又找不到原始备份、最后只能重新解压发行包重配一遍的事故。一条命令的事：
cp conf/applicationContext.xml conf/applicationContext.xml.bak.$(date +%Y%m%d)3.2 依赖打包：lib 目录方案的两个雷
类写好了要能被 Node 加载，最直接的方式是打成 jar 丢进lib/目录。Node 的启动脚本会把lib/*.jar全部加进 classpath，放进去就生效。
打包时有两个雷区。第一个是依赖范围。你的处理器如果引了 commons-lang3、fastjson 这类库，而 Node 自带的 lib 里已经有另一个版本，类加载器会按 classpath 顺序选一个，选到哪个看运气。轻则行为不一致，重则NoSuchMethodError启动失败。解决办法是把这些依赖在 pom 里设成provided，只编译不打包：
<dependency> <groupId>org.apache.commons</groupId> <artifactId>commons-lang3</artifactId> <version>3.12.0</version> <scope>provided</scope> </dependency>前提是你用的 API 在 Node 自带版本里也存在。不确定的话，先ls lib | grep commons-lang3看看版本，再决定用哪些方法。稳妥起见，处理器里尽量只用 JDK 自带的类，日志用 slf4j 就行，别引入额外的工具库。
第二个是打包插件配置。默认的 maven-jar-plugin 打出来的 jar 不带依赖、不带 Main-Class，这正好是我们要的。但如果你之前配过 shade 或 assembly 打成 fat jar，赶紧改回来，fat jar 里的类会覆盖 Node 自己的，后果不可控。
打完之后确认一下内容：
mvn clean package -DskipTests unzip -l target/custom-processor-1.0.jar看到里面只有你自己的几个 class 和配置文件就对了。
3.3 部署与灰度：多节点必须同步改
Otter 的 Node 通常不止一台，同一个 pipeline 可能跑在多个节点上。这时候只改一台是最危险的，你会得到"一部分数据被处理了、一部分没有"的诡异现象，而且随着负载均衡漂移，表现还会变化。
部署流程我一般这么走：
- 先在一台非核心节点上部署，jar 进
lib/，配置改好，重启 - 观察启动日志有没有报错，重点看
ClassNotFoundException、NoSuchMethodError、bean 初始化失败 - 造一条测试数据走一遍，确认目标表里字段符合预期
- 观察半小时，看同步延迟、错误数、Node 的 GC 情况
- 逐台滚动，每台都重复第 2 步的检查
重启命令一般在bin/下：
cd /data/otter/node sh bin/restart.sh # 跟一下启动日志 tail -f logs/node/node.log | grep -i -E "error|exception|processor"重启时注意 pipeline 的位点状态。正常情况下 Otter 会记住位点，重启后从上次位置继续。但如果位点已经推进、数据还在内存里没落库，重启可能丢一小段。所以别在同步高峰期重启，找一个业务低峰期，或者几分钟的延迟大家能接受的窗口。
回滚方案也提前准备好：把旧的lib/里那个 jar 和改之前的配置文件都留一份，出问题三分钟能退回去。cp命令成本很低，比在线 debug 便宜多了。
3.4 验证生效：别只看日志，要看数据
验证分三层，缺一层都不算数。
第一层看日志。处理器里打的debug或者info有没有出现。如果日志里一条都没有，说明类根本没被加载或者没被调用，回去查配置和 jar 路径。
第二层看目标表数据。这条最实在。造一条测试数据——新增、更新、删除各一条——然后去目标库select出来看。字段值对不对、时间戳有没有、脱敏有没有生效，一目了然。测试数据要覆盖到边界：手机号字段为空的行、超长的行、特殊字符的行。
第三层看指标。Otter 的管理端能看到同步延迟和 TPS。处理器加上之后，如果延迟曲线明显抬升，或者 TPS 掉了，说明你的代码成了瓶颈，得优化。这个观察至少持续一个业务周期，白天高峰和夜间批量都要覆盖到。
4. 顺序、幂等与异常：自定义处理最容易翻车的地方
4.1 处理器是单例，成员变量会出事
这一点必须先说，因为它最隐蔽。你在 Spring 里注册的处理器 bean 默认是单例，同一个实例会被多个 pipeline、多个线程并发调用。任何写在成员变量里的临时状态都会串：
// 错误示范 public class BadProcessor implements EventProcessor { private List<String> tempKeys = new ArrayList<>(); // 多线程共享，必炸 public boolean process(Event event) { tempKeys.clear(); // ... 填充 // ... 使用 } }两个线程同时进来，一个clear把另一个的数据清空了，然后数据处理错，还查不出原因——因为它不是每次都错，是偶发。
正确做法是处理器必须无状态：所有中间变量都在方法内声明成局部变量。需要跨调用保存的东西（比如缓存）用线程安全的容器，或者干脆别存。配置类的成员变量（像前面那个sourceTag）只读不改，是安全的。
4.2 异常处理：吞掉和抛出都是错
处理器里抛出一个RuntimeException，会发生什么？整个批次失败，Otter 记录错误并重试。如果这个异常是确定性的（比如某个字段格式永远解析不了），就会形成死循环：重试、失败、再重试，位点永远卡在那里，后面所有数据都堵住了。同步延迟一路飙升，最后告警。
反过来，无脑try-catch全部吞掉，日志也不打，那就是数据静默丢失。目标库缺了一条数据，谁也不知道，直到业务方对账发现。
我的做法是分级处理：能被修正的问题就地修正并记warn；不能修正但要跳过的记error并把原始数据落到本地文件或专门的失败表，然后继续；只有在系统性问题（比如配置错误、依赖缺失）时才抛出去，让链路停下来等人处理。
for (RowData rowData : event.getRowDatas()) { try { transform(rowData); } catch (Exception e) { // 单行失败不影响整批，但要留证据 logger.error("transform failed, table={}, pk={}, before={}, after={}", event.getTableName(), extractPk(rowData), rowData.getBeforeColumns(), rowData.getAfterColumns(), e); // 可选：写入本地失败文件，便于后续补偿 } } return true;日志里一定要打主键和原始数据。只打异常堆栈的话，你事后完全不知道是哪条数据出的问题，只能全表扫一遍。这个习惯我强烈建议养成，代价是几行代码，收益是排查时间从小时级降到分钟级。
4.3 幂等：同一条数据可能被处理不止一次
Otter 在异常、重启、位点回退的场景下都可能重放 binlog 事件。这意味着你的处理器会对同一条变更执行多次。如果处理器本身是纯数据加工（改改字段值），重复执行无所谓，结果一样。但如果你在里面做了有副作用的事——发 MQ、写日志表、调外部接口——就必须考虑幂等。
幂等键一般用「事件时间戳 + 表名 + 主键值」组合，或者直接用 Otter 事件对象上的唯一标识。发消息前先查一下有没有发过，或者让下游按这个键去重。这个成本不低，所以我的原则是：Event Processor 里尽量不要做副作用操作。真要发消息，考虑在目标库落一张 outbox 表，由下游自己消费，把同步链路和消息链路解耦。
4.4 性能：处理器跑在 ETL 线程里
Event Processor 是同步执行在 ETL 的处理线程上的。你的代码耗时直接等于同步链路的耗时。单条处理花 10ms，每秒 100 条变更，那就是 100% 的 CPU 时间被你的处理器吃掉了，TPS 直接腰斩。
所以有两条硬规矩：不在处理器里做远程调用，包括 RPC、HTTP、查数据库、读 Redis；不在处理器里做大量对象创建。前者是延迟不可控，后者是 GC 压力。
如果确实需要外部数据（比如查字典表做枚举转换），做法是启动时加载到本地内存，做成只读的ConcurrentHashMap，并且提供一个刷新机制（比如定时任务或者管理接口触发）。加载失败的场景要有兜底默认值，不能让整个同步卡住。
一个粗略的性能预算：单条RowData的处理控制在微秒级，一个批次（假设 100 行）的总处理时间控制在 10ms 以内。超过这个量级，就该想想是不是哪里写重了。用System.nanoTime()打点测一下，比猜准得多。
5. 常见问题与排查实录
下面这张表是我这些年踩过的、也帮别人排查过的典型问题，按现象归类，遇到直接对号入座：
| 现象 | 可能原因 | 排查方式 | 解决 |
|---|---|---|---|
| 处理器完全没生效 | jar 没放 lib、配置未注册、Node 没重启 | 启动日志搜自己的类名，加一条启动时打印的 info | 补齐部署步骤，重启验证 |
| Node 启动失败，ClassNotFound | 依赖缺失或 scope 配错 | 看启动日志首个异常堆栈 | 补依赖或改 provided |
| 同步延迟飙升，TPS 骤降 | 处理器内有远程调用或大循环 | jstack抓线程栈看卡在哪 | 去掉远程调用，改本地缓存 |
| 删除的数据在目标库还在 | 用了 afterColumns 取删除前镜像 | 打日志看 rowData 的 before/after | DELETE 用 beforeColumns |
| 目标库报 Unknown column | 处理器新增了字段但表里没有 | 比对生成 SQL 与表结构 | 改表结构或去掉新增列 |
| 目标库报类型转换异常 | setValue 传的类型和列类型不匹配 | 看报错 SQL 和参数 | 保持原类型，显式转换 |
| 数据偶发错乱、串行 | 处理器有成员变量状态 | 代码 review，看有无非 final 字段 | 改无状态，用局部变量 |
| 部分数据被处理部分没有 | 多节点只部署了一台 | 逐台比对 jar 的 md5 和配置 | 滚动全量部署 |
| 反复重试、位点卡住 | 处理器抛了确定性异常 | 看错误日志里的堆栈和数据 | 兜底 catch，落失败记录 |
| 同一条数据被处理多次 | 位点回退重放，处理器有副作用 | 查是否有重复的 MQ 或日志 | 加幂等键，或去掉副作用 |
表格之外，再补三个排查技巧。
第一个：加一条启动日志。处理器类里加个@PostConstruct或者afterPropertiesSet，打一行logger.info("xxx processor initialized, tag={}", sourceTag)。Node 启动时如果没有这行，说明 bean 根本没被创建，问题在配置或 jar，不在业务逻辑。这一条能砍掉一半的排查时间。
第二个：用管理端对比延迟曲线。处理器上线前后各截一张延迟图，横向对比。如果形态变了（比如从平稳变成锯齿状），说明处理器引入了周期性阻塞，多半是里面做了什么重活。
第三个：临时降级开关。给处理器加一个enabled配置项，出问题时改成 false 重启，先恢复同步，再慢慢查。有这个开关，事故处理的心态完全不一样。开关的判断放在process方法第一行，if (!enabled) return true;，成本可以忽略。
最后分享一个我在实际落地时的节奏安排。第一个处理器别一上来就接核心链路，先挑一张数据量小、业务不敏感的表试水，把它跑通、跑稳、观察一周。这一周里你会遇到各种没预料到的情况——字段类型不一致、空值处理、批量大小超预期——在小表上暴露出来，代价最小。等这套流程跑顺了，再往核心表上推。写扩展点这件事，代码本身真不难，难的是让它在一堆你控制不了的数据上稳稳地跑下去，而这个只能靠一次次小步验证积累出来的手感。
