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 负责产生 tuplebolt 负责处理 tuple数据就这样一条条在组件之间传递。我这次搭建的环境是三台机器一台做 nimbus 兼 zookeeper另外两台做 supervisor。操作系统是常见的 Linux 发行版JDK 用 1.8Hadoop 已经提前装好因为后面要演示从 HDFS 读数据写回 HDFS 的流程。下面我会把每一步的操作、参数含义、踩过的坑都摊开来讲尽量让照着做的人少走弯路。2. 基础环境准备与 JDK 安装细节2.1 主机规划与系统层配置三台机器的角色分配是这样的第一台同时跑 Nimbus 和 Zookeeper第二台和第三台跑 Supervisor。每台机器的配置不需要太高我用的都是 4 核 8G磁盘 100G对于学习和中小规模生产足够用了。主机名我分别设为 storm-master、storm-worker01、storm-worker02后面所有配置都基于这三个名字来写。系统层要做的事情不多但每件都不能漏。首先是关闭防火墙或者至少放行 Storm 用到的端口。Storm 涉及的端口比较多Nimbus 默认监听 6627Supervisor 默认监听 6700、6701、6702、6703UI 默认监听 8080Zookeeper 用 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.1storm-master:2888:3888 server.2storm-worker01:2888:3888 server.3storm-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.host1.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 和 UIstorm nimbus storm ui 在 storm-worker01 和 storm-worker02 上启动 Supervisorstorm 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 一个 tupletuple 的字段就是这一行的内容。为了保证可靠性我在 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 常见问题速查表问题现象可能原因排查方法解决方案拓扑提交后立即失败主类名错误或缺少 main 方法查看提交命令输出检查主类全限定名确认有 main 方法运行时 NoSuchMethodError依赖冲突mvn dependency:tree 分析排除冲突依赖用 shade 打包NotSerializableException自定义对象未序列化查看异常堆栈实现 Serializable 接口Supervisor 不显示在 UI网络或 Zookeeper 问题检查端口和 Zookeeper 连接修复网络或 Zookeeper 配置消息重复处理ack 超时查看 UI 的 fail 指标调大超时时间优化处理逻辑Worker 频繁重启内存不足查看 supervisor 日志减少槽位或调大 worker 内存UI 无法访问端口占用或进程未启动检查 8080 端口和 ui 进程更换端口或重启 ui拓扑运行缓慢并行度不足查看 UI 的延迟指标增加 task 或 worker 数量提示Storm UI 上的指标是排查问题的金矿尤其是 capacity、execute latency、process latency 这几个指标能直接反映组件的负载和处理效率。6. 从安装到运行的完整链路回顾整套环境搭下来从 JDK 安装到拓扑运行涉及的环节不少但每一步都有明确的逻辑。JDK 是基础版本必须锁死 1.8Zookeeper 是协调者配置要仔细Storm 本身的配置项不多但每一项都影响集群行为拓扑的设计和并行度设置决定了运行效率排查问题的能力则决定了你能不能在生产环境里稳住。我在实际操作中的体会是安装部署本身不难难的是理解每个配置背后的含义以及出问题时的排查思路。比如 supervisor.slots.ports 配几个不是拍脑袋决定的要根据内存和业务量算并行度设多少要看数据量和处理耗时可靠性开不开要看业务对数据丢失的容忍度。这些判断没有标准答案但有一套分析的方法。最后再分享一个小技巧在正式提交拓扑之前先用 local mode 跑一遍。Local mode 下 Storm 会在单个 JVM 里模拟整个拓扑的运行不需要集群调试起来非常快。很多逻辑错误和序列化问题在 local mode 下就能暴露出来省去了反复提交到集群的麻烦。等 local mode 跑通了再提交到集群成功率会高很多。这个内容后续还可以这样扩展把 spout 的数据源从 HDFS 换成消息队列bolt 的输出从 HDFS 换成数据库或缓存拓扑就变成了一个完整的实时处理管道。再进一步可以加入滑动窗口、状态管理、事务性拓扑等高级特性应对更复杂的业务场景。