Storm 1.2.3 集群搭建实战：从 Nimbus 到拓扑提交全链路解析

1. 流式计算环境搭建的整体思路与选型考量

流式计算框架在实时数据处理场景里一直是个绕不开的话题。我这次要聊的是一套基于 Storm 1.2.3 的集群搭建过程，底层依赖 JDK 1.8，配合 Hadoop 生态做数据源和存储的衔接。标题里提到的 nimbus、supervisor、topo、spouts、bolts、tuple、task 这些词，基本涵盖了 Storm 的核心概念和运行机制，我打算从安装部署一路讲到拓扑提交和任务调度，把整个链路串起来。

先说清楚这套东西是干什么的。Storm 是一个分布式实时计算系统，它的定位很明确：数据一条一条来，我就一条一条处理，延迟可以压到毫秒级。和 MapReduce 那种批处理不同，Storm 的拓扑一旦提交就会一直跑，除非你手动杀掉。这就决定了它适合的场景是实时监控、实时告警、流式 ETL、在线特征计算这类对时效性要求高的活儿。

那为什么选 Storm 1.2.3 这个版本？说实话，现在 Flink 风头很盛，但在一些存量项目里，Storm 依然是主力。1.2.3 是 1.x 系列里比较稳定的一个版本，社区文档齐全，和 Hadoop 2.x 的兼容性也经过大量验证。JDK 1.8 则是那个时期绝大多数大数据组件的标配，Storm 1.2.3 对 JDK 1.8 的支持非常成熟，再往上到 JDK 11 就会遇到一些模块化带来的反射问题，所以锁在 1.8 是最省心的选择。

整体架构上，Storm 集群是主从结构。Nimbus 是主节点，负责分配任务、监控集群状态；Supervisor 是从节点，负责接收 Nimbus 的指令，启动和停止工作进程。Zookeeper 在中间做协调，Nimbus 和 Supervisor 之间的状态同步、任务分配信息都靠它来维护。一个拓扑提交后，会被拆成多个 spout 和 bolt 实例，这些实例以 task 的形式分布在各个 supervisor 的 worker 进程里运行。tuple 则是数据在拓扑中流动的基本单位，spout 负责产生 tuple，bolt 负责处理 tuple，数据就这样一条条在组件之间传递。

我这次搭建的环境是三台机器，一台做 nimbus 兼 zookeeper，另外两台做 supervisor。操作系统是常见的 Linux 发行版，JDK 用 1.8，Hadoop 已经提前装好，因为后面要演示从 HDFS 读数据写回 HDFS 的流程。下面我会把每一步的操作、参数含义、踩过的坑都摊开来讲，尽量让照着做的人少走弯路。

2. 基础环境准备与 JDK 安装细节

2.1 主机规划与系统层配置

三台机器的角色分配是这样的：第一台同时跑 Nimbus 和 Zookeeper，第二台和第三台跑 Supervisor。每台机器的配置不需要太高，我用的都是 4 核 8G，磁盘 100G，对于学习和中小规模生产足够用了。主机名我分别设为 storm-master、storm-worker01、storm-worker02，后面所有配置都基于这三个名字来写。

系统层要做的事情不多，但每件都不能漏。首先是关闭防火墙，或者至少放行 Storm 用到的端口。Storm 涉及的端口比较多，Nimbus 默认监听 6627，Supervisor 默认监听 6700、6701、6702、6703，UI 默认监听 8080，Zookeeper 用 2181。我图省事直接把防火墙关了，生产环境建议按需放行。

然后是 hosts 文件，三台机器都要把主机名和 IP 的映射写进去，不然后面配置里用主机名会解析失败。再就是时间同步，集群里时间不一致会导致 Zookeeper 会话异常，进而影响 Nimbus 和 Supervisor 的心跳。我用的是 chrony 做时间同步，配置好之后三台机器时间差控制在毫秒级。

