企业官网建设流程全解析

1. Event Processor 在 otter 同步链路里到底站在哪个位置

做数据同步做到一定阶段，早晚会撞上"官方给的映射规则不够用"这堵墙。otter 这套基于增量日志的同步中间件，把库表映射、字段映射、过滤条件做成了可视化配置，日常的整表同步、字段改名、常量赋值都很顺手。可一旦碰上字段脱敏、脏数据拦截、跨库表名重写、按业务规则动态补值这类需求，纯配置就玩不转了。这时候就轮到Event Processor出场——它是 otter 留给使用方的一个自定义事件处理器扩展点，说白了就是让你在数据流转的某个环节插一段自己的 Java 代码，对正在同步的这一批行数据做任意加工。

先把三个问题说清楚。它是什么：Event Processor 是 otter Node 端的一个可插拔组件，实现固定接口、打成 jar 放进节点目录、在同步管道里注册类名，就能接管每一个同步事件的加工权。它能做什么：拿到当前事件的表名、库名、事件类型（增/改/删）以及变更前后的完整字段值，你可以读、可以改、可以丢弃，甚至可以把一张表的数据改写成另一张表的。它适合谁：适合已经在跑 otter 同步、后台配置已经满足不了业务规则、手上又有 Java 开发能力的团队。如果你只是想让 A 表字段 x 同步到 B 表字段 y，老老实实用映射配置就好，别为了炫技上自定义代码——多一行代码就多一份维护成本。

我见过太多人卡在同一件事上：代码写完了，jar 也扔进去了，重启节点，日志里干干净净，处理器就是不生效。问题基本都不在代码本身，而在对这条链路的位置理解错了。这一节我们先把链路捋清楚，后面写代码才不会瞎猜。

1.1 一条增量日志从解析到入库要过几道手

otter 的 Node 节点内部是一条流水线式的处理结构，大体可以拆成四段：抽取（Extract）→ 转换（Transform）→ 装载（Load）→ 落库（Sink）。上游有一个负责解析增量日志的客户端，把源库的变更解析成一串事件对象，放进节点内部的缓冲；抽取阶段从缓冲里把事件取出来，做初步整理；转换阶段按照管理端配置的映射规则，把源表字段一一对应到目标表字段；装载阶段拼出 INSERT / UPDATE / DELETE 语句；最后一步交给目标库执行。

Event Processor 挂的位置，绝大多数版本是在抽取完成、进入转换之前。这个位置非常关键，因为它决定了你手里的数据长什么样：此时字段名还是源库的原始字段名，映射规则还没有生效，目标表的表名和字段名都还没确定。换句话说，你在处理器里改的字段名，下游的映射配置得能认得出来；你在处理器里改的表名，下游映射也得跟着调整，否则转换阶段找不到对应关系，轻则字段丢失，重则整条管道报错停摆。

这里给一个我自己用的确认方法，比翻文档快得多：在process方法第一行打一条 info 日志，把tableName、eventType、变更前后字段名列表全部打出来，然后造一条测试数据，看日志里出现的是源表名还是目标表名、字段是原名还是映射后的名字。一次就能确认当前版本的实际挂载点。别嫌土，这个方法我在四个不同大版本上验证过，屡试不爽。

注意：不同小版本之间，处理器的挂载位置和类所在的包路径都可能微调，任何结论都要以你自己手上那套环境的实测为准，不要拿别人的博客当标准答案。

1.2 内置能力止步的地方，就是自定义处理器的起点

先明确内置配置能干哪些活，才不会把简单问题复杂化。常规的列映射、主键映射、常量赋值、简单过滤（比如按列值排除某些行）、字段拼接，这些在管理端的转换配置里都能配出来，改起来不用发版，出了问题改配置重启管道就行，运维成本极低。

真正需要动代码的，通常是下面这几类：

• 需要查外部状态才能决定的逻辑。比如同步时要根据用户 ID 去查一下该用户所属的租户等级，按等级决定这条数据要不要同步、某些字段要不要脱敏。这种依赖运行时外部信息的判断，配置表达不了。
• 跨字段的复杂加工。比如源表把地址拆成省、市、区三列，目标表要拼成一个 JSON 字符串；或者反过来从 JSON 里解析出多列。这类字符串解析、正则替换、加解密，配置里没有通用算子。
• 动态路由。同一张源表，按某个字段的值把数据分发到不同的目标表，或者按月份分表。表名要动态拼，配置只能写死。
• 元数据统一治理。比如所有表统一补一个数据来源标识、统一把软删除标记做一次转换、统一做敏感字段掩码。这种横切逻辑用代码实现一次，比在每张表的映射里重复配一百遍要靠谱。
• 脏数据兜底。源库历史遗留的脏数据（空字符串、非法日期、超长字段）会让目标库写入失败，导致整条管道卡住，需要一个地方做清洗和丢弃。

