Storm实时流处理：架构、拓扑、消息可靠性与选型指南

做实时数据处理这行的人，大概都绕不开 Storm 这个老名字。我第一次接触它的时候，手里还是一个跑批要半小时的日志统计系统，业务方天天催着要秒级看板，那种被追着要数据的焦虑感至今记得。Storm 简介这四个字看着像教科书的第一章，但真把它从下载安装跑到生产稳定，中间要趟的坑远比“简介”两个字厚得多。这篇不写空话，我把 Storm 的定位、架构模型、消息可靠性机制、环境搭建、拓扑编写、集群提交、排错路径和选型判断一次性讲透，适合三类读者：准备上手实时计算的新人、正在维护老 Storm 集群的开发运维、以及做技术选型时想搞清楚“这东西还行不行”的架构同学。看完你至少能做到：自己拉一套单机环境、写一个能跑通的拓扑、把它提交到集群，遇到常见报错知道往哪儿查。

1. Storm 到底是什么，什么时候该想到它

1.1 先把“流”和“批”这两个字分清楚

很多人第一次听到 Storm，是被“实时计算”这四个字带进来的，但真正卡住新人的往往不是 Storm 本身，而是没搞清楚流处理和批处理的边界。批处理是攒一批数据、一次性算完、出个结果，典型心态是“我等得起”；流处理是数据一条一条进来，来一条算一条，结果持续往外吐，典型心态是“我一条都不想等”。Storm 属于后者，它是一个纯流式的分布式计算框架，最小处理单元是单条消息，延迟通常在毫秒到几十毫秒这个量级。

这个定位直接决定了它的设计取舍。因为没有“批次”这个概念，Storm 不需要做微批切分，也就不存在批间隔带来的固有延迟；但代价是它没有内置的批式回fill能力，也不擅长做历史数据的重算。我见过不少团队拿 Storm 去做 T+1 的报表汇总，结果被复杂的窗口逻辑和状态管理折磨得够呛，最后换成别的方案。判断标准其实很简单：如果业务要的是“数据产生后几百毫秒内就要看到结果”，Storm 是对路的；如果要的是“每天凌晨跑一遍全量”，那用 Storm 就是拿螺丝刀敲钉子。

另一个容易混淆的点是“流式”和“消息队列”。消息队列解决的是传输和解耦，Storm 解决的是计算。实际架构里两者经常搭配出现：消息队列负责把数据可靠地缓冲住，Storm 从队列里消费、做清洗、聚合、关联、打分，再把结果写回存储或下游队列。搞清楚这个分工，后面配拓扑的时候思路会顺很多。

1.2 它的能力边界到底在哪儿

Storm 的强项是低延迟和高吞吐的组合，官方给出的能力区间是单节点每秒能处理百万级消息，这个数字在有状态计算、复杂窗口聚合的场景下会打折扣，但在“收一条、处理一条、转手一条”的管道型任务里非常能打。它的另一个强项是水平扩展的直观性：并行度调一调，多加几台机器，吞吐就上来了，不需要改业务代码。这种“加机器就变快”的朴素体验，在早期实时计算选型里是很有说服力的。

但边界也很清楚。第一，它原生的状态管理能力比较薄弱，跨消息的聚合状态需要自己找地方存，通常靠外部存储或者自己维护内存结构，重启恢复要额外设计。第二，它只保证“至少一次”的语义，要做到精确一次得靠下游去重或者自己在上游做幂等。第三，它的编程模型是偏底层的，你得自己定义数据怎么流、怎么分组、怎么容错，不像后来的框架那样把窗口、状态、检查点都封装好。这三点决定了 Storm 更适合“管道型、逻辑相对简单、对延迟敏感”的业务，而不是“复杂事件处理、长窗口、强一致”的业务。

我把这个边界总结成一句话给团队新人：Storm 擅长把数据搬得快、算得简单，不擅长把数据算得深、存得久。你拿它当实时管道和轻量计算层，它很称职；你拿它当实时数仓的计算引擎，它会让你很累。

1.3 哪类场景用着最舒服