还有一个容易被忽略的点是文件句柄数。Storm 运行时会打开大量文件描述符，默认的 1024 往往不够用。我在 /etc/security/limits.conf 里把 nofile 调到了 65536，同时也在系统层面做了相应调整。这个坑我在第一次搭建时踩过，拓扑跑着跑着就报 "Too many open files"，排查了半天才定位到。

2.2 JDK 1.8 安装与验证

JDK 的安装本身不复杂，但有几个细节值得说。我下载的是 Oracle JDK 1.8 的某个较新小版本，解压到 /usr/local/ 目录下，然后通过 /etc/profile 配置环境变量。这里要注意的是，JAVA_HOME 一定要指向 JDK 的根目录，而不是 JRE 目录，因为 Storm 启动脚本里会用到 JAVA_HOME/bin/java 这个路径。

环境变量配置如下：

export JAVA_HOME=/usr/local/jdk1.8.0_xxx export PATH=$JAVA_HOME/bin:$PATH export CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar

配置完之后执行 source /etc/profile，然后用 java -version 验证。这里有个小坑：有些系统自带 OpenJDK，如果 PATH 顺序不对，java -version 显示的还是系统自带的版本。我的做法是在 /etc/profile 里把 JAVA_HOME/bin 放在 PATH 的最前面，确保优先使用我们安装的 JDK。

三台机器都要做同样的操作。做完之后我会用一个小技巧验证：写一个简单的 Java 类打印 System.getProperty("java.version")，确认输出的是 1.8.0_xxx，而不是 1.7 或 1.9。这个步骤看起来多余，但在后面 Storm 启动报版本不兼容的时候，能帮你快速排除 JDK 的问题。

提示：JDK 安装路径不要带空格和特殊字符，Storm 的启动脚本对路径中的空格处理得不太好，容易导致 classpath 解析错误。

3. Zookeeper 集群部署与 Storm 安装配置

3.1 Zookeeper 单机与伪集群的取舍

Storm 强依赖 Zookeeper，没有 Zookeeper 它根本起不来。在学习环境下，我一开始只在一台机器上装了单机版 Zookeeper，后来发现 Supervisor 连不上 Nimbus，排查后发现是 Zookeeper 的 myid 和 server 配置对不上。所以这里我建议，即使是学习，也尽量按集群的方式来配，哪怕只在一台机器上跑伪集群，这样能提前暴露配置问题。

我的做法是在 storm-master 上部署一个 Zookeeper 实例，同时把另外两台机器也加入 Zookeeper 集群。Zookeeper 的配置文件 zoo.cfg 里，dataDir 指向一个独立目录，clientPort 用 2181，然后在文件末尾加上三个 server 的配置：

server.1=storm-master:2888:3888 server.2=storm-worker01:2888:3888 server.3=storm-worker02:2888:3888

每台机器的 dataDir 目录下要创建一个 myid 文件，内容分别是 1、2、3。这个 myid 必须和 zoo.cfg 里的编号一一对应，写错了 Zookeeper 启动时会报错。启动顺序是先启动三台 Zookeeper，然后用 zkServer.sh status 检查，应该看到一台 leader 两台 follower。

3.2 Storm 1.2.3 安装与核心配置项解读

Storm 的安装包解压到 /usr/local/storm 目录下，然后重点改 conf/storm.yaml 这个文件。这个文件是 Storm 的核心配置，里面每一项都影响集群的行为。我逐项说一下我的配置和背后的考虑。

首先是 storm.zookeeper.servers，这里列出 Zookeeper 集群的所有节点：

storm.zookeeper.servers: - "storm-master" - "storm-worker01" - "storm-worker02"

然后是 nimbus.seeds，指定 Nimbus 节点的主机名。在 1.x 版本里这个配置叫 nimbus.host，1.2.3 里改成了 nimbus.seeds，可以配多个实现高可用：

nimbus.seeds: ["storm-master"]

接下来是 supervisor.slots.ports，这个配置决定了每个 Supervisor 上能跑多少个 worker。每个端口对应一个 worker 槽位，我配了四个端口：

supervisor.slots.ports: - 6700 - 6701 - 6702 - 6703