反过来，有几件事不要塞进处理器：大批量的远程调用（比如每条变更都去调一次 HTTP 接口），这会直接把同步吞吐拉到地板上；复杂的状态聚合（比如攒够一万条再统一计算），内存和一致性问题很难兜住；以及任何需要强事务保证的写操作。

1.3 三种扩展方式的选择依据

otter 的扩展点不止一个，很多团队一开始就选错了方向，做到一半发现路走不通。我把常见的三种方式做个对照，方便你按场景对号入座。

选择的核心判断只有一条：你的加工逻辑依赖的信息，在"一行数据 + 固定规则"里能不能闭环。能闭环就走配置，不能闭环才写处理器。我曾经接手过一个项目，前面的人为了给某个字段加个固定前缀，硬写了三百行处理器代码，还引入了两个 jar 依赖，后来管道一升级全部失效——这种活配置里一行常量赋值就解决了。

还有一个容易被忽略的维度：变更频率。配置改动重启管道即可生效，代码改动要走打包、发版、重启节点的完整流程，两者发布周期差了一个数量级。所以那些"业务规则三天两头变"的逻辑，尽量设计成"代码里读配置"，而不是"把规则硬编码进代码"。

2. 摸清 Event Processor 的契约：接口、数据模型与注册方式

写扩展代码最怕的就是"照抄样例能跑，改一下就崩"。根子在于不知道接口的语义边界在哪里。这一节把方法签名、数据模型、注册方式三件事拆开讲透，后面写实现就是顺水推舟的事。

2.1 方法签名背后的一句话语义

核心接口极其精简，只有一个方法，返回布尔值：

public interface EventProcessor { /** * 处理一个同步事件 * @return true 表示该事件继续往下游流转；false 表示丢弃该事件 */ boolean process(EventData eventData); }

框架通常会提供一个抽象基类AbstractEventProcessor，把接口里的其他默认行为都实现了，你继承它、只覆写process即可。这个基类的意义在于向前兼容：将来接口新增方法，基类会补默认实现，你的子类不用动。

返回值只有两种语义，但很多人理解偏了：

• 返回true，事件带着你修改后的数据继续流向转换阶段。这是绝大多数情况的返回值。
• 返回false，这条事件被直接丢弃，不会写入目标库。

坑就在false上。它不是"忽略本次修改"，而是"这条数据不要了"。有人想表达"这个字段我不改"，顺手return false，结果线上丢了一批数据，追了两天才定位到。只要你不是明确要丢弃这条数据，一律返回true。

还有一个隐含契约：process抛出异常，后果由框架决定，通常会导致这条管道进入异常状态，甚至触发重试。所以任何可能出错的逻辑，都要在方法内部自己 try-catch 兜住，保证方法本身永远正常返回。我在方法体最外层加了一层try { ... } catch (Throwable t) { log.error(...); } return true;的壳子，异常情况下让数据按原样流下去，比让整条管道停摆要好得多。

2.2 EventData / RowData / Column 三层数据模型

事件对象是一个三层嵌套结构，理解这个嵌套关系是写好处理器的前提。

第一层EventData，描述的是"一次数据库操作事件"。关键属性包括：schemaName（库名）、tableName（表名）、eventType（事件类型，取值通常是 INSERT / UPDATE / DELETE，部分版本还有 DDL 相关的类型）、rowDatas（本次事件包含的行数据集合）、executeTime（源端执行时间戳）。注意rowDatas是个 List，一次事件可能带多行，比如一条 SQL 批量插入了十条记录，就是一个事件十行数据。写代码时如果默认只有一行，那就是给自己埋雷。

第二层RowData，描述的是"一行的变更"。核心是两个列表：beforeColumns（变更前的字段值）和afterColumns（变更后的字段值）。这两种事件类型的取值规律必须记牢：