从实际项目看，Storm 用着最舒服的场景有这么几类。第一类是实时清洗和标准化，原始日志进来，解析字段、丢弃脏数据、统一格式，然后转手给下游，逻辑线性、延迟敏感，Storm 的拓扑结构天然适配。第二类是实时监控和告警，对指标流做滑动窗口统计，超过阈值就触发通知，这类任务吞吐不算极端但延迟要求高，Storm 的毫秒级响应正好合适。第三类是实时特征加工，把用户行为流实时转成特征向量，喂给在线模型，这类任务的特点是计算逻辑固定、数据量大、要求稳定，Storm 的并行扩展能力能顶住。

反过来说，有几类场景我建议直接绕开 Storm：需要复杂事件序列匹配的，需要长周期状态（比如跨小时的会话聚合）的，需要精确一次语义且下游没法做幂等的，以及团队里没有熟悉分布式调试的人手的。最后这条不是技术问题，是人力问题，但往往最致命——Storm 的报错信息不算友好，集群层面的问题如果没人能定位，运维成本会非常高。

2. 架构与核心概念：先把这几个词吃透

2.1 Topology、Spout、Bolt、Tuple 这四件套

Storm 的编程模型非常简洁，核心就四个概念，理解透了后面写代码基本是填空。Topology 是顶层容器，你可以把它理解成一张有向无环图，图里的节点是计算单元，边是数据流向；它一旦提交就会一直运行，直到你手动杀掉，这点跟批处理任务跑完就退出完全不同。Spout 是图的入口，负责从外部数据源读取数据并发射出去，它不处理数据，只负责“进货”。Bolt 是图的中间和末端节点，负责真正的计算逻辑，可以过滤、聚合、关联、写库，一个拓扑里可以有多个 Bolt 串联或并联。

Tuple 是数据载体，也就是在 Spout 和 Bolt 之间流动的那条“消息”。它本质上是一个带命名字段的列表，字段名在声明流的时候定义，下游按名字取值。这里有个新人常踩的坑：Tuple 是不可变的，你没法在 Bolt 里修改它再传给下游，只能发射一个新的 Tuple。这个设计是为了保证并发安全，因为同一个 Tuple 可能被多个下游任务消费，改来改去会乱套。

还有一个概念叫 Stream，指的是从一个组件流出的一股数据。一个 Spout 或 Bolt 可以声明多股流，下游可以选择订阅哪一股，标签用 streamId 区分。这个能力在实际项目里很有用，比如一个解析 Bolt 可以同时吐出“正常数据流”和“异常数据流”，分别给不同的下游处理，省得在同一个流里做分支判断。

// Spout 声明输出字段 @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("sentence")); } // Bolt 声明输出字段，可以多股流 @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("word", "count")); declarer.declareStream("error-stream", new Fields("raw", "reason")); }

2.2 Nimbus、Supervisor、ZooKeeper 的分工

集群层面有三个角色，分工非常清晰。Nimbus 是主节点，负责接收拓扑提交、分配任务、监控集群状态，它本身不参与实际计算，更像一个调度中心。Supervisor 是工作节点，跑在每台从机上，负责启动和停止本机的 Worker 进程，并定期向 Nimbus 汇报心跳。ZooKeeper 是协调者，Nimbus 和 Supervisor 之间的状态同步、任务分配信息、心跳元数据都放在它上面。

这个架构有个特点值得注意：Nimbus 和 Supervisor 之间没有直接通信，全靠 ZooKeeper 中转。好处是解耦，Nimbus 挂了不影响正在运行的拓扑，Supervisor 挂了重启后能从 ZooKeeper 里读到自己的任务继续干。坏处是 ZooKeeper 成了关键依赖，ZK 一抖，整个集群的调度就会出问题。我在实际运维里遇到过 ZooKeeper 磁盘写满导致拓扑提交卡住的情况，排查了半天才定位到，所以监控 ZK 的健康度是必须做的事。

