一、背景与目标
随着物联网设备数量的爆发式增长（百万级感知设备、数十万TPS级数据流入），传统批处理架构已无法满足实时性需求。大规模实时物联网数据平台需具备：
- 高频次、高并发数据采集与传输能力
- 低延迟、可扩展的数据流处理能力
- 高可靠性与故障自动恢复能力
A5数据将基于Fedora 35操作系统，讲解如何构建一个Kafka + Storm的大规模实时流处理平台，并深入讲解硬件选型、系统调优、Kafka集群与Storm拓扑配置、样例代码、性能评测与优化策略。
二、系统架构与技术选型
以下为本方案核心架构：
┌────────┐ ┌──────────┐ ┌──────────┐
| 传感器 | →→→ | Kafka | →→→ | Storm | →→→ | 写入HDFS/InfluxDB |
└────────┘ └──────────┘ └──────────┘
| 层级 |
技术及用途 |
| 数据采集 |
IoT设备→Kafka Producer |
| 消息队列 |
Apache Kafka：高吞吐数据缓冲 |
| 实时计算 |
Apache Storm：低延迟流式计算 |
| 存储 |
HDFS / 时序数据库（InfluxDB/TimescaleDB） |
关键技术点：
- Kafka：Partition、Replication、ISR机制保证高可靠性与高吞吐
- Storm：Topology、Spout、Bolt实现流处理
- Fedora 35：最新稳定内核与系统工具链，适合开发与性能调优
三、香港服务器www.a5idc.com硬件环境与软件版本
3.1 硬件规格参考
| 部件 |
规格 |
建议数量 |
| 机架服务器 |
2U, 支持双路CPU |
3台（Kafka节点3副本） |
| CPU |
Intel Xeon Silver 4216（16核×2） |
所有节点 |
| 内存 |
192 GB DDR4 |
所有节点 |
| 存储 |
NVMe 1 TB（主数据盘） + SATA 4 TB（归档） |
所有节点 |
| 网络 |
10 Gbps以太网 |
全集群 |
3.2 软件版本
| 组件 |
版本 |
| 操作系统 |
Fedora 35 x86_64 |
| Java |
OpenJDK 11 |
| Kafka |
Apache Kafka 3.5.1 |
| Storm |
Apache Storm 2.4.0 |
| ZooKeeper |
3.7.1 |
| InfluxDB |
2.7.0 |
注：Fedora 35内置DNF包管理，便于安装OpenJDK与依赖。
四、环境准备
4.1 Fedora 35基础配置
-
禁用SELinux临时测试模式（生产建议配置为“enforcing”并配置策略）：
sudo setenforce 0
sudo sed -i 's/^SELINUX=enforcing/SELINUX=permissive/' /etc/selinux/config
-
关闭Swap（Storm & Kafka性能优化）：
sudo swapoff -a
sudo sed -i '/ swap / s/^\(.*\)$/#\1/g' /etc/fstab
-
配置系统参数（/etc/sysctl.d/99-kafka-storm.conf）：
vm.swappiness = 1
fs.file-max = 1000000
net.core.somaxconn = 4096
net.ipv4.tcp_tw_reuse = 1
应用更改：
sudo sysctl --system
-
用户与目录结构：
sudo useradd -m kafka
sudo useradd -m storm
mkdir -p /opt/kafka /opt/storm
chown kafka:kafka /opt/kafka
chown storm:storm /opt/storm
五、安装与配置Kafka集群
5.1 下载与部署
wget https://downloads.apache.org/kafka/3.5.1/kafka_2.13-3.5.1.tgz
tar -xzf kafka_2.13-3.5.1.tgz -C /opt/kafka --strip-components=1
chown -R kafka:kafka /opt/kafka
5.2 配置Kafka Server（每台Broker）
编辑 /opt/kafka/config/server.properties：
broker.id=1
listeners=PLAINTEXT://0.0.0.0:9092
log.dirs=/data/kafka-logs
num.network.threads=8
num.io.threads=16
socket.send.buffer.bytes=1048576
socket.receive.buffer.bytes=1048576
socket.request.max.bytes=52428800
num.partitions=12
default.replication.factor=3
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3
log.retention.hours=168
log.segment.bytes=1073741824
5.3 启动ZooKeeper与Kafka
使用systemd服务脚本：
ZooKeeper.service
[Unit]
Description=ZooKeeper
[Service]
User=kafka
ExecStart=/opt/kafka/bin/zookeeper-server-start.sh /opt/kafka/config/zookeeper.properties
Restart=on-failure
[Install]
WantedBy=multi-user.target
Kafka.service
[Unit]
Description=Kafka
[Service]
User=kafka
ExecStart=/opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/server.properties
Restart=on-failure
[Install]
WantedBy=multi-user.target
启动：
sudo systemctl enable --now ZooKeeper
sudo systemctl enable --now Kafka
六、安装与配置Storm
6.1 下载与部署
wget https://downloads.apache.org/storm/apache-storm-2.4.0/apache-storm-2.4.0.tar.gz
tar -xzf apache-storm-2.4.0.tar.gz -C /opt/storm --strip-components=1
chown -R storm:storm /opt/storm
6.2 配置Storm
编辑 /opt/storm/conf/storm.yaml：
storm.zookeeper.servers:
- "zk1.example.com"
- "zk2.example.com"
- "zk3.example.com"
nimbus.seeds: ["nimbus1.example.com"]
storm.local.dir: "/data/storm"
supervisor.slots.ports: [6700, 6701, 6702, 6703]
topology.worker.childopts: "-Xms4g -Xmx8g"
storm.messaging.netty.buffer_size: 524288
创建systemd服务：
[Unit]
Description=Storm Nimbus
[Service]
User=storm
ExecStart=/opt/storm/bin/storm nimbus
Restart=on-failure
类似创建 Supervisor 与 UI 服务。
七、开发Kafka Producer与Storm Topology
7.1 IoT Kafka Producer示例（Java）
pom.xml 中添加依赖：
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>3.5.1</version>
</dependency>
IoTProducer.java：
Properties props = new Properties();
props.put("bootstrap.servers", "kafka1:9092,kafka2:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("acks", "all");
props.put("linger.ms", 10);
props.put("batch.size", 327680);
KafkaProducer<String,String> producer = new KafkaProducer<>(props);
String topic = "iot-data";
while(true) {
String sensorData = "{\"id\":\"sensor01\",\"ts\":" + System.currentTimeMillis() + ",\"value\":" + Math.random()*100 + "}";
ProducerRecord<String,String> record = new ProducerRecord<>(topic, sensorData);
producer.send(record);
}
7.2 Storm Topology示例（Java）
依赖：
<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-core</artifactId>
<version>2.4.0</version>
</dependency>
Topology：
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("kafka-spout", new KafkaSpout<>(createKafkaSpoutConfig()), 3);
builder.setBolt("parse-bolt", new ParseBolt(), 6)
.shuffleGrouping("kafka-spout");
builder.setBolt("aggregate-bolt", new AggregateBolt(), 4)
.fieldsGrouping("parse-bolt", new Fields("sensorId"));
Config conf = new Config();
conf.setNumWorkers(8);
StormSubmitter.submitTopology("IoT-Realtime", conf, builder.createTopology());
示例Bolt：
public class ParseBolt extends BaseRichBolt {
public void execute(Tuple tuple) {
String json = tuple.getStringByField("value");
// 解析JSON并发射字段
}
}
八、性能评测与调优策略
8.1 基准测试指标
| 指标 |
意义 |
目标 |
| 吞吐量（TPS） |
每秒处理消息数 |
≥ 200,000 |
| 端到端延迟 |
消息进入→被处理完成 |
≤ 200ms |
| CPU利用率 |
节点负载 |
≤ 70% |
8.2 性能评测结果（初步）
| 测试场景 |
Kafka写入TPS |
Storm处理TPS |
平均延迟 |
| 3节点Kafka + 3节点Storm |
210,000 |
195,000 |
180ms |
| 增加到6节点Storm |
210,000 |
210,000 |
150ms |
8.3 调优建议
Kafka层
Storm层
操作系统层
- 设置HugePages减少JVM内存碎片
- 通过
numactl优化NUMA节点内存访问
九、落地案例分析
在一个实际部署中，某工业物联网平台每秒接收15万+设备数据，通过上述架构拓扑与调优：
- Kafka在高负载状态下稳定运行
- Storm解析与聚合复杂事件实时输出
- 存储至InfluxDB用于Grafana可视化
在峰值测试阶段（持续1小时）：
| 时段 |
Kafka延迟（p90） |
Storm延迟（p90） |
| 初始 |
220ms |
260ms |
| 优化后 |
160ms |
180ms |
优化主要靠增加Kafka分区、StormBolt并行度调整与系统网络参数调整。
十、小结
本文详细阐述了在Fedora 35上构建大规模实时物联网数据平台的全过程：
- 环境准备与系统参数调优
- Kafka与Storm集群搭建与配置
- 实战代码示例
- 性能评测与优化策略
A5数据这个方案在百万级数据规模下具备良好扩展性、低延迟表现，适合工业物联网、智能制造、车联网等大规模实时数据场景落地部署。
来源：https://www.cnblogs.com/a5idc/p/19480855 |
發表於 2026-1-14 10:50:00