• INSERT：beforeColumns为空或全为 null，只有afterColumns有值。
• DELETE：afterColumns为空或全为 null，只有beforeColumns有值。
• UPDATE：两个都有值，但只有真正发生变化的字段才会出现在afterColumns里（具体取决于日志格式和配置，有些配置下是全字段），这一点极易踩坑。

第三层Column，描述的是"一个字段"。属性有columnName、columnValue（通常是字符串形式）、columnType（数据库类型）、isNull、isKey（是否主键）等。值的类型几乎都是字符串，写代码时要自己转数字、转日期，转换失败的兜底必须写。

这里有个非常实用的技巧：UPDATE 事件里要拿"变更前的完整行"，不能只依赖beforeColumns。因为某些日志格式下beforeColumns只包含主键和变更字段。如果业务逻辑需要整行旧值，稳妥做法是在处理器里维护一个以主键为键的本地缓存，或者干脆改成只依赖afterColumns的新值做判断。我一般倾向于后者，因为缓存会引入内存和一致性问题，能不引入就不引入。

2.3 类加载路径与配置注册的两个入口

代码写完了，还有个"框架怎么找到你的类"的问题。otter 的 Node 端采用了自己的类加载方式，你打的 jar 必须放在它扫描的目录下，配置里填全限定类名才能实例化。

第一个入口是节点本地目录。Node 安装目录下一般有一个专门放扩展 jar 的目录（有些版本叫extend，有些是lib），把编译好的 jar 丢进去，重启节点，类就能被加载到。这里最容易出问题的是依赖冲突：你的处理器依赖了某个第三方库，恰好和 Node 自带的版本不一致，结果启动时NoSuchMethodError。规避办法是把依赖尽量精简，别一股脑把整个项目的依赖树打进去，能不用第三方库就不用。

第二个入口是管道配置。在管理端的通道配置里，找到对应的同步管道，参数里有一个处理器相关的配置项（不同版本字段名可能是processor或extendProcessor之类），填入com.yourcompany.sync.processor.XxxProcessor这样的全限定类名。有些部署方式下也可以直接写在节点的配置文件里，但走管理端配置的好处是改完自动下发，不用挨个登录机器。

注意：填的是全限定类名，不是简单类名。少写包路径，框架一样能启动，但运行时就是找不到类，而且往往只打一行不起眼的警告日志，很容易被忽略。

两个入口都配好之后，还有一个隐含要求：你的处理器类必须有一个无参构造函数。框架是通过反射直接 new 出来的，不经过任何依赖注入容器。这一点决定了你不能在处理器里直接@Autowired一个 Service——它压根不在 Spring 容器里。后面第 4 节会讲怎么优雅地解决外部依赖问题。

3. 一个能直接落地的自定义处理器实现

光讲理论没用，这一节给一个我实际项目里用过的完整实现，场景是"订单表同步时的敏感字段脱敏 + 脏数据过滤 + 动态表名重写"。代码可以直接改成你的业务逻辑。

3.1 先把处理场景钉死：脱敏、过滤、改写

场景设定是这样的：源库有一张订单表，同步到目标库的分析型库。同步过程中需要满足三条业务规则。

规则一，敏感字段脱敏。手机号字段receiver_mobile要在目标库保留前三位和后四位，中间四位替换成星号。身份证字段receiver_id_card只保留前六位和后四位。这两个字段在源库是明文，分析团队只能看脱敏后的值。

规则二，脏数据丢弃。源库历史上有一批tenant_id为0的测试数据，还有一些order_no为空串的异常记录，这些数据同步到目标库会污染统计口径，需要拦截掉。

规则三，按月分表。目标库把订单数据按月拆成order_detail_202401这样的表，表名要按订单创建时间动态拼接。但注意，这个改写动作必须和下游的映射配置配合，否则转换阶段会报错找不到目标表。

这三条规则分别对应了处理器的三种能力：改值、丢弃、改元信息，覆盖面比较全，也基本涵盖了日常能遇到的大部分需求。

3.2 完整代码与逐段说明

先看代码。类路径以你依赖里的实际结构为准，我把关键位置都加了注释。