再往下拆，Worker 是真正跑计算的 JVM 进程，一个 Supervisor 上可以起多个 Worker，每个 Worker 里跑若干个 Executor 线程，每个 Executor 里跑一个或多个 Task 实例。这四层关系（Supervisor → Worker → Executor → Task）是理解并行度的关键，后面会专门展开讲。

注意：Nimbus 在 Storm 里不是单点故障的致命点，因为它不参与数据路径。但如果你用的是高可用模式，需要配多个 Nimbus 并让它们通过 ZooKeeper 选主，配置项是nimbus.seeds，这个参数在集群部署时一定要配对，否则提交拓扑会报找不到主节点。

2.3 八种 Stream Grouping 怎么选

Stream Grouping 决定了一条 Tuple 从上游组件发射后，会被路由到下游的哪个 Task。这个选择直接影响数据分布、计算正确性和性能，是拓扑设计里最需要动脑子的地方。

实际项目里用得最多的是 Shuffle 和 Fields。Shuffle 适合那种“来一条处理一条、彼此独立”的逻辑，比如格式转换、字段脱敏；Fields 适合“同一个 key 要聚到一起”的逻辑，比如按用户 ID 统计、按商品 ID 累加。这里有个经验：用 Fields Grouping 的时候，尽量选基数大且分布均匀的字段，如果拿地区、设备类型这种低基数字段做分组，很容易出现某个 Task 扛了 80% 的数据，其他 Task 闲着看戏，这就是典型的数据倾斜。

All Grouping 要特别小心，它的语义是广播，上游发一条，下游每个 Task 都收一条。如果下游有 100 个 Task，数据量就被放大 100 倍，带宽和 CPU 都吃不消。这个 Grouping 一般只用在“配置更新通知”这种低频小消息的场景。Global Grouping 则是另一个极端，所有数据集中到一个 Task，只适合全局唯一计数的场景，用之前先想清楚这个 Task 会不会成为瓶颈。

3. 环境搭建与第一个可运行的拓扑

3.1 单机环境准备与版本对齐

Storm 的环境搭建不算复杂，但版本对齐是第一个坎。我的建议是 JDK 用 8 或 11，Storm 用 2.x 系列，ZooKeeper 用 3.5 以上，这三个版本的组合比较稳。JDK 版本太高（比如 17）会遇到模块化相关的反射限制，Storm 内部有些依赖会报访问异常，折腾起来很费时间。

单机跑的话，最省事的方式是用官方的发行包，解压后改两个配置文件就能起。conf/storm.yaml里至少要配三项：storm.zookeeper.servers指向本机，nimbus.seeds指向本机，storm.local.dir指定一个本地目录用来存临时文件。这个 local.dir 千万别用临时目录，机器重启后被清空会导致拓扑状态异常。

# 启动 ZooKeeper（如果本机没有现成的） zkServer.sh start # 启动 Nimbus bin/storm nimbus # 启动 Supervisor bin/storm supervisor # 启动 UI，方便看拓扑状态，默认 8080 端口 bin/storm ui

这四个进程起完之后，访问 8080 端口能看到管理界面，就说明环境通了。这里有个小坑：Storm 的 UI 进程和 Nimbus 是分开的，UI 挂了不影响计算，但你会失去可视化的排查入口，所以生产环境要把 UI 也纳入进程守护。另外，storm.local.dir所在的磁盘要留足空间，拓扑运行过程中会写不少临时文件，磁盘满了我见过不止一次。

如果不想折腾原生部署，用容器方式起一个单机环境也可以，但要留意容器内的主机名解析，Storm 组件之间靠主机名通信，容器里如果 hostname 解析不到自己，Nimbus 会一直报连不上。解决办法是在容器里把主机名写进 hosts 文件，这个细节不注意会卡很久。

3.2 工程结构与依赖

写拓扑的工程一般用 Maven 管理，依赖里最关键的是storm-client，它包含了拓扑编程需要的所有 API。注意作用域要设成provided，因为集群上已经有这些类了，你打进 jar 里会造成类冲突。

<dependency> <groupId>org.apache.storm</groupId> <artifactId>storm-client</artifactId> <version>2.6.0</version> <scope>provided</scope> </dependency>

