5分钟上手EMQX+Flink：打造工业级IoT实时流处理管道

你是否正面临IoT设备数据洪流难以实时分析的困境？工厂传感器每秒钟产生的10万条数据还在依赖批处理？本文将带你用EMQX+Apache Flink构建毫秒级响应的流处理管道，零基础也能快速上手。

为什么选择EMQX与Flink组合

在工业物联网（IIoT）场景中，设备数据具有高并发（百万级连接）、低延迟（毫秒级响应）和不规则性三大特性。EMQX作为开源MQTT消息服务器，能稳定支撑1亿级设备连接README.md，而Apache Flink的流处理能力可实现事件时间语义和状态管理，两者结合形成完美的数据处理闭环。

组件	核心优势	适用场景
EMQX	支持MQTT 5.0、WebSocket等多协议接入	设备数据采集层
Flink	Exactly-Once语义、窗口计算	实时数据处理层

数据流转架构设计

下图展示了从工业传感器到业务看板的完整数据链路：

graph LR
    A[PLC传感器] -->|MQTT| B[EMQX Broker]
    B -->|Kafka桥接| C[Apache Kafka]
    C -->|Flink SQL| D[实时计算]
    D --> E{业务存储}
    E -->|MySQL| F[历史数据查询]
    E -->|Redis| G[实时仪表盘]

关键实现模块：

• 协议接入：emqx_gateway_mqttsn/
• 数据转发：emqx_bridge_kafka/
• 规则引擎：emqx_rule_engine/

step-by-step实施指南

1. EMQX配置Kafka桥接

在EMQX Dashboard中创建Kafka桥接，将设备数据实时转发至Kafka集群：

bridges.kafka.my_bridge {
  enable = true
  bootstrap_servers = "kafka:9092"
  topic = "iot_data"
  producer {
    acks = "all"
    compression.type = "lz4"
  }
}

配置文件路径：emqx_bridge_kafka_schema.hocon

2. 创建数据处理规则

通过EMQX规则引擎筛选关键数据字段，转换为Flink友好的JSON格式：

SELECT 
  clientid as device_id,
  payload.temperature as temp,
  payload.humidity as humi,
  timestamp as collect_time
FROM "sensor/data"
WHERE temp > 30

规则引擎文档：emqx_rule_engine/

3. Flink流处理实现

使用Flink SQL消费Kafka数据，计算5分钟滑动窗口内的温度平均值：

CREATE TABLE iot_source (
  device_id STRING,
  temp DOUBLE,
  humi DOUBLE,
  collect_time TIMESTAMP(3)
) WITH (
  'connector' = 'kafka',
  'topic' = 'iot_data',
  'properties.bootstrap.servers' = 'kafka:9092',
  'format' = 'json'
);

CREATE TABLE alert_sink (
  device_id STRING,
  avg_temp DOUBLE,
  window_start TIMESTAMP(3),
  window_end TIMESTAMP(3)
) WITH (
  'connector' = 'jdbc',
  'url' = 'jdbc:mysql://db:3306/iot_db',
  'table-name' = 'high_temp_alerts'
);

INSERT INTO alert_sink
SELECT 
  device_id,
  AVG(temp) as avg_temp,
  TUMBLE_START(collect_time, INTERVAL '5' MINUTE) as window_start,
  TUMBLE_END(collect_time, INTERVAL '5' MINUTE) as window_end
FROM iot_source
GROUP BY TUMBLE(collect_time, INTERVAL '5' MINUTE), device_id
HAVING AVG(temp) > 35;

性能优化实践

1. 连接优化：启用EMQX的连接复用功能，配置文件：emqx.conf
2. 批处理调优：设置Kafka生产者批量大小为16KB，emqx_bridge_kafka/
3. 状态管理：Flink使用RocksDB作为状态后端，设置合理的checkpoint间隔

常见问题排查

问题现象	可能原因	解决方案
数据延迟 > 1s	Kafka分区数不足	增加分区至32个
Flink任务重启	状态后端配置错误	检查flink-conf.yaml
EMQX连接抖动	网络不稳定	启用emqx_cluster_link/

总结与后续展望

通过本文方案，你已成功构建从设备端到业务端的实时数据处理链路。建议进一步探索：

• 边缘计算场景：emqx_edge/
• AI异常检测：emqx_ai_completion/
• 可视化监控：emqx_dashboard/

收藏本文，关注项目README.md获取更多工业物联网最佳实践！下一期我们将详解如何基于此架构实现预测性维护系统。