package com.example.sync.processor; import xxx.otter.node.extend.processor.AbstractEventProcessor; import xxx.otter.node.extend.processor.EventData; import xxx.otter.node.extend.processor.RowData; import xxx.otter.node.extend.processor.Column; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.text.SimpleDateFormat; import java.util.Date; import java.util.List; /** * 订单同步处理器：脱敏 + 脏数据过滤 + 按月分表 * 无参构造由框架反射调用，不要在这里做耗时初始化 */ public class OrderSyncProcessor extends AbstractEventProcessor { private static final Logger logger = LoggerFactory.getLogger(OrderSyncProcessor.class); /** 目标表名前缀 */ private static final String TARGET_TABLE_PREFIX = "order_detail_"; /** 需要脱敏的手机号字段 */ private static final String COL_MOBILE = "receiver_mobile"; /** 需要脱敏的身份证字段 */ private static final String COL_ID_CARD = "receiver_id_card"; @Override public boolean process(EventData eventData) { // 最外层兜底，任何异常都不能让管道停摆 try { doProcess(eventData); } catch (Throwable t) { logger.error("process error, schema={}, table={}, eventType={}", eventData.getSchemaName(), eventData.getTableName(), eventData.getEventType(), t); } // 除非明确要丢数据，永远返回 true return true; } private boolean doProcess(EventData eventData) { String tableName = eventData.getTableName(); // 只处理自己关心的表，其他表原样放行 if (tableName == null || !tableName.equalsIgnoreCase("t_order")) { return true; } List<RowData> rowDatas = eventData.getRowDatas(); if (rowDatas == null || rowDatas.isEmpty()) { return true; } for (RowData rowData : rowDatas) { List<Column> afterColumns = rowData.getAfterColumns(); // INSERT 和 UPDATE 只需要处理变更后的值；DELETE 直接放行 if (afterColumns == null || afterColumns.isEmpty()) { continue; } // 规则二：脏数据判定，命中即整体丢弃 if (isDirty(rowData)) { logger.warn("dirty data skipped, table={}", tableName); return false; } // 规则一：遍历字段做脱敏 for (Column column : afterColumns) { String name = column.getColumnName(); if (COL_MOBILE.equalsIgnoreCase(name)) { column.setValue(maskMiddle(column.getValue(), 3, 4)); } else if (COL_ID_CARD.equalsIgnoreCase(name)) { column.setValue(maskMiddle(column.getValue(), 6, 4)); } } } // 规则三：按订单创建时间重写目标表名 rewriteTableName(eventData); return true; } /** 脏数据判定：租户为 0 或订单号为空 */ private boolean isDirty(RowData rowData) { String tenantId = null; String orderNo = null; for (Column column : rowData.getAfterColumns()) { if ("tenant_id".equalsIgnoreCase(column.getColumnName())) { tenantId = column.getValue(); } else if ("order_no".equalsIgnoreCase(column.getColumnName())) { orderNo = column.getValue(); } } if ("0".equals(tenantId)) { return true; } return orderNo == null || orderNo.trim().isEmpty(); } /** 按 create_time 重写表名，格式 order_detail_yyyyMM */ private void rewriteTableName(EventData eventData) { String createTime = null; List<RowData> rows = eventData.getRowDatas(); if (rows == null || rows.isEmpty()) { return; } for (Column column : rows.get(0).getAfterColumns()) { if ("create_time".equalsIgnoreCase(column.getColumnName())) { createTime = column.getValue(); break; } } if (createTime == null) { return; } String suffix = parseYearMonth(createTime); if (suffix != null) { eventData.setTableName(TARGET_TABLE_PREFIX + suffix); } } /** 从各种可能的日期格式里抠出 yyyyMM，失败返回 null 走原表 */ private String parseYearMonth(String raw) { String[] patterns = {"yyyy-MM-dd HH:mm:ss", "yyyy-MM-dd", "yyyyMMddHHmmss"}; for (String pattern : patterns) { try { SimpleDateFormat sdf = new SimpleDateFormat(pattern); sdf.setLenient(false); Date date = sdf.parse(raw); return new SimpleDateFormat("yyyyMM").format(date); } catch (Exception ignore) { // 换下一种格式继续试 } } return null; } /** 保留前 front 位和后 back 位，中间用星号填充 */ private String maskMiddle(String value, int front, int back) { if (value == null || value.length() <= front + back) { return value; } StringBuilder sb = new StringBuilder(value.length()); sb.append(value, 0, front); for (int i = 0; i < value.length() - front - back; i++) { sb.append('*'); } sb.append(value, value.length() - back, value.length()); return sb.toString(); } }

这段代码有几个刻意的设计，值得单独说一下。

第一，双层方法结构。process只负责兜底，真正的逻辑在doProcess里。这样日志上下文更清晰，也方便将来拆分。

第二，第一时间做表名过滤。不同步的表直接return true放行，避免无关表的数据进入后续逻辑，白白消耗 CPU。如果这条管道同步了几百张表，这一行的性能收益非常可观。

第三，脏数据判定用return false整体丢弃。注意这里的位置——在脱敏之前判，避免对要丢弃的数据做无用计算。

第四，表名重写放在最后。因为要用到afterColumns里的create_time，先处理字段再改表名，顺序上更自然。但这里有个隐患：如果一条事件包含多行不同月份的数据，按第一行的时间改表名会导致其他行落到错误的表。真实场景里这种情况必须显式处理，要么按行拆分事件（框架不一定支持），要么在源头保证一次事件不跨月。我在项目里的做法是加一条告警日志，检测到跨月就记下来人工排查，同时按第一行执行，至少保证数据不丢。

提示：column.setValue()是否可用取决于版本实现。如果你的版本里Column对象是只读的，就改成替换整个RowData里的 column 列表，或者用反射设置字段。踩坑前先去依赖包里反编译看一眼Column有没有 setter，五分钟的事。

3.3 打包部署与配置生效的验证步骤

代码写完只是第一步，落地还有一串动作。我把实际操作顺序列一遍，照做基本不会漏。

第一步，编译打包。用 Maven 打一个不含第三方依赖的瘦 jar，前提是你的处理器只用了 JDK 自带的类和框架提供的类。如果确实需要第三方库，打包时必须做依赖收敛，具体做法是检查 Node 自带 jar 的版本，把自己依赖的版本向它对齐。

mvn clean package -DskipTests # 产物在 target/ 下，注意不要把整个依赖树打进去

第二步，放置 jar。把 jar 复制到 Node 安装目录下框架约定的扩展目录，权限要与 Node 的运行用户保持一致，否则会出现"文件在但读不了"的诡异现象。

第三步，配置类名。在管理端的通道 → 管道配置里填写全限定类名，保存并下发。

第四步，重启节点。绝大多数版本下，新增的类不会热加载，必须重启 Node 进程。

第五步，验证生效。造一条测试数据，观察两件事：目标库里敏感字段是否变成掩码；Node 日志里那条自定义的 warn 日志是否打出来了。如果两边都没反应，直接跳到第 5 节看排查路径。

# 观察日志，确认处理器被调用 tail -f logs/node/node.log | grep -i "OrderSyncProcessor"

这里有个我踩过的坑：测试数据要用一条真的 UPDATE，不要用 INSERT。因为 INSERT 事件的beforeColumns是空的，如果你在代码里依赖了变更前的值做判断，INSERT 测试通过、线上 UPDATE 却出错，或者反过来。两种事件类型都要覆盖测一遍，DELETE 也别落下。

4. 参数怎么算、性能怎么调：几个关键数值的来路

自定义处理器一旦上线，就站在了同步链路的必经之路上，它的性能直接决定了整条管道的吞吐。这一节讲几个必须算清楚的数值，以及并发和内存上最容易翻车的地方。

4.1 单事件内存占用估算与堆大小倒推

很多人调优是拍脑袋加内存，加了还不解决问题。正确的顺序是先算清楚一个事件对象占多少内存，再倒推堆大小。

估算方法：一个Column对象本身有对象头（约 16 字节）、若干引用字段（每个 4 到 8 字节，取决于是否开启压缩指针），加上字符串值本身。Java 里一个字符串对象的额外开销大约是 40 字节出头（对象头 + 数组头 + 引用），再加上字符数据本身（ASCII 字符通常一字节一个）。

按一张 50 列的宽表、每列平均 32 字节的值来算：

• 单个字段的字符串开销约 40 + 32 = 72 字节，Column包装约 40 字节，合计约 112 字节。
• 一行 50 列约 5.6 KB。
• 一个RowData同时持有 before 和 after 两份，UPDATE 事件约 11 KB。
• 一次事件默认携带的行数取决于上游批量参数，常见的批次在几百到上千行。

如果一批是 500 行，单个事件对象就是 5.5 MB 左右。节点内部还有多级缓冲队列，通常是"抽取队列 + 转换队列 + 装载队列"，每个队列能囤积若干批次。按每级队列囤 20 个批次、三级队列算，光原始数据就要 300 MB 以上，加上解析过程中的临时对象和 GC 需要预留的空间，堆内存低于 4 GB 基本就是在悬崖边上跑。

我的经验数值供参考：小表（列数少于 20）同步、QPS 在千级以内，节点堆给 4 GB 够用；宽表（列数 50 以上）或 QPS 上万，直接给 8 GB 起步，并且要观察 GC 日志里的老年代回收频率，如果 Full GC 间隔小于半小时，就该考虑继续加压或者拆分管道。

注意：别在处理器里持有大对象引用。我见过有人在处理器里缓存了整张维表做本地查找，维表几十万行，一个节点直接吃掉几个 G，最后 OOM 把整条管道带崩。本地缓存不是不能用，但必须设上限、设过期，最好用带容量限制的缓存实现。

4.2 批处理与单条处理的取舍

处理器是逐事件调用的，但事件内部本身带了一批行。这里有两个层面的"批"：框架给的事件批量，和你在业务上想要的聚合批量。

框架批量由上游配置决定，调大能提升吞吐（减少调用次数和网络往返），但会拉高峰值内存。我的建议是保持默认值先跑，观察 GC 和目标库的写入压力，如果目标库写入是瓶颈，调大批量反而会让单次写入事务过大，可能引发锁等待，得不偿失。

业务聚合，也就是"攒够 N 条再统一处理"，这个要非常谨慎。处理器是无状态的逐事件回调，你想聚合就得自己维护一个队列和定时器，这会带来三个问题：一是数据延迟，攒批的过程就是延迟；二是节点重启时队列里的数据直接丢失；三是多线程并发下的同步开销。除非你的场景明确能接受延迟和少量丢失（比如埋点统计、日志汇聚），否则不要做业务聚合，老老实实逐条处理。

有一个折中的优化手段值得一试：把昂贵的操作提到类加载阶段做一次。比如初始化一个 HTTP 客户端、加载一次配置文件、建立一次数据库连接池。这些放在静态初始化块或者懒加载里做一次，比在process里每次都初始化要快几个数量级。但记住，静态初始化失败会导致类加载失败，整个处理器直接不可用，所以初始化代码本身也要有 try-catch 和降级方案。

4.3 并发安全与状态缓存的处理手法

这是最容易出线上事故的地方。处理器实例在节点里通常是单例，被多个工作线程并发调用。这意味着你写在类里的任何非 final 成员变量，都是共享的可变状态。

几条硬性规则：

• 所有成员变量要么static final（常量、不可变集合），要么干脆不定义。需要中间变量就在方法内部声明。
• 如果确实需要缓存（比如字典映射、脱敏规则表），用线程安全的容器，并且必须是只读语义，绝对不能有"读不到就写入"的复合操作。
• SimpleDateFormat是线程不安全的经典代表。我上面代码里每次 new 一个实例，看着浪费，但避免了并发下的日期错乱。更好的做法是用DateTimeFormatter（线程安全）或者用ThreadLocal包一层。
• 计数器要用原子类。想在处理器里统计处理了多少条、丢弃了多少条，用AtomicLong，别用long自增，那个数字在多线程下基本没意义。

关于外部依赖的注入问题，前面提到处理器不在 Spring 容器里，@Autowired不生效。我的做法是一个静态持有者的模式：

public class OrderSyncProcessor extends AbstractEventProcessor { // 静态持有，类加载时初始化一次 private static volatile RuleService ruleService; static { try { ruleService = RuleServiceLoader.load(); } catch (Throwable t) { logger.error("rule service init failed, fallback to empty", t); ruleService = RuleService.empty(); } } }

关键在fallback。初始化失败不能抛出去，必须降级成一个空实现，让数据至少能原样流过去。同步链路的原则永远是数据不丢优先于数据加工正确，加工错了可以事后修，数据丢了就真没了。

5. 现场排查实录：处理器不生效与数据异常

这一节是纯经验，都是我或者同事真实遇到、真实排查过的。按现象分类，方便你对号入座。

5.1 处理器压根没被加载的三种情况

现象：日志里连一行初始化输出都没有，目标库数据跟没配处理器时一模一样。

第一种情况，类名写错了。最常见的是漏了包路径，或者包路径大小写不一致，或者类名拼错一个字母。框架加载失败往往只打一行带ClassNotFoundException的警告，日志级别还不高，很容易被后面的业务日志淹没。排查方法：直接去 Node 日志里搜类名的关键字，看有没有异常栈。

第二种情况，jar 放错目录了。不同版本约定的扩展目录不一样，放到lib下和放到extend下效果可能完全不同。最保险的办法是去看启动脚本里 classpath 是怎么拼的，一行命令就能确认。

# 看启动命令里的 classpath 组成 ps -ef | grep node | head -1

第三种情况，配置没下发。管理端改完配置，理论上会自动推送到节点，但如果节点和管理端之间的连接有问题，配置可能还停留在旧版本。验证方法是在节点本地看配置文件里的管道参数，确认处理器类名是不是已经写进去了。改完配置后重启节点，能规避掉一大半这类问题。

还有一种隐蔽的情况：配置了多个处理器，但框架只认最后一个或者第一个，你以为配上了实际覆盖了。这个要看具体实现，稳妥做法是一个管道只配一个处理器，需要多个逻辑就合并到一个类里，用方法拆分。

5.2 数据错乱、漏同步、重复写入的追查路径

数据错乱。表现是目标库某行数据跟源库对不上，字段值串行或者部分字段丢失。第一嫌疑人是在处理器里改了字段名或者表名，但下游映射配置没同步调整。转换阶段按照字段名匹配，名对不上就取不到值，最后写入的是空值或者默认值，不报错但数据是错的。追查办法是加日志把处理前后的字段名列表都打出来，逐条比对。

第二个嫌疑人是事件里有多行数据但代码只处理了第一行。我上面代码里rewriteTableName就只看了rows.get(0)，这在跨月场景下是有问题的，必须配合告警。写代码时凡是出现get(0)，都要停下来问一句：真的只有一行吗？

漏同步。表现是源库有的数据目标库没有。第一嫌疑人就是return false用错了地方，条件判断写反了，把所有正常数据都当成脏数据丢了。排查方式是给丢弃动作加日志，把丢弃的主键和判定字段值都打出来，抽样核对。第二嫌疑人是异常被吞掉了——我在process外面包了 try-catch 保证不中断，代价是异常时数据按原样往下走，如果这个异常恰好发生在"本该修改字段"的地方，数据就会以错误的形式流下去。所以 catch 里必须打 error 日志，并且带上关键上下文。

重复写入。处理器本身不负责写入，理论上不产生重复。但如果你的处理器里有外部写操作（比如写一张日志表、调一个接口），而管道因为下游失败触发了重试，你的外部写操作就会执行多次。这就是我在 2.1 节强调"不要做写操作"的原因。真有这个需求，必须在外部写操作上做幂等，按事件里的主键 + 时间戳去重。

5.3 常见问题速查表

6. 几条踩坑之后的经验沉淀

聊完技术细节，说几条我认为比代码本身更重要的东西。

可观测性必须自己加，不能指望框架。处理器里的日志一定要带够上下文：库名、表名、事件类型、主键值、判定结果。别只写一句"处理完成"，出问题时这种日志等于没写。我的习惯是每条丢弃的数据都打 warn 并带上主键，每条异常都打 error 并带完整堆栈，正常处理打 debug 只在排查时打开。日志量是可控的，因为丢弃和异常本身就是少数情况。

给处理器留一个开关。在代码里加一个从系统属性或者环境变量读取的开关，出问题时能秒级关掉处理器让数据按原样流过去，而不是急急忙忙去改配置、重启节点。这个开关的实现成本极低，一条if (!enabled) return true;的事，但关键时刻能救命。我在三四个项目里都加了这个开关，有两次真的用上了。

上线前一定要用真实数据量压测。用小批量测试数据跑通的代码，在几千 TPS 的压力下表现可能完全不同。压测时重点看三个指标：节点 CPU、老年代 GC 频率、目标库写入延迟。处理器引入的开销主要体现在 CPU 上，如果压测发现 CPU 打满，先用采样工具抓一次热点，八成是在循环里做了字符串拼接或者正则匹配。

变更管理上把处理器当成核心代码对待。这意味着：有完善的日志、有版本号、有回滚方案、有变更记录。我见过最惨的一次事故，是有人在生产环境的处理器里直接改了个字段名，忘了同步修改下游映射，导致某张表的某个字段空了三天才被发现。如果当时有变更记录和字段级别的核对机制，这个问题活不过十分钟。