这四个端口意味着每个 Supervisor 最多同时运行四个 worker 进程。端口数量的选择要根据机器的内存来定，每个 worker 默认占 768M 内存，四个 worker 就是 3G 左右，加上系统和其他进程，8G 内存的机器配四个槽位比较合适。配多了会导致内存不足，worker 启动失败。

还有一个重要配置是 storm.local.dir，指定 Storm 存放本地状态数据的目录：

storm.local.dir: "/data/storm"

这个目录要提前创建好，并且确保运行 Storm 的用户有读写权限。Storm 会把 jar 包、临时文件、心跳信息等放在这里。如果这个目录所在的分区空间不足，会导致拓扑提交失败或者 worker 异常退出。

UI 相关的配置我保持默认，端口是 8080。如果 8080 被占用，可以改 ui.port。另外我建议把 storm.log.dir 也显式配一下，方便集中管理日志。

配置改完之后，要把 storm.yaml 分发到三台机器的相同路径下。这里有个细节：storm.yaml 对缩进非常敏感，必须用空格，不能用 Tab。我见过有人用 Tab 缩进导致配置解析失败，Storm 启动时报的错又很模糊，排查起来很费劲。

3.3 环境变量与启动脚本

在 /etc/profile 里加上 Storm 的环境变量：

export STORM_HOME=/usr/local/storm export PATH=$STORM_HOME/bin:$PATH

然后在 storm-master 上启动 Nimbus 和 UI：

storm nimbus & storm ui &

在 storm-worker01 和 storm-worker02 上启动 Supervisor：

storm supervisor &

启动之后，用 jps 命令检查进程。Nimbus 节点上应该看到 nimbus、ui、zookeeper 的进程；Supervisor 节点上应该看到 supervisor 和 zookeeper 的进程。如果进程没起来，去 logs 目录下看日志，常见的问题是端口被占用、Zookeeper 连不上、或者 storm.local.dir 权限不对。

注意：Storm 的启动脚本默认不会把日志输出到控制台，所有日志都在 logs 目录下。排查问题时，nimbus.log 和 supervisor.log 是最先要看的文件。

4. 拓扑结构解析与核心概念落地

4.1 Spout、Bolt、Tuple、Task 的关系梳理

在提交拓扑之前，有必要把这几个核心概念的关系理清楚，不然后面调优和排查问题会一头雾水。

Spout 是拓扑的数据源，它负责从外部系统读取数据并发射 tuple。一个 spout 可以发射多条流，每条流用 stream id 区分。Spout 的核心方法是 nextTuple()，Storm 会不断调用这个方法，你在这个方法里写读取数据并 emit 的逻辑。还有一个 ack() 和 fail() 方法，用于实现消息的可靠处理。

Bolt 是拓扑的处理单元，它接收 tuple，做计算或转换，然后决定是否发射新的 tuple。Bolt 可以实现过滤、聚合、连接、写数据库等各种逻辑。一个拓扑里可以有多个 bolt，形成多级处理链。

Tuple 是数据在拓扑中流动的基本单位，本质上是一个有序的元素列表，每个元素可以是任意类型。Tuple 在 spout 和 bolt 之间传递，传递的方式由分组策略决定，比如 shuffle grouping 是随机分发，fields grouping 是按字段哈希分发。

Task 是拓扑的执行单元，一个 spout 或 bolt 可以设置多个 task 并行执行。Task 的数量决定了这个组件的并行度。而 worker 是进程级别的，一个 worker 里可以跑多个 executor，每个 executor 对应一个或多个 task。这里的关系是：worker > executor > task。理解这个层级关系，对后面调优并行度和排查资源问题非常关键。

4.2 一个可运行的拓扑示例设计

我设计了一个简单的拓扑来演示整个流程：一个 spout 从 HDFS 读取日志数据，发射给第一个 bolt 做字段解析，解析后的数据再发给第二个 bolt 做统计聚合，最后写入 HDFS。这个拓扑虽然简单，但涵盖了 spout、bolt、tuple、分组策略、可靠性等核心要素。

