Hadoop学习教程，从入门到精通，Storm 完整知识点详解(17）
·
Storm 完整知识点详解
一、什么是 Storm
1.1 Storm 概述
Apache Storm 是一个免费开源、分布式、高容错的实时计算系统。Storm 使得持续不断的流数据流的处理变得简单，类似于 Hadoop 对批量数据（批处理）的处理方式，Storm 处理的是实时流数据（流处理）。
1.2 Storm 核心特点
| 特点 | 说明 |
|---|---|
| 实时性 | 数据产生即处理，毫秒级延迟 |
| 分布式 | 可水平扩展到大规模集群 |
| 容错性 | 自动重启失败的组件，保证数据不丢失 |
| 语言无关 | 支持任何编程语言（Java、Python、Ruby等） |
| 可靠的消息处理 | 保证每条消息至少被处理一次（at-least-once） |
1.3 Storm 与 Hadoop 对比
| 对比项 | Hadoop | Storm |
|---|---|---|
| 处理方式 | 批处理（Batch） | 实时流处理（Streaming） |
| 延迟 | 分钟/小时级 | 毫秒/秒级 |
| 数据模型 | 静态数据集 | 无限持续数据流 |
| 核心进程 | JobTracker/TaskTracker | Nimbus/Supervisor |
| 编程模型 | MapReduce | Spout/Bolt |
| 是否终止 | 处理完数据后任务终止 | 持续运行，永不停止 |
1.4 Storm 应用场景
- 实时数据分析（网站点击流、用户行为分析）
- 实时监控告警（系统日志监控、异常检测）
- 在线机器学习（实时推荐系统）
- 实时ETL（数据实时抽取转换加载）
- 分布式RPC（远程过程调用）
二、Storm 核心概念
2.1 Tuple（元组）
Tuple 是 Storm 中最基本的数据结构，是一个命名的值列表，Tuple 中的字段可以是任意类型的对象。
// 创建一个Tuple示例
// Storm的Tuple是由框架自动生成的，这里展示的是Tuple的使用方式
// 在Spout或Bolt中，emit发送的就是Tuple
// 例如发送一个包含两个字段的Tuple
collector.emit(new Values("hello", 1));
// Values("hello", 1) 就是一个Tuple，包含两个字段
// 第一个字段值为字符串 "hello"
// 第二个字段值为整数 1
2.2 Stream（流）
Stream 是 Storm 中最核心的抽象概念，是一个无限的 Tuple 序列。
Stream 的特征：
1. 无限性 —— 数据持续不断产生
2. 有序性 —— Tuple 按照产生的顺序排列
3. 命名性 —— 每个 Stream 可以定义一个唯一的ID（默认为 "default"）
2.3 Spout（数据源）
Spout 是 Topology 中数据流的源头，负责从外部数据源读取数据并发射到 Topology 中。
/**
* 自定义Spout示例
* 继承 BaseRichSpout 类
* Spout 负责从外部数据源（如Kafka、日志文件等）读取数据
* 并将数据以Tuple的形式发射（emit）到Topology中
*/
public class MySpout extends BaseRichSpout {
// Spout发射器，用于向下游Bolt发送Tuple
private SpoutOutputCollector collector;
// 随机数生成器，用于模拟数据
private Random random;
/**
* open方法：Spout初始化时调用
* 类似于MapReduce中Setup方法，只执行一次
* @param conf Storm配置信息
* @param context Topology上下文，包含Topology相关信息
* @param collector 发射器，用于发送Tuple
*/
@Override
public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
// 初始化SpoutOutputCollector发射器
this.collector = collector;
// 初始化随机数生成器
this.random = new Random();
}
/**
* nextTuple方法：Spout的核心方法
* 会被Storm框架循环调用，不断产生新的Tuple
* 该方法中实现从外部数据源读取数据的逻辑
* 注意：如果暂时没有数据，应该使用Thread.sleep()避免空循环消耗CPU
*/
@Override
public void nextTuple() {
// 模拟产生数据的句子
String[] sentences = {
"hello storm hello world",
"apache storm real time",
"big data streaming processing"
};
// 随机选择一个句子
String sentence = sentences[random.nextInt(sentences.length)];
// 使用collector发射Tuple
// new Values(sentence) 创建一个包含一个字段的Tuple
// 该Tuple会被发送到下游的Bolt进行处理
this.collector.emit(new Values(sentence));
// 休眠100毫秒，避免过快产生数据
Utils.sleep(100);
}
/**
* declareOutputFields方法：声明输出Tuple的字段名
* 这里的字段名用于下游Bolt通过字段名获取对应的值
* @param declarer 输出字段声明器
*/
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
// 声明输出字段名为 "sentence"
// downstream Bolt可以通过 tuple.getValueByField("sentence") 获取值
declarer.declare(new Fields("sentence"));
}
}
2.4 Bolt（处理单元）
Bolt 负责接收 Tuple、处理数据、发射新的 Tuple 到下游 Bolt 或输出结果。
/**
* 自定义Bolt示例：将句子拆分为单词
* 继承 BaseRichBolt 类
* Bolt 负责处理接收到的Tuple，并将处理结果发射到下游
*/
public class SplitBolt extends BaseRichBolt {
// Bolt发射器，用于向下游发送Tuple
private OutputCollector collector;
/**
* prepare方法：Bolt初始化时调用
* 类似于Spout的open方法，只执行一次
* @param stormConf Storm配置信息
* @param context Topology上下文
* @param collector 输出发射器
*/
@Override
public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
// 初始化OutputCollector发射器
this.collector = collector;
}
/**
* execute方法：Bolt的核心方法
* 每接收到一个Tuple，该方法就会被调用一次
* 在此方法中实现具体的处理逻辑
* @param input 接收到的Tuple对象
*/
@Override
public void execute(Tuple input) {
// 通过字段名 "sentence" 获取Tuple中对应的值
String sentence = input.getStringByField("sentence");
// 将句子按空格拆分为单词数组
String[] words = sentence.split(" ");
// 遍历每个单词，发射到下游Bolt
for (String word : words) {
// 发射一个包含单词字段的Tuple
this.collector.emit(new Values(word));
}
// 确认（ack）该Tuple已经被成功处理
// 这是Storm可靠消息处理机制的关键
// 如果不ack，Spout会认为消息处理失败，会重新发送
this.collector.ack(input);
}
/**
* declareOutputFields方法：声明输出Tuple的字段名
* @param declarer 输出字段声明器
*/
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
// 声明输出字段名为 "word"
declarer.declare(new Fields("word"));
}
}
三、Storm Topology（拓扑）
3.1 Topology 概述
Topology 是 Storm 中实时计算逻辑的封装，是一个由 Spout 和 Bolt 组成的 DAG（有向无环图）。
Topology 的特点：
1. 一旦提交，将永不停止运行（除非手动kill）
2. Spout 是数据源节点，Bolt 是处理节点
3. 数据在 Spout 和 Bolt 之间通过 Stream 传递
4. Topology 是 Storm 任务提交的基本单位
3.2 Topology 结构图
┌──────────┐
│ Spout │ ← 数据源（如Kafka）
└────┬─────┘
│ Stream
┌────────┼────────┐
▼ ▼ ▼
┌────────┐ ┌────────┐ ┌────────┐
│ Bolt A │ │ Bolt A │ │ Bolt A │ ← 第一层Bolt（如拆分单词）
└───┬────┘ └───┬────┘ └───┬────┘
│ │ │
└──────────┼──────────┘
▼
┌──────────┐
│ Bolt B │ ← 第二层Bolt（如计数聚合）
└──────────┘
3.3 Topology 构建代码
/**
* 构建并提交Storm Topology
* TopologyBuilder 是构建Topology的核心工具类
* 通过它可以设置Spout和Bolt之间的关系
*/
public class WordCountTopology {
// 定义Spout的ID标识符
private static final String SPOUT_ID = "sentence-spout";
// 定义拆分Bolt的ID标识符
private static final String SPLIT_BOLT_ID = "split-bolt";
// 定义计数Bolt的ID标识符
private static final String COUNT_BOLT_ID = "count-bolt";
public static void main(String[] args) throws Exception {
// 1. 创建TopologyBuilder实例
TopologyBuilder builder = new TopologyBuilder();
// 2. 设置Spout组件
// setSpout方法参数说明：
// 参数1：组件ID（字符串标识符）
// 参数2：Spout实例
// 参数3：并行度（executor数量，即该Spout的线程数）
builder.setSpout(SPOUT_ID, new MySpout(), 2);
// 3. 设置第一个Bolt组件（拆分单词）
// setBolt方法参数说明：
// 参数1：组件ID
// 参数2：Bolt实例
// 参数3：并行度
// shuffleGrouping说明：
// 定义该Bolt的数据来源以及数据分组方式
// 参数1：上游组件ID
// shuffleGrouping表示随机分组，数据均匀随机分配到各task
builder.setBolt(SPLIT_BOLT_ID, new SplitBolt(), 4)
.shuffleGrouping(SPOUT_ID);
// 4. 设置第二个Bolt组件（单词计数）
// fieldsGrouping说明：
// 相同字段值的Tuple会被发送到同一个task中处理
// 这样可以保证同一个单词的计数在同一个task中完成
builder.setBolt(COUNT_BOLT_ID, new CountBolt(), 4)
.fieldsGrouping(SPLIT_BOLT_ID, new Fields("word"));
// 5. 创建Storm配置
Config config = new Config();
// 设置Worker进程数量
config.setNumWorkers(2);
// 设置调试模式
config.setDebug(true);
// 6. 提交Topology
if (args != null && args.length > 0) {
// 集群模式：通过命令行参数指定Topology名称
// StormSubmitter.submitTopology 提交到远程集群
StormSubmitter.submitTopology(args[0], config, builder.createTopology());
} else {
// 本地模式：在本地JVM中运行Topology
// 适合开发和调试
LocalCluster cluster = new LocalCluster();
cluster.submitTopology("word-count-topology", config, builder.createTopology());
// 本地模式运行60秒后关闭
Utils.sleep(60000);
// 关闭本地集群
cluster.shutdown();
}
}
}
3.4 完整单词计数 Bolt 代码
/**
* 单词计数Bolt
* 接收上游Bolt发送的单词，进行计数统计
* 继承 BaseRichBolt 类
*/
public class CountBolt extends BaseRichBolt {
// Bolt发射器
private OutputCollector collector;
// 用于存储单词及其对应的计数
// Key：单词字符串，Value：该单词出现的次数
private Map<String, Long> wordCountMap;
/**
* prepare方法：初始化Bolt
* @param stormConf Storm配置
* @param context Topology上下文
* @param collector 输出发射器
*/
@Override
public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
// 初始化发射器
this.collector = collector;
// 初始化HashMap用于存储单词计数
this.wordCountMap = new HashMap<>();
}
/**
* execute方法：处理接收到的每个Tuple
* 每接收到一个单词Tuple，就在map中进行累加计数
* @param input 接收到的Tuple
*/
@Override
public void execute(Tuple input) {
// 从Tuple中获取 "word" 字段的值
String word = input.getStringByField("word");
// 获取该单词当前的计数，如果不存在则为0
Long count = wordCountMap.getOrDefault(word, 0L);
// 计数加1，并更新到map中
wordCountMap.put(word, count + 1);
// 打印当前单词的计数结果到控制台
System.out.println("单词: " + word + ", 计数: " + (count + 1));
// 确认Tuple处理完成
this.collector.ack(input);
}
/**
* 声明输出字段
* CountBolt作为最终输出节点，不需要声明输出字段
* @param declarer 输出字段声明器
*/
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
// 最终输出节点，可以不声明输出字段
// 如果需要输出，可以声明如 new Fields("word", "count")
}
}
四、Storm 集群架构
4.1 架构组件
┌─────────────────────────────────────────────────────────┐
│ Storm 集群架构 │
├─────────────────────────────────────────────────────────┤
│ │
│ ┌──────────┐ ┌──────────────┐ │
│ │ Nimbus │◄───────►│ ZooKeeper │ │
│ │ (主节点) │ │ (协调服务) │ │
│ └──────────┘ └──────┬───────┘ │
│ │ │
│ ┌──────────────────┼──────────────────┐ │
│ ▼ ▼ ▼ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ Supervisor │ │ Supervisor │ │ Supervisor │ │
│ │ (工作节点) │ │ (工作节点) │ │ (工作节点) │ │
│ ├─────────────┤ ├─────────────┤ ├─────────────┤ │
│ │ Worker │ │ Worker │ │ Worker │ │
│ │ Worker │ │ Worker │ │ Worker │ │
│ │ ┌───────┐ │ │ ┌───────┐ │ │ ┌───────┐ │ │
│ │ │Executor│ │ │ │Executor│ │ │ │Executor│ │ │
│ │ │Executor│ │ │ │Executor│ │ │ │Executor│ │ │
│ │ └───────┘ │ │ └───────┘ │ │ └───────┘ │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
│ │
└─────────────────────────────────────────────────────────┘
4.2 各组件详解
4.2.1 Nimbus（主节点）
Nimbus 是 Storm 集群的"主节点"，负责：
1. 接收客户端提交的Topology
2. 将Topology代码分发到各个Supervisor节点
3. 分配Task给Supervisor
4. 监控Topology的运行状态
5. 处理Topology的失败重分配
特点：
- Nimbus 是无状态的，所有状态存储在ZooKeeper中
- Nimbus 进程挂掉后重启不会影响正在运行的Topology
- Storm 1.x以后支持Nimbus HA（高可用）
4.2.2 Supervisor（工作节点）
Supervisor 是 Storm 集群的"工作节点"，负责：
1. 接收Nimbus分配的任务
2. 管理本节点上的Worker进程
3. 启动或停止Worker进程
4. 通过ZooKeeper与Nimbus通信
一个Supervisor节点可以运行多个Worker进程
4.2.3 Worker（工作进程）
Worker 是运行在Supervisor节点上的JVM进程，负责：
1. 执行Topology的一个子集（一个或多个Executor）
2. 每个Worker进程执行一个Topology的一部分
3. Worker进程之间通过Netty或ZeroMQ通信
一个Topology由多个Worker进程共同完成
4.2.4 Executor（执行线程）
Executor 是 Worker 中的一个线程，负责：
1. 执行一个或多个Task（同一类型的）
2. 默认情况下，一个Executor执行一个Task
3. Executor 是真正执行Spout或Bolt逻辑的线程
4.2.5 Task（任务）
Task 是 Storm 中最小的处理单元：
1. Task 对应一个 Spout 或 Bolt 的实例
2. 默认情况下，一个Executor对应一个Task
3. 可以通过 setNumTasks() 设置一个Executor对应多个Task
4.3 层级关系
Topology
└── Worker（进程级）
└── Executor（线程级）
└── Task（实例级）
举例：
Topology设置为 2个Worker，每个Worker 2个Executor
则总共有 4个Executor（4个线程）
如果每个Executor运行1个Task，则总共有4个Task
4.4 ZooKeeper 在 Storm 中的作用
ZooKeeper 在 Storm 集群中的角色：
1. Nimbus 和 Supervisor 之间的通信桥梁
2. 存储集群状态信息（Topology配置、Task分配等）
3. 监控节点心跳，检测Supervisor是否存活
4. 存储Task的错误信息和统计信息
注意：Storm 本身不存储大量数据在ZooKeeper中
主要存储元数据和协调信息
五、Storm 流分组（Stream Grouping）
5.1 流分组概述
流分组（Stream Grouping）定义了Tuple 如何从 Spout/Bolt 发送到下游 Bolt 的各个 Task。
5.2 七种内置分组方式
5.2.1 Shuffle Grouping（随机分组）
/**
* Shuffle Grouping（随机分组）
*
* 特点：
* Tuple 随机均匀地分发到下游Bolt的各个Task中
* 每个Tuple只发送到一个Task
* 保证每个Task接收到大致相同数量的Tuple
*
* 适用场景：
* 不需要关心数据分配到哪个Task
* 各Task处理逻辑完全独立
*/
builder.setBolt("split-bolt", new SplitBolt(), 4)
// 随机分组：Tuple均匀随机分配到4个split-bolt的Task
.shuffleGrouping("spout-id");
5.2.2 Fields Grouping（字段分组）
/**
* Fields Grouping（字段分组）
*
* 特点：
* 根据指定字段的值进行分组
* 相同字段值的Tuple一定会被发送到同一个Task
* 不同字段值可能被发送到不同Task
* 类似于MapReduce中Partitioner的逻辑
*
* 适用场景：
* 需要保证相同Key的数据由同一个Task处理
* 如：单词计数中，同一单词必须由同一Task计数
*/
builder.setBolt("count-bolt", new CountBolt(), 4)
// 字段分组：相同"word"值的Tuple发送到同一个Task
.fieldsGrouping("split-bolt-id", new Fields("word"));
5.2.3 All Grouping（全复制分组）
/**
* All Grouping（全复制分组）
*
* 特点：
* 每个Tuple被发送到下游Bolt的所有Task
* 相当于广播（broadcast）
*
* 适用场景：
* 每个Task都需要处理所有数据
* 如：全局配置更新、全局缓存刷新
*/
builder.setBolt("all-bolt", new AllBolt(), 4)
// 全复制：每个Tuple发送到所有4个Task
.allGrouping("spout-id");
5.2.4 Global Grouping（全局分组）
/**
* Global Grouping（全局分组）
*
* 特点：
* 所有Tuple都被发送到下游Bolt的某一个Task
* 具体是哪个Task取决于Task ID最小的那个
* 保证全局有序
*
* 适用场景：
* 需要集中处理所有数据
* 如：全局统计、全局排序
*/
builder.setBolt("global-bolt", new GlobalBolt(), 4)
// 全局分组：所有Tuple发送到ID最小的那个Task
.globalGrouping("spout-id");
5.2.5 None Grouping（无分组）
/**
* None Grouping（无分组）
*
* 特点：
* 等价于 Shuffle Grouping
* 目前和随机分组效果相同
* 保留给未来使用
*
* 说明：Storm开发团队保留该分组方式
* 将来可能会有不同的语义
*/
builder.setBolt("none-bolt", new SomeBolt(), 4)
// 无分组：等同于随机分组
.noneGrouping("spout-id");
5.2.6 Direct Grouping（直接分组）
/**
* Direct Grouping（直接分组）
*
* 特点：
* 生产者（Bolt）直接指定Tuple发送到哪个Task
* 需要使用 emitDirect() 方法发送
* 必须知道目标Task的ID
*
* 适用场景：
* 需要精确控制数据路由
*/
// 在Bolt的execute方法中使用直接分组
public class DirectBolt extends BaseRichBolt {
private OutputCollector collector;
// 假设目标Task ID为0
private int targetTaskId = 0;
@Override
public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
this.collector = collector;
}
@Override
public void execute(Tuple input) {
// 直接发送到指定的Task ID
collector.emitDirect(targetTaskId, new Values("direct-data"));
collector.ack(input);
}
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("data"));
}
}
// 在TopologyBuilder中设置
builder.setBolt("direct-bolt", new DirectBolt(), 4)
// 直接分组
.directGrouping("spout-id");
5.2.7 Local or Shuffle Grouping（本地或随机分组）
/**
* Local or Shuffle Grouping（本地或随机分组）
*
* 特点：
* 优先将Tuple发送到同一个Worker进程中的Task
* 如果同一个Worker中没有目标Task，则退化为Shuffle Grouping
* 减少网络传输，提高性能
*
* 适用场景：
* 性能优化场景
* 尽量减少跨Worker的网络通信
*/
builder.setBolt("local-bolt", new SomeBolt(), 4)
// 本地优先随机分组
.localOrShuffleGrouping("spout-id");
5.3 自定义流分组
/**
* 自定义流分组
* 实现 CustomStreamGrouping 接口
* 可以自定义Tuple如何分配到下游Task
*/
public class MyCustomGrouping implements CustomStreamGrouping {
// 存储所有目标Task的ID列表
private List<Integer> targetTasks;
/**
* prepare方法：初始化分组
* @param context Topology上下文
* @param stream 输入流信息
* @param targetTasks 所有可能接收Tuple的Task ID列表
*/
@Override
public void prepare(WorkerTopologyContext context,
GlobalStreamId stream,
List<Integer> targetTasks) {
// 保存目标Task列表
this.targetTasks = targetTasks;
}
/**
* chooseTasks方法：决定Tuple发送到哪个Task
* @param values Tuple的值列表
* @return 目标Task ID的列表
*/
@Override
public List<Integer> chooseTasks(List<Object> values) {
// 获取Tuple的第一个值（假设是字符串）
String value = values.get(0).toString();
// 使用字符串的hash值对Task数量取模
// 保证相同字符串总是路由到同一个Task
int index = Math.abs(value.hashCode()) % targetTasks.size();
// 返回目标Task ID
List<Integer> result = new ArrayList<>();
result.add(targetTasks.get(index));
return result;
}
}
// 在Topology中使用自定义分组
builder.setBolt("custom-bolt", new SomeBolt(), 4)
// 使用自定义分组
.customGrouping("spout-id", new MyCustomGrouping());
5.4 分组方式对比总结
┌──────────────────────┬───────────────────────────────────────┐
│ 分组方式 │ 特点 │
├──────────────────────┼───────────────────────────────────────┤
│ Shuffle Grouping │ 随机均匀分发，每个Tuple到一个Task │
│ Fields Grouping │ 按字段值分组，同值Tuple到同一Task │
│ All Grouping │ 广播，每个Tuple到所有Task │
│ Global Grouping │ 全局，所有Tuple到最小ID的Task │
│ None Grouping │ 等同于Shuffle Grouping │
│ Direct Grouping │ 发送方指定目标Task │
│ Local or Shuffle │ 优先本地，其次随机 │
└──────────────────────┴───────────────────────────────────────┘
六、Storm 集群环境搭建
6.1 环境准备
# 系统要求
# 操作系统：CentOS 7.x / Ubuntu 18.04+
# JDK：JDK 1.8+
# ZooKeeper：3.4.x+
# Storm：2.x（本例使用 apache-storm-2.4.0）
# 1. 配置hosts文件（所有节点）
# 编辑 /etc/hosts
192.168.1.101 nimbus1
192.168.1.102 supervisor1
192.168.1.103 supervisor2
192.168.1.104 supervisor3
6.2 安装 ZooKeeper（所有节点）
# 解压ZooKeeper
tar -zxvf apache-zookeeper-3.7.0-bin.tar.gz -C /opt/module/
# 进入ZooKeeper目录
cd /opt/module/apache-zookeeper-3.7.0-bin
# 创建配置文件
cp conf/zoo_sample.cfg conf/zoo.cfg
# 编辑配置文件
# vi conf/zoo.cfg
# 配置内容如下：
tickTime=2000
initLimit=10
syncLimit=5
dataDir=/opt/module/apache-zookeeper-3.7.0-bin/zkData
clientPort=2181
# 集群配置
server.1=nimbus1:2888:3888
server.2=supervisor1:2888:3888
server.3=supervisor2:2888:3888
# 创建数据目录
mkdir -p /opt/module/apache-zookeeper-3.7.0-bin/zkData
# 在每个节点上创建myid文件
# 节点1 (nimbus1):
echo "1" > /opt/module/apache-zookeeper-3.7.0-bin/zkData/myid
# 节点2 (supervisor1):
echo "2" > /opt/module/apache-zookeeper-3.7.0-bin/zkData/myid
# 节点3 (supervisor2):
echo "3" > /opt/module/apache-zookeeper-3.7.0-bin/zkData/myid
# 启动ZooKeeper（每个节点都要执行）
bin/zkServer.sh start
# 检查状态
bin/zkServer.sh status
6.3 安装 Storm
# 1. 解压Storm安装包（所有节点）
tar -zxvf apache-storm-2.4.0.tar.gz -C /opt/module/
# 2. 配置环境变量（所有节点）
# vi /etc/profile
# 添加以下内容：
export STORM_HOME=/opt/module/apache-storm-2.4.0
export PATH=$PATH:$STORM_HOME/bin
# 使环境变量生效
source /etc/profile
# 3. 编辑Storm配置文件
# vi $STORM_HOME/conf/storm.yaml
6.4 Storm 配置文件详解
########## storm.yaml 配置文件 ##########
# Storm ZooKeeper集群地址
# 多个地址用逗号分隔
storm.zookeeper.servers:
- "nimbus1"
- "supervisor1"
- "supervisor2"
# ZooKeeper端口号
storm.zookeeper.port: 2181
# Storm本地数据存储目录
# 存储Topology的jar包、配置等
storm.local.dir: "/opt/module/apache-storm-2.4.0/storm-data"
# Nimbus节点地址
# 客户端通过该地址提交Topology
nimbus.host: "nimbus1"
# Nimbus RPC端口
nimbus.thrift.port: 6627
# Supervisor管理的端口范围
# 每个Worker进程占用一个端口
supervisor.slots.ports:
- 6700
- 6701
- 6702
- 6703
# Worker进程的堆内存大小（单位：MB）
worker.heap.memory.mb: 768
# Worker进程之间的通信方式
# 可选值："netty" 或 "zmq"（推荐netty）
storm.messaging.transport: "org.apache.storm.messaging.netty.Context"
# UI监控端口
ui.port: 8080
# 日志目录
storm.log.dir: "/opt/module/apache-storm-2.4.0/logs"
6.5 启动集群
# 1. 启动ZooKeeper集群（每个ZK节点执行）
zkServer.sh start
# 2. 在Nimbus节点上启动Nimbus进程
# 前台启动（调试用）：
storm nimbus
# 后台启动（生产用）：
nohup storm nimbus &
# 3. 在Nimbus节点上启动UI监控服务
nohup storm ui &
# 4. 在每个Supervisor节点上启动Supervisor进程
nohup storm supervisor &
# 5. 在Nimbus节点上启动Logviewer（日志查看服务）
nohup storm logviewer &
# 6. 验证集群状态
# 查看进程
jps
# 浏览器访问Storm UI
# http://nimbus1:8080
6.6 提交 Topology
# 打包Topology为jar文件
mvn clean package
# 提交Topology到集群
storm jar target/storm-demo-1.0.jar \
com.example.WordCountTopology \
word-count-topology
# 查看当前运行的Topology
storm list
# 停止Topology
storm kill word-count-topology
# 重新平衡Topology（调整并行度）
storm rebalance word-count-topology -n 4 -e split-bolt=6
七、案例分析一：单词计数（Word Count）
7.1 设计思路
整体流程：
1. SentenceSpout → 随机产生英文句子
2. SplitBolt → 将句子拆分为单词
3. CountBolt → 统计每个单词出现的次数
数据流向：
SentenceSpout --shuffleGrouping--> SplitBolt --fieldsGrouping--> CountBolt
组件说明：
┌────────────────┐ ┌──────────────┐ ┌──────────────┐
│ SentenceSpout │────>│ SplitBolt │────>│ CountBolt │
│ 发射句子Tuple │ │ 拆分单词 │ │ 统计计数 │
│ 输出：sentence │ │ 输出：word │ │ 输出：无 │
└────────────────┘ └──────────────┘ └──────────────┘
7.2 Maven 项目配置
<!-- pom.xml -->
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.example</groupId>
<artifactId>storm-wordcount</artifactId>
<version>1.0-SNAPSHOT</version>
<properties>
<!-- Storm版本 -->
<storm.version>2.4.0</storm.version>
<!-- Java编译版本 -->
<maven.compiler.source>1.8</maven.compiler.source>
<maven.compiler.target>1.8</maven.compiler.target>
<!-- 项目编码 -->
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
<!-- Storm核心依赖 -->
<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-client</artifactId>
<version>${storm.version}</version>
<!-- provided表示该依赖在集群上已经存在，打包时不包含 -->
<scope>provided</scope>
</dependency>
<!-- Storm服务端依赖（本地模式运行需要） -->
<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-server</artifactId>
<version>${storm.version}</version>
<scope>provided</scope>
</dependency>
</dependencies>
<build>
<plugins>
<!-- Maven打包插件 -->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<version>3.2.4</version>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>
7.3 完整代码编写
7.3.1 SentenceSpout（数据源）
package com.example.storm.wordcount;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;
import org.apache.storm.utils.Utils;
import java.util.Map;
import java.util.Random;
/**
* SentenceSpout - 数据源组件
*
* 功能：随机产生英文句子，发射到Topology中
* 继承 BaseRichSpout，这是Storm提供的Spout基础实现类
* BaseRichSpout 提供了IRichSpout接口的默认实现
* 开发者只需要重写需要的方法即可
*/
public class SentenceSpout extends BaseRichSpout {
// ============================================
// 成员变量定义
// ============================================
/**
* SpoutOutputCollector：Spout的输出收集器
* 用于将Tuple发送到下游的Bolt
* 在open方法中由框架注入
*/
private SpoutOutputCollector collector;
/**
* Random：随机数生成器
* 用于随机选择句子
*/
private Random random;
/**
* 模拟数据源的句子数组
* 实际项目中，这里替换为从Kafka、数据库等读取数据
*/
private String[] sentences = {
"hello storm hello world",
"apache storm is a real time",
"big data processing framework",
"storm is fast and reliable",
"hello apache storm hello",
"real time stream processing",
"distributed and fault tolerant"
};
// ============================================
// 核心方法实现
// ============================================
/**
* open方法 - 初始化方法
*
* 调用时机：Spout被初始化时调用一次
* 功能：初始化资源（如数据库连接、文件句柄等）
*
* @param conf Storm配置（storm.yaml中的配置项）
* @param context Topology上下文信息
* 包含Topology ID、Task ID、Worker信息等
* @param collector 输出收集器，用于向下游Bolt发送Tuple
*/
@Override
public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
// 将框架传入的collector保存为成员变量
// 后续在nextTuple方法中使用它来发射数据
this.collector = collector;
// 初始化随机数生成器
this.random = new Random();
System.out.println("SentenceSpout 初始化完成，Task ID: " + context.getThisTaskId());
}
/**
* nextTuple方法 - 核心数据发射方法
*
* 调用时机：被Storm框架循环调用
* 功能：产生/读取数据并发射Tuple
*
* 注意事项：
* 1. 该方法会被不断循环调用，如果没有数据应休眠
* 2. 不要使用while(true)阻塞，会占用过多CPU
* 3. 每次调用应该发射一个或零个Tuple
*/
@Override
public void nextTuple() {
// 随机选择一个句子
String sentence = sentences[random.nextInt(sentences.length)];
// 使用collector发射一个Tuple
// new Values(sentence) 等价于 new Values("hello storm hello world")
// Values是ArrayList的子类，用于封装Tuple的字段值
//
// emit方法默认使用 "default" 流
// 下游Bolt通过 shuffleGrouping("spout-id") 接收
this.collector.emit(new Values(sentence));
// 休眠500毫秒，控制数据发射频率
// 避免产生过多数据导致系统过载
// 生产环境中通常不需要休眠，因为数据来自外部系统
Utils.sleep(500);
}
/**
* declareOutputFields方法 - 声明输出字段
*
* 功能：告诉Storm框架该Spout发射的Tuple包含哪些字段
* 这些字段名用于下游Bolt通过字段名获取值
*
* @param declarer 输出字段声明器
*/
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
// 声明输出包含一个字段，名为 "sentence"
// 下游Bolt中可以通过以下方式获取值：
// tuple.getStringByField("sentence")
// 或 tuple.getString(0) -- 通过索引获取
declarer.declare(new Fields("sentence"));
}
}
7.3.2 SplitBolt（拆分单词）
package com.example.storm.wordcount;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import java.util.Map;
/**
* SplitBolt - 单词拆分组件
*
* 功能：接收上游Spout发送的句子Tuple
* 将句子拆分为单个单词
* 将每个单词作为一个新的Tuple发射到下游
*
* 继承 BaseRichBolt，这是Storm提供的Bolt基础实现类
*/
public class SplitBolt extends BaseRichBolt {
// ============================================
// 成员变量定义
// ============================================
/**
* OutputCollector：Bolt的输出收集器
* 用于向下游Bolt发送Tuple
* 也用于ack（确认）/fail（失败）上游的Tuple
*/
private OutputCollector collector;
// ============================================
// 核心方法实现
// ============================================
/**
* prepare方法 - 初始化方法
*
* 调用时机：Bolt被初始化时调用一次
* 功能：初始化资源
*
* 与Spout的open方法类似，但参数类型不同
* Bolt使用 OutputCollector，Spout使用 SpoutOutputCollector
*
* @param stormConf Storm配置信息
* @param context Topology上下文信息
* @param collector 输出收集器
*/
@Override
public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
// 保存collector到成员变量
this.collector = collector;
System.out.println("SplitBolt 初始化完成，Task ID: " + context.getThisTaskId());
}
/**
* execute方法 - 核心处理方法
*
* 调用时机：每接收到一个Tuple，该方法被调用一次
* 功能：处理接收到的数据，产生新的Tuple发送到下游
*
* 重要：处理完成后必须调用 collector.ack(input) 确认
* 否则Spout会认为消息处理失败，触发重发机制
*
* @param input 接收到的上游Tuple
*/
@Override
public void execute(Tuple input) {
// 第1步：从接收到的Tuple中获取句子
// getStringByField通过字段名获取对应的值
// 这里的 "sentence" 对应 SentenceSpout 中声明的字段名
String sentence = input.getStringByField("sentence");
// 第2步：将句子按空格拆分为单词数组
// split(" ") 以空格为分隔符拆分字符串
// 例如 "hello world" → ["hello", "world"]
String[] words = sentence.split(" ");
// 第3步：遍历每个单词，发射到下游Bolt
for (String word : words) {
// 过滤空白字符串（处理连续空格的情况）
if (word != null && !word.trim().isEmpty()) {
// 发射一个包含单词的Tuple到下游
// new Values(word.trim()) 创建包含一个字段的Tuple
// word.trim() 去除单词两端的空白字符
this.collector.emit(new Values(word.trim()));
}
}
// 第4步：确认Tuple处理完成
// ack（acknowledge）告诉Storm框架该Tuple已被成功处理
// Storm的可靠性机制会追踪每个Tuple的处理状态
// 如果Tuple处理链路中某个环节失败，Spout可以重新发送
this.collector.ack(input);
}
/**
* declareOutputFields方法 - 声明输出字段
*
* @param declarer 输出字段声明器
*/
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
// 声明输出包含一个字段，名为 "word"
// 下游CountBolt通过 tuple.getStringByField("word") 获取单词
declarer.declare(new Fields("word"));
}
}
7.3.3 CountBolt（单词计数）
package com.example.storm.wordcount;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import java.util.HashMap;
import java.util.Map;
/**
* CountBolt - 单词计数组件
*
* 功能：接收上游SplitBolt发送的单词Tuple
* 统计每个单词出现的次数
* 输出计数结果
*
* 注意：该Bolt使用 fieldsGrouping 接收数据
* 保证相同单词总是由同一个Task处理
* 这样才能正确地进行计数累加
*/
public class CountBolt extends BaseRichBolt {
// ============================================
// 成员变量定义
// ============================================
/**
* OutputCollector：输出收集器
*/
private OutputCollector collector;
/**
* wordCountMap：单词计数存储
* Key：单词字符串
* Value：该单词累计出现的次数
*
* 因为使用了fieldsGrouping，相同的word总是发送到同一个Task
* 所以同一个Task中的wordCountMap只包含部分单词的计数
* 这不影响最终结果的正确性
*/
private Map<String, Long> wordCountMap;
// ============================================
// 核心方法实现
// ============================================
/**
* prepare方法 - 初始化方法
*
* @param stormConf Storm配置
* @param context Topology上下文
* @param collector 输出收集器
*/
@Override
public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
// 初始化输出收集器
this.collector = collector;
// 初始化单词计数Map
// 使用HashMap存储每个单词的累计计数
this.wordCountMap = new HashMap<>();
System.out.println("CountBolt 初始化完成，Task ID: " + context.getThisTaskId());
}
/**
* execute方法 - 核心处理方法
*
* 每接收到一个单词Tuple，更新该单词的计数
*
* @param input 接收到的Tuple
*/
@Override
public void execute(Tuple input) {
// 第1步：从Tuple中获取单词
// "word" 对应 SplitBolt 中声明的字段名
String word = input.getStringByField("word");
// 第2步：获取该单词当前的计数
// getOrDefault方法：如果key存在返回对应的value，否则返回默认值0L
// 0L 表示Long类型的0
Long count = wordCountMap.getOrDefault(word, 0L);
// 第3步：计数加1并更新到Map中
wordCountMap.put(word, count + 1);
// 第4步：输出结果
// 格式："单词: hello, 计数: 3"
System.out.println("【单词计数结果】单词: " + word + ", 计数: " + (count + 1));
// 第5步：发射结果Tuple到下游（可选）
// 如果还有下游Bolt，可以通过emit发送
// 如果是最终Bolt，可以不发射或发射到输出
this.collector.emit(new Values(word, count + 1));
// 第6步：确认Tuple处理完成
this.collector.ack(input);
}
/**
* declareOutputFields方法 - 声明输出字段
*
* @param declarer 输出字段声明器
*/
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
// 声明输出字段：单词和计数
declarer.declare(new Fields("word", "count"));
}
}
7.3.4 WordCountTopology（拓扑构建与提交）
package com.example.storm.wordcount;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;
import org.apache.storm.utils.Utils;
/**
* WordCountTopology - 单词计数拓扑
*
* 功能：组装Spout和Bolt，构建Topology并提交运行
*
* 拓扑结构：
* SentenceSpout ──(shuffle)──> SplitBolt ──(fields:word)──> CountBolt
*
* 并行度设置：
* SentenceSpout: 2个executor（2个线程并行产生数据）
* SplitBolt: 4个executor（4个线程并行拆分单词）
* CountBolt: 4个executor（4个线程并行计数）
*/
public class WordCountTopology {
// ============================================
// 组件ID常量定义
// ============================================
/** Spout组件ID */
private static final String SPOUT_ID = "sentence-spout";
/** 拆分Bolt组件ID */
private static final String SPLIT_BOLT_ID = "split-bolt";
/** 计数Bolt组件ID */
private static final String COUNT_BOLT_ID = "count-bolt";
/** Topology名称 */
private static final String TOPOLOGY_NAME = "word-count-topology";
// ============================================
// 主方法
// ============================================
/**
* 主方法：构建并提交Topology
*
* @param args 命令行参数
* 本地模式：不传参数
* 集群模式：args[0] = topology名称
* @throws Exception 可能抛出的异常
*/
public static void main(String[] args) throws Exception {
// ============================================
// 第1步：创建TopologyBuilder
// ============================================
// TopologyBuilder 是构建Topology的核心工具类
// 它提供了设置Spout和Bolt的方法
TopologyBuilder builder = new TopologyBuilder();
// ============================================
// 第2步：设置Spout组件
// ============================================
// setSpout方法说明：
// 参数1 (SPOUT_ID): 组件的唯一标识符
// 参数2 (new SentenceSpout()): Spout实例
// 参数3 (2): 并行度，即executor数量（线程数）
//
// 每个executor运行一个Spout实例
// 设置为2表示有2个线程同时运行SentenceSpout
builder.setSpout(SPOUT_ID, new SentenceSpout(), 2);
// ============================================
// 第3步：设置SplitBolt组件
// ============================================
// setBolt方法说明：
// 参数1 (SPLIT_BOLT_ID): Bolt的唯一标识符
// 参数2 (new SplitBolt()): Bolt实例
// 参数3 (4): 并行度
//
// shuffleGrouping(SPOUT_ID):
// 数据来源：SPOUT_ID对应的SentenceSpout
// 分组方式：shuffleGrouping（随机分组）
// 效果：Spout发射的Tuple均匀随机分配到4个SplitBolt Task
builder.setBolt(SPLIT_BOLT_ID, new SplitBolt(), 4)
.shuffleGrouping(SPOUT_ID);
// ============================================
// 第4步：设置CountBolt组件
// ============================================
// fieldsGrouping(SPLIT_BOLT_ID, new Fields("word")):
// 数据来源：SPLIT_BOLT_ID对应的SplitBolt
// 分组方式：fieldsGrouping（字段分组）
// 分组字段："word"
// 效果：相同单词的Tuple一定会被发送到同一个CountBolt Task
// 这保证了同一个单词的计数在同一个Task中累加
//
// 例如：
// 单词"hello" 始终发送到 CountBolt Task 0
// 单词"storm" 始终发送到 CountBolt Task 1
builder.setBolt(COUNT_BOLT_ID, new CountBolt(), 4)
.fieldsGrouping(SPLIT_BOLT_ID, new Fields("word"));
// ============================================
// 第5步：创建配置对象
// ============================================
Config config = new Config();
// 设置Worker进程数
// 整个Topology运行在2个Worker进程中
// 每个Worker是一个独立的JVM进程
config.setNumWorkers(2);
// 开启调试模式
// 在调试模式下，Storm会记录更多的日志信息
config.setDebug(true);
// 设置消息超时时间（秒）
// 如果一个Tuple在30秒内没有被完整处理，视为超时
// 超时后Spout会重新发送该Tuple
config.setMessageTimeoutSecs(30);
// ============================================
// 第6步：提交Topology
// ============================================
if (args != null && args.length > 0) {
// ---- 集群模式 ----
// 当通过命令行传入参数时，使用集群模式提交
// StormSubmitter.submitTopology() 将Topology提交到远程Storm集群
// args[0] 作为Topology的名称
StormSubmitter.submitTopology(args[0], config, builder.createTopology());
System.out.println("Topology [" + args[0] + "] 已提交到集群");
} else {
// ---- 本地模式 ----
// 不传参数时使用本地模式
// LocalCluster 在本地JVM中模拟一个Storm集群
// 适合开发和调试阶段使用
System.out.println("启动本地模式...");
// 创建本地集群实例
LocalCluster cluster = new LocalCluster();
// 提交Topology到本地集群
cluster.submitTopology(TOPOLOGY_NAME, config, builder.createTopology());
System.out.println("Topology [" + TOPOLOGY_NAME + "] 已提交到本地集群");
// 本地模式运行60秒
// 实际生产中不会这样写，Topology会一直运行
Utils.sleep(60000);
// 关闭Topology
cluster.killTopology(TOPOLOGY_NAME);
System.out.println("Topology [" + TOPOLOGY_NAME + "] 已关闭");
// 关闭本地集群
cluster.shutdown();
System.out.println("本地集群已关闭");
}
}
}
7.4 程序运行
# ===== 本地模式运行 =====
# 直接运行main方法，不传参数即可
# 在IDE中右键 Run 或执行：
mvn clean compile exec:java -Dexec.mainClass="com.example.storm.wordcount.WordCountTopology"
# ===== 集群模式运行 =====
# 1. 打包
mvn clean package
# 2. 提交到Storm集群
storm jar target/storm-wordcount-1.0-SNAPSHOT.jar \
com.example.storm.wordcount.WordCountTopology \
word-count-topology
# 3. 查看运行结果
# 方式一：查看Worker日志
tail -f $STORM_HOME/logs/workers-artifacts/word-count-topology/*/worker.log
# 方式二：访问Storm UI
# 浏览器打开 http://nimbus1:8080
# 4. 停止Topology
storm kill word-count-topology
预期输出结果示例：
========================================
【单词计数结果】单词: hello, 计数: 1
【单词计数结果】单词: storm, 计数: 1
【单词计数结果】单词: hello, 计数: 2
【单词计数结果】单词: world, 计数: 1
【单词计数结果】单词: apache, 计数: 1
【单词计数结果】单词: storm, 计数: 2
【单词计数结果】单词: is, 计数: 1
【单词计数结果】单词: a, 计数: 1
【单词计数结果】单词: real, 计数: 1
【单词计数结果】单词: time, 计数: 1
【单词计数结果】单词: hello, 计数: 3
...
========================================
八、案例分析二：Storm 与 Kafka 整合
8.1 设计思路
整体架构：
Kafka Topic ──> KafkaSpout ──> SplitBolt ──> CountBolt ──> 存储/输出
数据流向：
┌──────────┐ ┌────────────┐ ┌────────────┐ ┌────────────┐
│ Kafka │───>│ KafkaSpout │───>│ SplitBolt │───>│ CountBolt │
│ Topic │ │ 读取Kafka │ │ 拆分单词 │ │ 单词计数 │
│(数据源) │ │ 消息 │ │ │ │ │
└──────────┘ └────────────┘ └────────────┘ └────────────┘
组件说明：
1. KafkaSpout：Storm提供的官方Kafka集成Spout
从Kafka Topic中消费消息
自动管理消费偏移量（offset）
支持消息可靠处理（at-least-once语义）
2. SplitBolt： 将Kafka消息（句子）拆分为单词
3. CountBolt： 统计每个单词的出现次数
Kafka Topic设计：
Topic名称：storm-input
分区数：3
消息格式：每条消息为一个英文句子
8.2 Maven 依赖配置
<!-- pom.xml -->
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.example</groupId>
<artifactId>storm-kafka-integration</artifactId>
<version>1.0-SNAPSHOT</version>
<properties>
<storm.version>2.4.0</storm.version>
<kafka.version>3.3.1</kafka.version>
<maven.compiler.source>1.8</maven.compiler.source>
<maven.compiler.target>1.8</maven.compiler.target>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
<!-- ==================== Storm 核心依赖 ==================== -->
<!-- Storm客户端库：包含TopologyBuilder、Config等核心类 -->
<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-client</artifactId>
<version>${storm.version}</version>
<scope>provided</scope>
</dependency>
<!-- Storm服务端库：本地模式运行需要 -->
<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-server</artifactId>
<version>${storm.version}</version>
<scope>provided</scope>
</dependency>
<!-- ==================== Storm-Kafka 集成依赖 ==================== -->
<!-- Storm官方提供的Kafka集成组件 -->
<!-- 包含KafkaSpout、KafkaSpoutConfig等类 -->
<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-kafka-client</artifactId>
<version>${storm.version}</version>
</dependency>
<!-- ==================== Kafka 客户端依赖 ==================== -->
<!-- Kafka Java客户端库 -->
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>${kafka.version}</version>
</dependency>
</dependencies>
<build>
<plugins>
<!-- Maven Shade插件：将所有依赖打成一个fat jar -->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<version>3.2.4</version>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
<configuration>
<transformers>
<!-- 解决多个jar中META-INF冲突问题 -->
<transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
<mainClass>com.example.storm.kafka.KafkaWordCountTopology</mainClass>
</transformer>
</transformers>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>
8.3 Kafka 环境准备
# 1. 创建Kafka Topic
# 进入Kafka安装目录
cd /opt/module/kafka_2.13-3.3.1
# 创建Topic：storm-input
# --partitions 3：创建3个分区
# --replication-factor 1：副本因子为1（测试环境）
bin/kafka-topics.sh --create \
--bootstrap-server localhost:9092 \
--topic storm-input \
--partitions 3 \
--replication-factor 1
# 2. 验证Topic创建成功
bin/kafka-topics.sh --describe \
--bootstrap-server localhost:9092 \
--topic storm-input
# 3. 启动生产者（用于测试发送数据）
bin/kafka-console-producer.sh \
--bootstrap-server localhost:9092 \
--topic storm-input
# 在生产者控制台输入测试数据：
# hello storm hello world
# apache storm is great
# big data streaming processing
# hello kafka integration with storm
8.4 完整代码编写
8.4.1 SplitSentenceBolt（拆分句子Bolt）
package com.example.storm.kafka;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import java.util.Map;
/**
* SplitSentenceBolt - 句子拆分Bolt
*
* 功能：
* 接收KafkaSpout发送的Kafka消息（一个Tuple对应Kafka中的一条消息）
* 将消息内容（句子）拆分为单个单词
* 将每个单词发射到下游CountBolt
*
* 数据流转：
* KafkaSpout ──Tuple(value="hello storm")──> SplitSentenceBolt
* SplitSentenceBolt ──Tuple(word="hello")──> CountBolt
* SplitSentenceBolt ──Tuple(word="storm")──> CountBolt
*/
public class SplitSentenceBolt extends BaseRichBolt {
// 输出收集器
private OutputCollector collector;
/**
* 初始化方法
*
* @param stormConf Storm配置
* @param context Topology上下文
* @param collector 输出收集器
*/
@Override
public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
// 保存输出收集器
this.collector = collector;
}
/**
* 处理接收到的Tuple
*
* KafkaSpout发送的Tuple中，消息内容存储在 "value" 字段中
* 也可以通过 tuple.getValue(0) 获取第一条消息内容
*
* @param input 接收到的Tuple
*/
@Override
public void execute(Tuple input) {
try {
// 第1步：从Tuple中获取Kafka消息内容
// KafkaSpout默认将消息的value存储在 "value" 字段中
// getStringByField("value") 获取消息的字符串值
String message = input.getStringByField("value");
// 打印接收到的原始消息（调试用）
System.out.println("【SplitBolt】接收到Kafka消息: " + message);
// 第2步：将消息（句子）拆分为单词
// 使用正则表达式 "\\s+" 匹配一个或多个空白字符
// 这样可以处理连续空格、制表符等情况
String[] words = message.split("\\s+");
// 第3步：遍历单词并发射
for (String word : words) {
// 去除两端空白和标点符号
// 转换为小写，保证计数时大小写一致
String cleanWord = word.trim().toLowerCase()
.replaceAll("[^a-zA-Z0-9]", "");
// 过滤空字符串
if (!cleanWord.isEmpty()) {
// 发射单词Tuple到下游CountBolt
// 使用 "word" 字段名
this.collector.emit(new Values(cleanWord));
}
}
// 第4步：确认Tuple处理成功
this.collector.ack(input);
} catch (Exception e) {
// 异常处理：标记Tuple处理失败
// fail方法告诉Storm框架该Tuple处理失败
// Storm会通知KafkaSpout进行重试
System.err.println("【SplitBolt】处理Tuple异常: " + e.getMessage());
this.collector.fail(input);
}
}
/**
* 声明输出字段
*
* @param declarer 输出字段声明器
*/
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
// 声明输出字段为 "word"
declarer.declare(new Fields("word"));
}
}
8.4.2 WordCountBolt（单词计数Bolt）
package com.example.storm.kafka;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;
import java.util.HashMap;
import java.util.Map;
/**
* WordCountBolt - 单词计数Bolt
*
* 功能：
* 接收SplitSentenceBolt发送的单词Tuple
* 使用HashMap统计每个单词的出现次数
* 实时打印计数结果
*
* 注意：
* 该Bolt作为最终处理节点，不需要发射Tuple到下游
* 所以不需要在execute中调用emit方法
* 也不需要声明输出字段
*/
public class WordCountBolt extends BaseRichBolt {
// 输出收集器
private OutputCollector collector;
/**
* 单词计数Map
* 使用HashMap存储每个单词的出现次数
* Key：单词字符串
* Value：累计出现次数
*/
private Map<String, Long> wordCountMap;
/**
* 初始化方法
*
* @param stormConf Storm配置
* @param context Topology上下文
* @param collector 输出收集器
*/
@Override
public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
// 初始化输出收集器
this.collector = collector;
// 初始化计数Map
this.wordCountMap = new HashMap<>();
}
/**
* 处理接收到的单词Tuple
*
* @param input 接收到的Tuple
*/
@Override
public void execute(Tuple input) {
try {
// 第1步：获取单词
String word = input.getStringByField("word");
// 第2步：获取当前计数并加1
// getOrDefault：如果Map中存在该key则返回对应value，否则返回默认值0L
long newCount = wordCountMap.getOrDefault(word, 0L) + 1;
// 第3步：更新Map中的计数
wordCountMap.put(word, newCount);
// 第4步：打印实时计数结果
// 输出格式：【Kafka词频统计】单词: hello, 当前计数: 3
System.out.println("【Kafka词频统计】单词: " + word +
", 当前计数: " + newCount +
", 总共统计单词数: " + wordCountMap.size());
// 第5步：确认处理成功
this.collector.ack(input);
} catch (Exception e) {
// 异常处理：标记Tuple处理失败
System.err.println("【CountBolt】处理Tuple异常: " + e.getMessage());
this.collector.fail(input);
}
}
/**
* 声明输出字段
*
* 作为最终Bolt，没有下游Bolt，可以不声明输出字段
*
* @param declarer 输出字段声明器
*/
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
// 最终Bolt，不声明输出字段
// 如果需要输出到外部存储，可以在此声明字段
}
}
8.4.3 KafkaWordCountTopology（拓扑构建）
package com.example.storm.kafka;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.kafka.spout.KafkaSpout;
import org.apache.storm.kafka.spout.KafkaSpoutConfig;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;
import org.apache.storm.utils.Utils;
/**
* KafkaWordCountTopology - Storm与Kafka整合的单词计数拓扑
*
* 功能：
* 从Kafka Topic中消费消息
* 实现单词拆分和计数统计
*
* 拓扑结构：
* KafkaSpout ──(shuffle)──> SplitSentenceBolt ──(fields:word)──> WordCountBolt
*
* 关键知识点：
* 1. KafkaSpoutConfig：配置Kafka消费者参数
* 2. KafkaSpout：从Kafka消费消息的Spout
* 3. 消费者组：Storm使用消费者组管理Kafka偏移量
*/
public class KafkaWordCountTopology {
// ============================================
// 常量定义
// ============================================
/** Kafka Broker地址 */
private static final String KAFKA_BROKERS = "nimbus1:9092,supervisor1:9092,supervisor2:9092";
/** Kafka Topic名称 */
private static final String KAFKA_TOPIC = "storm-input";
/** Kafka消费者组ID */
private static final String KAFKA_GROUP_ID = "storm-consumer-group";
/** Topology名称 */
private static final String TOPOLOGY_NAME = "kafka-word-count-topology";
/** 组件ID */
private static final String KAFKA_SPOUT_ID = "kafka-spout";
private static final String SPLIT_BOLT_ID = "split-bolt";
private static final String COUNT_BOLT_ID = "count-bolt";
// ============================================
// 主方法
// ============================================
/**
* 主方法：构建并提交Kafka-Storm整合拓扑
*
* @param args 命令行参数
* @throws Exception 异常
*/
public static void main(String[] args) throws Exception {
// ============================================
// 第1步：创建TopologyBuilder
// ============================================
TopologyBuilder builder = new TopologyBuilder();
// ============================================
// 第2步：配置KafkaSpout
// ============================================
/**
* 创建KafkaSpoutConfig
*
* KafkaSpoutConfig是配置KafkaSpout的核心类
* 它定义了Kafka消费者的连接参数和行为
*/
KafkaSpoutConfig<String, String> kafkaSpoutConfig = KafkaSpoutConfig
// 设置Kafka Broker地址和Topic
// firstBootstrapBrokers：Kafka Broker列表
// withStringDeserializer：Key和Value的反序列化器
.builder(KAFKA_BROKERS, KAFKA_TOPIC)
// 设置消费者组ID
// Storm使用该ID在Kafka中管理消费偏移量（offset）
// 不同的消费者组独立消费，互不影响
.setGroupId(KAFKA_GROUP_ID)
// 设置Kafka消费者属性
// 处理偏移量重置策略
// earliest：从最早的可用消息开始消费
// 可选值："earliest"、"latest"、"none"
.setProp(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
// 设置每次拉取的最大记录数
// 控制每次poll()方法返回的最大消息数
// 避免一次拉取过多消息导致内存溢出
.setProp(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500)
// 设置记录处理的重试策略
// 第一个参数：重试次数（1表示最多重试1次）
// 第二个参数：重试间隔时间
.setRetry(new KafkaSpoutConfig.RetryService(1, 2000))
// 构建配置对象
.build();
// ============================================
// 第3步：设置Spout和Bolt组件
// ============================================
// 设置KafkaSpout
// KafkaSpout是Storm官方提供的Kafka集成Spout
// 它封装了Kafka消费者，自动处理：
// - 消费者与Kafka的连接
// - 消息消费和偏移量管理
// - 消息可靠处理（at-least-once）
// - 消费者Rebalance
builder.setSpout(KAFKA_SPOUT_ID, new KafkaSpout<>(kafkaSpoutConfig), 3);
// 并行度为3，对应Kafka Topic的3个分区
// 每个KafkaSpout Task处理一个Kafka分区
// 设置SplitSentenceBolt
// 随机分组：KafkaSpout的消息均匀分配到各个SplitBolt Task
builder.setBolt(SPLIT_BOLT_ID, new SplitSentenceBolt(), 4)
.shuffleGrouping(KAFKA_SPOUT_ID);
// shuffleGrouping确保消息均匀分配到4个SplitBolt Task
// 每条消息只被一个SplitBolt Task处理
// 设置WordCountBolt
// 字段分组：保证相同单词由同一个CountBolt Task处理
builder.setBolt(COUNT_BOLT_ID, new WordCountBolt(), 4)
.fieldsGrouping(SPLIT_BOLT_ID, new Fields("word"));
// fieldsGrouping("word") 确保：
// 单词"hello"始终发送到同一个CountBolt Task
// 单词"storm"始终发送到同一个CountBolt Task
// 这样每个Task都能正确累计对应单词的计数
// ============================================
// 第4步：创建配置
// ============================================
Config config = new Config();
// 设置Worker进程数
config.setNumWorkers(3);
// 设置消息超时时间
// 如果Tuple在指定时间内没有被ack，视为处理失败
config.setMessageTimeoutSecs(60);
// 开启调试模式（生产环境应关闭）
config.setDebug(false);
// 设置每个Topology的Kafka客户端最大偏移量滞后
// 用于监控Kafka消费延迟
// ============================================
// 第5步：提交Topology
// ============================================
if (args != null && args.length > 0) {
// 集群模式
StormSubmitter.submitTopology(args[0], config, builder.createTopology());
System.out.println("Topology [" + args[0] + "] 已提交到集群");
} else {
// 本地模式
System.out.println("以本地模式启动 Kafka-Storm 单词计数...");
LocalCluster cluster = new LocalCluster();
cluster.submitTopology(TOPOLOGY_NAME, config, builder.createTopology());
System.out.println("Topology 已提交，等待Kafka消息...");
System.out.println("请在Kafka生产者中输入英文句子进行测试");
// 本地模式运行120秒
Utils.sleep(120000);
// 关闭
cluster.killTopology(TOPOLOGY_NAME);
cluster.shutdown();
System.out.println("Topology已关闭");
}
}
}
注意：上述代码中
KafkaSpoutConfig.RetryService和ConsumerConfig需要正确导入：
// 需要的导入语句
import org.apache.kafka.clients.consumer.ConsumerConfig; // Kafka消费者配置类
8.4.4 使用 Builder 模式的简化版 KafkaSpoutConfig
/**
* KafkaSpoutConfig 配置详解
*
* 这是创建KafkaSpoutConfig的更详细方式
* 包含了更多的配置选项说明
*/
public class KafkaSpoutConfigBuilder {
/**
* 创建并返回KafkaSpoutConfig对象
*
* @return 配置好的KafkaSpoutConfig
*/
public static KafkaSpoutConfig<String, String> createKafkaSpoutConfig() {
// 创建KafkaSpoutConfig
KafkaSpoutConfig<String, String> config = KafkaSpoutConfig
// 基本连接配置
// 参数1：Kafka Broker地址（多个用逗号分隔）
// 参数2：要消费的Topic名称
.builder("nimbus1:9092", "storm-input")
// ========== 反序列化配置 ==========
// 设置Key的反序列化器
// StringDeserializer将byte[]转为String
.setProp(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringDeserializer")
// 设置Value的反序列化器
.setProp(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringDeserializer")
// ========== 消费者组配置 ==========
// 消费者组ID
// 相同group.id的消费者属于同一组
// 同一组内的消费者共同消费Topic的所有分区
.setGroupId("my-storm-consumer-group")
// ========== Offset管理配置 ==========
// 当消费者组首次消费（没有已提交的offset）时的策略
// "earliest"：从Topic最早的消息开始消费
// "latest"：从最新消息开始消费
// "none"：抛出异常
.setProp(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
// 是否自动提交offset
// false：由Storm管理offset提交（推荐）
.setProp(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false)
// ========== 性能配置 ==========
// 每次poll()的最大记录数
.setProp(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100)
// ========== 构建 ==========
.build();
return config;
}
}
8.5 程序运行
# ===== 准备工作 =====
# 1. 启动ZooKeeper集群
# 每个ZK节点执行：
zkServer.sh start
# 2. 启动Kafka集群
# 每个Kafka节点执行：
kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties
# 3. 启动Storm集群
# Nimbus节点：
storm nimbus &
storm ui &
# 每个Supervisor节点：
storm supervisor &
# 4. 创建Kafka Topic（如果还没创建）
kafka-topics.sh --create \
--bootstrap-server nimbus1:9092 \
--topic storm-input \
--partitions 3 \
--replication-factor 2
# ===== 运行程序 =====
# 5. 打包项目
mvn clean package
# 6. 提交Topology到Storm集群
storm jar target/storm-kafka-integration-1.0-SNAPSHOT.jar \
com.example.storm.kafka.KafkaWordCountTopology \
kafka-word-count-topology
# 7. 启动Kafka生产者，输入测试数据
kafka-console-producer.sh \
--bootstrap-server nimbus1:9092 \
--topic storm-input
# 在生产者控制台输入：
# > hello storm hello world
# > apache storm is great
# > hello kafka integration with storm
# > big data real time processing
# > hello storm kafka hello
# 8. 查看Storm Worker日志，观察输出结果
tail -f $STORM_HOME/logs/workers-artifacts/kafka-word-count-topology/*/worker.log
# ===== 预期输出 =====
# 【SplitBolt】接收到Kafka消息: hello storm hello world
# 【Kafka词频统计】单词: hello, 当前计数: 1, 总共统计单词数: 1
# 【Kafka词频统计】单词: storm, 当前计数: 1, 总共统计单词数: 2
# 【Kafka词频统计】单词: hello, 当前计数: 2, 总共统计单词数: 2
# 【Kafka词频统计】单词: world, 当前计数: 1, 总共统计单词数: 3
# 【SplitBolt】接收到Kafka消息: apache storm is great
# 【Kafka词频统计】单词: apache, 当前计数: 1, 总共统计单词数: 4
# 【Kafka词频统计】单词: storm, 当前计数: 2, 总共统计单词数: 4
# ...
# 9. 停止Topology
storm kill kafka-word-count-topology
九、Storm 可靠性机制详解
9.1 Tuple 树（Tuple Tree）
Storm可靠性机制的核心是Tuple Tree（元组树）：
Spout发射一个Tuple
│
├── Bolt A 处理后发射新Tuple
│ ├── Bolt B 处理
│ └── Bolt C 处理
└── Bolt D 处理后发射新Tuple
└── Bolt E 处理
这棵树中所有的Tuple都是从Spout最初发射的Tuple衍生出来的
如果整棵树中的所有Tuple都被成功ack → Spout调用ack()
如果树中任何一个Tuple失败或超时 → Spout调用fail()
9.2 可靠性 API
/**
* 演示Storm可靠性机制的Bolt实现
*/
public class ReliableBolt extends BaseRichBolt {
private OutputCollector collector;
@Override
public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
this.collector = collector;
}
@Override
public void execute(Tuple input) {
try {
// ---- 处理业务逻辑 ----
String data = input.getStringByField("word");
// ---- 锚定（Anchoring）发射 ----
// 锚定发射：将新发射的Tuple与输入Tuple关联
// 这样新Tuple也成为Tuple Tree的一部分
// 如果新Tuple没有被下游ack，输入Tuple也会被认为处理失败
this.collector.emit(input, new Values(data.toUpperCase()));
// ---- 非锚定发射 ----
// 非锚定发射：新Tuple与输入Tuple无关
// 新Tuple的处理失败不会影响输入Tuple
// this.collector.emit(new Values(data.toUpperCase()));
// ---- 多Tuple锚定发射 ----
// 一次锚定多个输出Tuple
// this.collector.emit(input, new Values("a"));
// this.collector.emit(input, new Values("b"));
// ---- 确认输入Tuple ----
// 告诉Storm框架该Tuple处理成功
// 所有下游Tuple都ack后，Spout才会收到ack
this.collector.ack(input);
} catch (Exception e) {
// ---- 标记失败 ----
// 告诉Storm框架该Tuple处理失败
// 会导致Spout重新发送原始Tuple
this.collector.fail(input);
}
}
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("upperWord"));
}
}
9.3 Spout 可靠性
/**
* 支持可靠性机制的Spout实现
*/
public class ReliableSpout extends BaseRichSpout {
private SpoutOutputCollector collector;
// 存储已发射但未确认的Tuple
// Key：消息ID，Value：消息内容
// 用于失败重发
private Map<Long, String> pendingMessages;
// 消息ID生成器
private long msgId = 0;
@Override
public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
this.collector = collector;
this.pendingMessages = new HashMap<>();
}
@Override
public void nextTuple() {
String message = "hello storm";
long id = msgId++;
// 存储待确认的消息
pendingMessages.put(id, message);
// 发射Tuple时指定消息ID（第一个参数）
// 这个ID会在ack()或fail()回调中传回
this.collector.emit(new Values(message), id);
}
/**
* ack回调方法
* 当Tuple Tree中所有Tuple都被成功处理后调用
* @param msgId 消息ID
*/
@Override
public void ack(Object msgId) {
// 从pending中移除已确认的消息
pendingMessages.remove(msgId);
System.out.println("消息处理成功，ID: " + msgId);
}
/**
* fail回调方法
* 当Tuple Tree中任何一个Tuple处理失败或超时时调用
* @param msgId 消息ID
*/
@Override
public void fail(Object msgId) {
// 获取失败的消息
String message = pendingMessages.get(msgId);
System.err.println("消息处理失败，ID: " + msgId + "，将重新发送: " + message);
// 重新发送失败的消息
this.collector.emit(new Values(message), msgId);
}
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("message"));
}
}
十、Storm 配置参数详解
10.1 Topology 级别配置
/**
* Topology级别常用配置参数
* 在提交Topology时通过Config对象设置
*/
public class StormConfigExample {
public static Config createConfig() {
Config config = new Config();
// ========== 并行度相关 ==========
// Worker进程数量
// 每个Worker是一个独立的JVM进程
// 总Worker数 = 设置的Worker数
config.setNumWorkers(4);
// ========== 可靠性相关 ==========
// 消息超时时间（秒）
// Tuple从Spout发射开始计时
// 如果超过该时间Tuple树没有完全ack，Spout的fail()被调用
// 默认30秒
config.setMessageTimeoutSecs(60);
// 是否跳过已过期的Tuple
// true：如果Tuple超时，Spout的ack/fail不会被调用
// 默认false
config.setSkipMissingKryoRegistrations(true);
// ========== 调试相关 ==========
// 调试模式
// true：记录更详细的日志
config.setDebug(false);
// Topology运行的最大时间（毫秒）
// 0表示永不过期
config.setMaxTaskParallelism(10);
// ========== 序列化相关 ==========
// 注册自定义序列化器
// 如果Tuple中包含自定义对象，需要注册对应的序列化器
// config.registerSerialization(MyCustomClass.class);
return config;
}
}
10.2 Component 级别并行度
/**
* Component级别并行度设置
* 在TopologyBuilder中设置
*/
public class ParallelismExample {
public static void main(String[] args) {
TopologyBuilder builder = new TopologyBuilder();
// ========== 并行度参数说明 ==========
// 参数1：组件ID
// 参数2：组件实例
// 参数3：并行度（executor数量，即线程数）
// 设置Spout并行度为2（2个线程）
builder.setSpout("spout", new MySpout(), 2);
// 设置Bolt并行度为6（6个线程）
// 并通过setNumTasks设置Task数量为12
// 意味着6个executor平均每个executor运行2个task
builder.setBolt("bolt", new MyBolt(), 6)
.setNumTasks(12) // 设置Task总数为12
.shuffleGrouping("spout");
// ========== 并行度关系 ==========
// executor（线程）数 <= task（任务）数
//
// 示例：
// executor数=6, task数=12
// 每个executor运行 12/6 = 2 个task
//
// executor数=6, task数=6（默认）
// 每个executor运行 1 个task
}
}
十一、知识要点总结
11.1 核心概念总结
1. Storm 是分布式实时流处理框架
2. Topology = Spout + Bolt 组成的DAG
3. Spout 是数据源头，Bolt 是处理单元
4. Tuple 是数据传输的基本单元
5. Stream 是无限的Tuple序列
6. Stream Grouping 控制Tuple如何路由到Bolt Task
7. 可靠性机制通过Tuple Tree和ack/fail实现
11.2 流分组选择指南
场景 → 推荐分组方式
───────────────────────────────────────────
无需数据关联，均匀分配 → shuffleGrouping
相同Key数据需要同一Task处理 → fieldsGrouping
需要广播到所有Task → allGrouping
需要集中到一个Task处理 → globalGrouping
性能优先，优先本地Task → localOrShuffleGrouping
需要精确控制路由 → directGrouping
自定义路由逻辑 → customGrouping
11.3 Storm vs 其他流处理框架
┌──────────┬────────────┬───────────┬───────────┐
│ 特性 │ Storm │ Spark │ Flink │
│ │ │ Streaming │ │
├──────────┼────────────┼───────────┼───────────┤
│ 处理模型 │ 逐条处理 │ 微批处理 │ 逐条处理 │
│ 延迟 │ 毫秒级 │ 秒级 │ 毫秒级 │
│ 吞吐量 │ 中等 │ 高 │ 高 │
│ 状态管理 │ 无内置 │ 有 │ 有 │
│ 窗口支持 │ 有限 │ 有 │ 丰富 │
│ 精确一次 │ at-least-once│ exactly-once│ exactly-once│
│ 编程复杂度│ 简单 │ 中等 │ 中等 │
│ 生态系统 │ 成熟 │ 非常成熟 │ 快速成长 │
└──────────┴────────────┴───────────┴───────────┘
更多推荐
7
0- 0
已为社区贡献45条内容
所有评论(0)