工程结构上我习惯按职责分包：spout放数据源，bolt放计算逻辑，topology放组装入口，util放工具类。这个结构不是强制的，但拓扑一大，Bolt 数量上到十几个，没有清晰的分包会很难维护。还有一个建议是：把每个 Bolt 的处理逻辑写成纯函数式的，输入一个 Tuple 输出若干 Tuple，不依赖共享的可变状态，这样单测好写，并发也不会出问题。

依赖冲突是提交拓扑时最常见的报错来源。表现是本地跑得好好的，提交到集群就抛NoSuchMethodError或ClassNotFoundException。原因通常是你的工程引入了某个库的 A 版本，Storm 集群自带的是 B 版本，运行时加载了集群的版本，方法签名对不上。排查办法是用mvn dependency:tree看依赖树，把冲突的库排除掉或者统一版本。我一般会在打包前跑一遍依赖树，把重复的 group 标出来人工确认，这个习惯能省掉大量线上排查时间。

3.3 WordCount 拓扑完整代码

拿词频统计做例子虽然老套，但它把 Spout、Bolt、分组、拓扑组装四个环节都覆盖到了，是最适合入门的完整样本。下面这个版本我加了注释，可以直接抄。

public class WordCountTopology { // 数据源：不断发射句子 public static class SentenceSpout extends BaseRichSpout { private SpoutOutputCollector collector; private String[] sentences = { "storm is a realtime framework", "storm processes streams fast", "realtime computing needs low latency" }; private int index = 0; @Override public void open(Map<String, Object> conf, TopologyContext context, SpoutOutputCollector collector) { this.collector = collector; } @Override public void nextTuple() { // 带 msgId 发射，用于开启 ack 追踪 String sentence = sentences[index % sentences.length]; collector.emit(new Values(sentence), UUID.randomUUID().toString()); index++; // 控制发射速率，避免本地模式打满 CPU try { Thread.sleep(100); } catch (InterruptedException ignored) {} } @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("sentence")); } } // 切词 public static class SplitBolt extends BaseRichBolt { private OutputCollector collector; @Override public void prepare(Map<String, Object> conf, TopologyContext context, OutputCollector collector) { this.collector = collector; } @Override public void execute(Tuple input) { String sentence = input.getStringByField("sentence"); for (String word : sentence.split("\\s+")) { // 锚定到输入 Tuple，保证 ack 树完整 collector.emit(input, new Values(word)); } collector.ack(input); } @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("word")); } } // 计数 public static class CountBolt extends BaseRichBolt { private OutputCollector collector; private Map<String, Integer> counts = new HashMap<>(); @Override public void prepare(Map<String, Object> conf, TopologyContext context, OutputCollector collector) { this.collector = collector; } @Override public void execute(Tuple input) { String word = input.getStringByField("word"); int count = counts.getOrDefault(word, 0) + 1; counts.put(word, count); collector.emit(new Values(word, count)); collector.ack(input); } @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("word", "count")); } } public static void main(String[] args) throws Exception { TopologyBuilder builder = new TopologyBuilder(); builder.setSpout("sentence-spout", new SentenceSpout(), 1); builder.setBolt("split-bolt", new SplitBolt(), 2) .shuffleGrouping("sentence-spout"); builder.setBolt("count-bolt", new CountBolt(), 4) .fieldsGrouping("split-bolt", new Fields("word")); Config config = new Config(); config.setDebug(false); if (args != null && args.length > 0 && "local".equals(args[0])) { // 本地模式，不提交集群，方便调试 LocalCluster cluster = new LocalCluster(); cluster.submitTopology("word-count", config, builder.createTopology()); Thread.sleep(30000); cluster.shutdown(); } else { StormSubmitter.submitTopologyWithProgressBar( "word-count", config, builder.createTopology()); } } }

这段代码里有几个点值得单独说。第一，Spout 里 emit 带了 msgId，这是开启 ack 追踪的前提，如果只传 Values 不传 msgId，这条消息就失去了可靠性保证。第二，SplitBolt 里用的是collector.emit(input, new Values(word))，把新 Tuple 锚定到输入 Tuple 上，这是构建 ack 树的关键，不锚定的话下游处理失败上游不会重发。第三，CountBolt 用的是 fieldsGrouping，按 word 字段哈希，保证同一个单词落到同一个 Task，计数才准确。如果用 shuffleGrouping，每个 Task 各算各的，结果就是错的。

本地模式那段也很重要。新人调试拓扑不要一上来就提交集群，用 LocalCluster 在 IDE 里跑，日志直接打在控制台，断点能打，改代码不用重新打包，效率差好几倍。等本地跑通了再提交集群，问题会少很多。

3.4 本地模式调试与集群提交

本地模式和集群模式的行为差异主要在三个方面。一是并行度，本地模式下 Nimbus、Supervisor 都在同一个 JVM 里模拟，Worker 进程不会真的起，所以你看不到进程级的问题；二是依赖，本地模式用的是工程里的依赖，集群模式用的是集群上的依赖，这就是为什么本地好使、提交就报错；三是序列化，本地模式有时会走一套宽松的序列化路径，集群模式更严格，字段类型不支持会直接抛异常。

提交集群的命令很简单，把打好的 jar 传上去执行：

bin/storm jar word-count.jar com.example.WordCountTopology word-count-prod

最后那个参数是拓扑名，这个名字在集群里要唯一，重复提交会提示已存在。提交成功后去 UI 上能看到拓扑的运行状态，包括每个组件的吞吐、延迟、失败数。这几个指标是排查问题的起点，后面讲排错会用到。

打包的时候要注意两件事。一是主类要写对，manifest 里没有主类的话提交会报找不到入口。二是依赖要打全，但要把 storm-client 排除掉。我习惯用 maven-shade-plugin 打一个 fat jar，配置里把storm-client和storm-server排掉，日志框架也排掉，避免和集群的日志实现冲突。

4. 消息可靠性、并行度与关键配置

4.1 ack 机制到底怎么工作

Storm 的可靠性保证靠的是 ack 机制，这套机制的核心是“消息树”。Spout 发射一条消息时带上 msgId，这条消息就是一个根节点；每个 Bolt 处理完发射新消息时锚定到上游，新消息就成了子节点；所有叶子节点都处理成功，整棵树才算完成，Spout 会收到 ack 回调；任何一个节点失败或者超时，Spout 收到 fail 回调，可以决定重发。

负责追踪这棵树的是 Acker 组件，它是 Storm 内置的一类特殊 Task。Acker 不存整棵树，只存一个异或值，每条消息进来异或一次 msgId，树完成时异或值归零，就判定成功。这个设计非常省内存，代价是它只能告诉你“成没成”，不能告诉你“哪条失败了”。所以实际排查失败原因，还是得靠日志。

这套机制带来的是“至少一次”语义，不是“精确一次”。因为失败重发会导致重复，下游可能收到两条一样的消息。要解决重复，有两条路：一是下游做幂等，比如写库用主键覆盖；二是上游自己做去重表，记录处理过的 msgId 并定期清理。我在项目里更倾向第一种，简单直接，第二种的去重表本身又是一套需要维护的状态。

// Spout 里实现 ack 和 fail 回调 @Override public void ack(Object msgId) { // 处理成功，可以从待确认队列里移除 } @Override public void fail(Object msgId) { // 处理失败，重新发射 String sentence = pending.get(msgId); collector.emit(new Values(sentence), msgId); }

提示：ack 机制是有成本的，每条消息都要经过 Acker 计算。如果业务本身对重复不敏感，比如纯日志清洗，可以关掉可靠性：Spout 发射时不带 msgId，Bolt 里不调用 ack。关掉之后吞吐能提升不少，这个取舍要根据业务来定，别默认开着了事。

4.2 并行度的四层关系

并行度是 Storm 里最容易配错的地方，因为它是四层叠加的关系。最上层是 Worker 数，也就是 JVM 进程数，由topology.workers控制；第二层是 Executor 数，也就是线程数，由组件声明时的并行度参数控制；第三层是 Task 数，也就是组件实例数，默认等于 Executor 数，但可以单独设得更大；第四层是集群的槽位数，每个 Supervisor 能起多少个 Worker 由supervisor.slots.ports决定。

举个具体例子。假设你设了topology.workers=2，某个 Bolt 的并行度是 4，那这 4 个 Executor 会被分到 2 个 Worker 里，每个 Worker 跑 2 个线程。如果这个 Bolt 的 Task 数设成 8，那每个 Executor 里跑 2 个 Task 实例，线程在执行时轮转处理这两个实例。这个设计的意图是“线程数可以小于实例数”，方便后续扩容时不用改拓扑结构，改 Task 数就行。

实际配置的经验值：Worker 数按集群资源来，一般一个 Worker 给 1 到 2GB 堆内存；Bolt 的并行度按吞吐来，先给个 2 到 4，压测后调整；Task 数初期和并行度保持一致，别一上来就设复杂。我见过有人把并行度设成 100，结果集群只有 8 个槽位，拓扑提交上去一半的 Executor 分不到资源，一直处于 pending 状态，看着像卡住了，其实是在等。

还有一个坑是topology.max.spout.pending这个参数，它限制 Spout 最多有多少条消息处于“已发射但未确认”状态。这个值设大了，内存会涨；设小了，吞吐上不去。默认是 null，表示不限，生产环境一定要设一个具体值，我一般从 1000 开始调，压测看吞吐和延迟的平衡点。

4.3 关键配置参数速查

Storm 的配置项很多，但真正影响运行的也就那么十几个。我把常用的整理成表，方便对照。

这里重点说 Kryo 注册。Storm 默认用 Java 序列化，性能和体积都不理想，Kryo 更快更省。但用 Kryo 有个前提：所有在 Tuple 里传递的自定义对象必须在配置里注册，否则运行时会抛序列化异常。这个异常报错信息通常指向某个类，照着加就行，但如果是嵌套对象，得把嵌套的类也注册上，容易漏。我的做法是写一个统一的注册方法，把所有领域对象集中注册，新加对象时只改这一处。

5. 常见问题与排查实录

5.1 提交阶段的坑

提交阶段的报错大概占新人问题的六成，集中在这几类。第一类是ClassNotFoundException，通常是 fat jar 没打全，或者依赖被排掉了。排查办法是把 jar 解开看BOOT-INF或根目录下有没有对应的类，没有就是打包配置的问题。第二类是NoSuchMethodError，版本冲突的典型表现，用依赖树定位，排除掉冲突版本。第三类是NotSerializableException，Tuple 里传了不支持序列化的对象，要么注册 Kryo，要么换成基本类型。

还有一类比较隐蔽的是权限问题。集群的storm.local.dir如果权限不对，Supervisor 起 Worker 时会失败，日志里报的是文件创建失败，但拓扑状态看起来是“已提交”，实际一直没跑起来。这个坑我踩过一次，查了半小时才发现是目录属主不对。建议部署时统一用同一个用户跑所有 Storm 进程，目录权限设成 755，别用 root 和普通用户混着来。

5.2 运行阶段的数据倾斜与背压

数据倾斜的表现是：拓扑整体吞吐上不去，但看单个组件的指标，某个 Task 的处理量远高于其他 Task。原因通常是 fieldsGrouping 选的字段分布不均，比如按省份分组，某些省份的数据量是其他省份的几十倍。解决办法有几种：一是换分组字段，选基数更大的字段；二是给热点 key 加随机后缀打散，下游再做二次聚合；三是在分组前先做一层预聚合，把数据量压下来。

背压的表现是 Spout 的发射速率被压住，延迟指标持续上涨。Storm 没有显式的背压机制，它的背压是通过队列满和超时体现的。当下游处理不过来，接收队列堆积，上游的 emit 会阻塞或者超时，Spout 的 pending 数量打到上限就停止发射。排查思路是从末端往前看，找到处理最慢的那个组件，通常是某个 Bolt 里有同步 IO 或者复杂计算。解决办法是提高该组件的并行度，或者把同步 IO 改成异步批量写，减少每条消息的处理时间。

我在一个项目里遇到过背压，定位下来是写数据库的 Bolt 每条消息都单独提交一次事务，TPS 上不去。改成攒 100 条批量提交后，吞吐翻了十几倍。这个经验说明：Bolt 里的 IO 操作一定要批量，单条处理在低流量时看不出问题，流量一上来就是瓶颈。

5.3 内存、GC 与序列化问题

内存问题通常表现为 Worker 频繁 Full GC 甚至 OOM。原因有几个常见来源：一是topology.max.spout.pending设得太大，未确认消息堆积占满内存；二是 Bolt 里维护了无界的缓存 Map，只增不减；三是序列化用了 Java 原生方式，对象体积大。排查手段是看 GC 日志和堆转储，确认是哪个对象占了大头。

我的处理顺序是：先看 pending 配置，把它调小验证；再看业务代码里有没有无界集合，改成有界或者加过期清理；最后才考虑换序列化方式。前两个是配置和代码问题，改起来快；第三个涉及全局改动，影响面大，放最后。

还有一个容易忽略的点是日志。Bolt 里如果打了大量日志，尤其是把整条 Tuple 打出来，IO 和内存都会受影响。生产环境的日志级别建议设成 WARN，只在异常时打详细日志，正常路径不打。这个习惯看似小事，高流量下能省不少资源。

6. 选型对比：Storm 还值不值得投入

6.1 与主流流处理框架的横向对比

这些年流处理框架演进很快，Storm 的定位也发生了变化。把它和几个主流方案放一起对比，判断会更清晰。

从这个表能看出来，Storm 的优势是延迟低、模型简单、上手快；劣势是状态管理和语义保证弱、生态增长放缓。如果你的业务是纯管道型、对精确一次没要求、团队想快速上线，Storm 依然是合理选择；如果业务涉及复杂状态、长窗口、强一致，那应该考虑流批一体类的方案。

这里我要说句实在话：技术选型不能只看框架本身，要看团队的人手和维护成本。一个已经稳定跑了几年 Storm 集群的团队，为了追新技术全面迁移，迁移期间的双跑、数据对账、回滚预案，这些隐性成本往往比框架本身的差异大得多。反过来，一个全新项目从零开始，也没有历史包袱，那选一个生态更活跃的方案会更省心。

6.2 老系统怎么平稳过渡

对于已经在用 Storm 的团队，我的建议是“分层处理，不急着一刀切”。先把拓扑按业务重要性分级：核心链路且运行稳定的，继续保留，做好监控和容量规划；新需求且逻辑复杂的，考虑用新框架单独起一条链路，和老的并行跑；边缘的、低频的拓扑，评估后该下线就下线，减少维护面。

迁移的时候有个技巧：把 Storm 拓扑里纯粹的计算逻辑抽成独立的类，不依赖 Storm 的 API，这样换框架时这部分代码可以直接复用，只需要重写数据接入和输出的部分。这个重构不改变现有行为，风险低，但为后续迁移铺了路。我在一个项目里就是这么做的，后来换框架时计算逻辑几乎没动，省了大量回归测试的时间。

还有一点是数据一致性。双跑期间，新老链路的结果要做对账，建议按分钟级别比对关键指标，差异超过阈值就告警。对账逻辑本身要简单可靠，不要用复杂的 SQL，避免对账系统自己出问题。

最后分享一个我在实际运维中的体会：Storm 这类框架的价值，很大程度上取决于你有没有把它“养好”。同样一套集群，有人跑得稳如老狗，有人天天救火，差别不在框架，在监控、在容量规划、在对报错的熟悉程度。我习惯给每个拓扑配三个基础监控：Spout 的 pending 数、各 Bolt 的吞吐和延迟、Worker 的 GC 频率。这三个指标正常，拓扑基本不会出大问题；哪个异常了，顺着查就能定位。把这套监控建起来，比纠结用哪个框架实在得多。