Spout 的实现上，我用了一个简单的文件读取逻辑，逐行读取 HDFS 上的日志文件，每读一行就 emit 一个 tuple，tuple 的字段就是这一行的内容。为了保证可靠性，我在 emit 的时候带了消息 id，并在 ack 回调里记录偏移量。这样即使某个 tuple 处理失败，也能重新发射。

第一个 bolt 负责解析日志行，按分隔符拆分成多个字段，然后 emit 出去。这里我用了 fields grouping，按日志来源的某个字段做哈希，保证同一来源的数据落到同一个 task 上，方便后续做有状态聚合。

第二个 bolt 做统计，维护一个计数器，每隔一段时间把统计结果写入 HDFS。这个 bolt 的并行度我设成了 2，因为统计逻辑本身不重，两个 task 足够。

拓扑的代码写完之后，用 Maven 打成 jar 包，然后用 storm jar 命令提交：

storm jar my-topology.jar com.example.MyTopology MyTopology

提交之后，可以在 Storm UI 上看到拓扑的运行状态，包括每个 spout 和 bolt 的 emit 数、ack 数、fail 数、延迟等指标。这些指标是排查问题的第一手资料。

4.3 并行度与资源配置的实操计算

并行度的设置是拓扑调优的核心。我以这个示例拓扑为例，说一下我的计算过程。

假设日志数据量是每秒 10000 条，每条日志解析耗时 0.1ms，统计耗时 0.05ms。那么解析 bolt 需要的并行度大约是 10000 * 0.1ms = 1 秒的处理量，一个 task 每秒能处理 10000 条，所以一个 task 就够了。但考虑到峰值和容错，我设了 2 个 task。统计 bolt 同理，设了 2 个 task。

Spout 的并行度取决于数据源的读取速度。如果 HDFS 上一个文件只能被一个 spout 读取，那 spout 的并行度就受限于文件数量。我用了两个 spout task，分别读取不同的文件。

Worker 的数量由 supervisor.slots.ports 决定，每个 supervisor 四个槽位，两个 supervisor 就是八个 worker。拓扑的 worker 数量可以在提交时指定，也可以由 Storm 自动分配。我一般会显式指定，比如 setNumWorkers(4)，这样四个 worker 分布在两个 supervisor 上，每个 supervisor 跑两个 worker。

内存方面，每个 worker 默认 768M，四个 worker 就是 3G。加上 Nimbus、Supervisor、Zookeeper 本身的内存占用，8G 的机器是够用的。但如果拓扑里用了大量缓存或者大对象，就需要调大 worker 的内存，通过 worker.childopts 参数来设置。

5. 常见问题排查与避坑经验实录

5.1 拓扑提交失败与类冲突排查

拓扑提交失败是最常见的问题之一。我遇到过的原因有几种：一是 jar 包里的依赖和 Storm 自带的依赖冲突，比如日志框架的版本不一致，导致 NoSuchMethodError；二是主类名写错或者没有 main 方法；三是 jar 包没有包含必要的依赖，提交时报 ClassNotFoundException。

排查这类问题的思路是：先看 storm jar 命令的输出，如果有异常堆栈，直接定位到具体的类和方法。如果是类冲突，用 mvn dependency:tree 分析依赖树，把冲突的依赖排除掉。我一般会在打包时用 shade 插件把依赖打成一个 fat jar，同时把 Storm 相关的依赖设为 provided，避免和集群里的 jar 冲突。

还有一个坑是序列化问题。Storm 的 tuple 在传输时需要序列化，如果自定义的对象没有实现 Serializable 接口，或者用了不可序列化的字段，运行时会报 NotSerializableException。我的做法是尽量用基本类型和字符串，如果必须用自定义对象，就实现 Serializable 并加上 serialVersionUID。

5.2 Supervisor 无法连接 Nimbus 的排查路径

Supervisor 启动后连不上 Nimbus，表现是 UI 上看不到 supervisor 节点，或者 supervisor 日志里不断报连接超时。这个问题的排查路径我总结了几步。

