Apache Storm 1.1.0自定义序列化器serialize/deserialize方法未调用问题

我来帮你分析下这个问题，这种情况我之前也遇到过几次，通常是以下几个原因导致的，咱们一步步排查：

1. 确认数据是否需要跨Worker传输 #

Storm只有在跨Worker节点传输Tuple的时候才会触发序列化逻辑。如果你的Spout和接收ObjectNode的Bolt被分配到了同一个Worker里，数据会直接以对象引用的形式在JVM内传递，根本不会调用序列化/反序列化方法。

你可以通过调整拓扑的并行度配置来验证：

// 强制让Spout和Bolt运行在不同Worker
config.setNumWorkers(2);
builder.setSpout("my-spout", new MySpout(), 1).setNumTasks(1);
builder.setBolt("my-bolt", new MyBolt(), 1).setNumTasks(1).shuffleGrouping("my-spout");

提交后再观察序列化方法是否被调用。

2. 检查序列化器的注册时机与配置传递 #

一定要确保在创建TopologyBuilder之前就完成序列化器的注册，并且把配置对象正确传给StormSubmitter.submitTopology方法。如果注册操作晚于拓扑构建，或者提交时用了默认Config，你的自定义序列化器根本不会被Storm加载。

正确的顺序应该是这样：

// 1. 初始化配置并注册序列化器
Config config = new Config();
config.put(Config.TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION, false);
config.registerSerialization(ObjectNode.class, ObjectNodeSerializer.class);

// 2. 构建拓扑
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("spout", new MySpout());
builder.setBolt("bolt", new MyBolt()).shuffleGrouping("spout");

// 3. 用配置好的Config提交拓扑
StormSubmitter.submitTopology("custom-serialization-topo", config, builder.createTopology());

3. 验证ObjectNode的实际类型 #

有时候你以为传递的是ObjectNode，但实际可能是它的子类（比如某些JSON库的自定义实现）。Storm只会对完全匹配注册类型的对象使用对应的序列化器，子类不会自动继承父类的序列化配置。

你可以在Spout发射Tuple前打印对象的实际类型：

ObjectNode node = ...;
System.out.println("实际类型：" + node.getClass().getName());

如果发现是子类，要么注册该子类的序列化器，要么把注册类型改成更通用的父类（比如JsonNode.class，如果你的序列化器能处理所有JsonNode子类的话）。

4. 检查序列化器的实现是否符合Storm规范 #

Storm的自定义序列化器必须实现org.apache.storm.serialization.ISerializer接口，并且要有无参构造函数（Storm会通过反射实例化它）。虽然你说Bolt初始化时能实例化，但还是要确认接口方法是否正确实现：

import org.apache.storm.serialization.ISerializer;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.Map;

public class ObjectNodeSerializer implements ISerializer {

    // 必须要有无参构造
    public ObjectNodeSerializer() {}

    @Override
    public void prepare(Map conf) {
        // 这里可以做初始化操作，比如加载JSON序列化工具
    }

    @Override
    public void serialize(Object obj, OutputStream stream) throws IOException {
        // 你的序列化逻辑
        ObjectNode node = (ObjectNode) obj;
        // ...
    }

    @Override
    public Object deserialize(InputStream stream) throws IOException {
        // 你的反序列化逻辑
        // ...
        return node;
    }
}

5. 开启调试日志排查细节 #

如果以上步骤都没问题，可以开启Storm的调试日志，查看序列化过程的具体细节：

config.put(Config.TOPOLOGY_DEBUG, true);

或者调整log4j配置，把org.apache.storm.serialization包的日志级别设为DEBUG，这样能看到Storm选择了哪个序列化器，有没有隐藏的报错信息。

内容的提问来源于stack exchange，提问作者robynico