第一步，检查网络连通性。在 supervisor 节点上 ping nimbus 的主机名，确认能通。如果不通，检查 hosts 文件和 DNS 配置。

第二步，检查端口。在 supervisor 节点上用 telnet 或 nc 测试 nimbus 的 6627 端口是否开放。如果不通，检查 nimbus 是否正常启动，防火墙是否放行。

第三步，检查 Zookeeper。Supervisor 和 Nimbus 之间的协调依赖 Zookeeper，如果 Zookeeper 集群有问题，两者就无法通信。用 zkCli.sh 连接 Zookeeper，查看 /storm 节点下的内容，应该能看到 nimbus 和 supervisor 的注册信息。

第四步，检查 storm.yaml 配置。nimbus.seeds 和 storm.zookeeper.servers 必须配置正确，主机名要和实际一致。我遇到过配置里写的是 IP，但 hosts 文件里没有对应映射，导致解析失败的情况。

5.3 消息丢失与可靠性机制的取舍

Storm 提供了消息可靠性机制，通过 ack 和 fail 来保证 tuple 被完整处理。但这个机制是有代价的，每个 tuple 都要跟踪，会增加内存和网络开销。我在实际使用中的体会是：如果业务对数据丢失不敏感，可以关闭可靠性，把 ack 关掉，性能会提升不少；如果业务要求 exactly-once 或 at-least-once，就必须开启可靠性，并且要确保 spout 的 ack 和 fail 逻辑正确实现。

开启可靠性后，常见的坑是 ack 超时。默认的 ack 超时是 30 秒，如果某个 bolt 处理时间过长，tuple 会超时失败，然后 spout 重新发射。这会导致重复处理。我的做法是根据业务处理时间调整 topology.message.timeout.secs，同时优化 bolt 的处理逻辑，避免单条处理时间过长。

还有一个坑是内存泄漏。如果 spout 发射了 tuple 但没有收到 ack 或 fail，这些 tuple 会一直留在内存里，时间长了就 OOM。排查方法是看 UI 上的 pending 指标，如果持续增长，说明有 tuple 没有被确认。这时候要检查 bolt 的 ack 调用是否遗漏，或者处理逻辑是否抛异常导致没有 ack。

5.4 常见问题速查表

提示：Storm UI 上的指标是排查问题的金矿，尤其是 capacity、execute latency、process latency 这几个指标，能直接反映组件的负载和处理效率。

6. 从安装到运行的完整链路回顾

整套环境搭下来，从 JDK 安装到拓扑运行，涉及的环节不少，但每一步都有明确的逻辑。JDK 是基础，版本必须锁死 1.8；Zookeeper 是协调者，配置要仔细；Storm 本身的配置项不多，但每一项都影响集群行为；拓扑的设计和并行度设置决定了运行效率；排查问题的能力则决定了你能不能在生产环境里稳住。

我在实际操作中的体会是，安装部署本身不难，难的是理解每个配置背后的含义，以及出问题时的排查思路。比如 supervisor.slots.ports 配几个，不是拍脑袋决定的，要根据内存和业务量算；并行度设多少，要看数据量和处理耗时；可靠性开不开，要看业务对数据丢失的容忍度。这些判断没有标准答案，但有一套分析的方法。

最后再分享一个小技巧：在正式提交拓扑之前，先用 local mode 跑一遍。Local mode 下 Storm 会在单个 JVM 里模拟整个拓扑的运行，不需要集群，调试起来非常快。很多逻辑错误和序列化问题在 local mode 下就能暴露出来，省去了反复提交到集群的麻烦。等 local mode 跑通了，再提交到集群，成功率会高很多。

这个内容后续还可以这样扩展：把 spout 的数据源从 HDFS 换成消息队列，bolt 的输出从 HDFS 换成数据库或缓存，拓扑就变成了一个完整的实时处理管道。再进一步，可以加入滑动窗口、状态管理、事务性拓扑等高级特性，应对更复杂的业务场景。