EMQ X 规则引擎中如何存储消息到MongoDB数据库

发布时间：2021-09-29 11:28:24 作者：柒染
来源：亿速云阅读：219

# EMQ X 规则引擎中如何存储消息到MongoDB数据库

## 1. 引言

EMQ X（现更名为EMQX）作为一款开源的分布式MQTT消息服务器，凭借其高并发、低延迟的特性，已成为物联网领域的核心基础设施。在实际应用中，将设备上报的消息持久化存储是业务系统的基本需求。MongoDB以其灵活的文档模型、水平扩展能力和丰富的查询功能，成为物联网消息存储的热门选择。

本文将详细介绍如何通过EMQX规则引擎，将MQTT消息高效、可靠地存储到MongoDB数据库中，涵盖从原理到实践的完整流程。

## 2. EMQX规则引擎概述

### 2.1 规则引擎核心概念

EMQX规则引擎是一个基于SQL的轻量级数据处理框架，允许用户通过配置而非编码的方式实现消息处理。其核心组件包括：

- **规则(Rule)**：定义数据处理的逻辑条件（WHERE子句）和字段提取方式（SELECT部分）
- **动作(Action)**：指定满足条件后的操作（如存储到数据库、转发到其他主题等）

### 2.2 规则引擎工作流程

1. 消息到达EMQX Broker
2. 规则引擎匹配消息主题或内容
3. 对匹配的消息执行字段提取、过滤和转换
4. 触发配置的动作（如写入MongoDB）

```sql
示例规则SQL：
SELECT 
    clientid,
    payload.temperature as temp,
    payload.humidity as hum,
    timestamp as ts
FROM 
    "sensor/data"
WHERE 
    payload.temperature > 30

3. MongoDB集成准备

3.1 MongoDB部署建议

对于物联网场景推荐以下配置： - 副本集（Replica Set）部署确保高可用 - 分片集群（Sharded Cluster）应对海量数据 - WiredTiger存储引擎优化写入性能 - 合理设置TTL索引自动过期旧数据

3.2 数据库用户权限配置

创建专用用户并授予最小必要权限：

use iot_data
db.createUser({
  user: "emqx_user",
  pwd: "securepassword",
  roles: [
    { role: "readWrite", db: "iot_data" },
    { role: "clusterMonitor", db: "admin" }
  ]
})

3.3 集合设计建议

考虑以下优化策略： - 按设备类型分集合（如sensor_data、gateway_logs） - 嵌入式文档结构存储复杂数据 - 创建合适的索引：

  db.sensor_data.createIndex({ "clientid": 1, "timestamp": -1 })
  db.sensor_data.createIndex({ "timestamp": 1 }, { expireAfterSeconds: 2592000 })

4. 配置规则存储到MongoDB

4.1 安装MongoDB资源

通过EMQX Dashboard配置： 1. 进入”规则引擎” → “资源” 2. 选择”MongoDB Replica Set”或”MongoDB Single”类型 3. 填写连接参数：

   Server: mongodb://emqx_user:securepassword@10.0.0.1:27017,10.0.0.2:27017/iot_data?replicaSet=rs0
   Pool Size: 8
   SSL: 禁用（生产环境建议启用）

4.2 创建规则

场景示例：存储温度传感器数据

SELECT
    clientid,
    payload.temp as temperature,
    payload.hum as humidity,
    now_timestamp() as recorded_at,
    metadata.topic as topic
FROM 
    "sensor/+/data"
WHERE
    is_defined(payload.temp)

4.3 配置MongoDB动作

动作类型选择：”Data Persistence - MongoDB”
关联已创建的MongoDB资源

配置操作参数：

Collection: sensor_readings
Operation: insert_one

Payload Template:


{
"device_id": ${clientid},
"temperature": ${temperature},
"humidity": ${humidity},
"topic": "${topic}",
"recorded_at": ${recorded_at}
}

4.4 高级配置选项

批量写入：启用enable_batch提高吞吐量


{
"enable_batch": true,
"batch_size": 100,
"batch_time": "10s"
}

写入确认：设置write_concern为majority确保数据安全
重试策略：配置失败重试次数和间隔

5. 性能优化实践

5.1 写入性能调优

参数	推荐值	说明
pool_size	CPU核心数×2	连接池大小
batch_size	100-500	批量写入文档数
batch_time	5-10s	批量写入最大等待时间
write_concern	w1	对性能要求高时可降低要求

5.2 消息处理优化

主题通配符优化：
- 避免过度宽泛的通配符（如#）
- 使用层级匹配（如sensor/+/data）
SELECT子句优化：
- 只提取必要字段
- 在SQL中进行简单计算（如payload.temp/10）
条件过滤前置： “`sql – 不推荐（过滤后置） SELECT * FROM “sensor/data” WHERE upper(clientid) LIKE ‘DEV%’

– 推荐（过滤前置） SELECT clientid, payload FROM “$events/client_connected” WHERE clientid LIKE ‘dev%’


## 6. 监控与故障排查

### 6.1 关键监控指标

通过EMQX Prometheus接口监控：
- `emqx_rule_exec_pass_count`：规则匹配成功次数
- `emqx_rule_exec_fail_count`：规则执行失败次数
- `emqx_mongo_queries_failure`：MongoDB查询失败数

MongoDB Atlas或MMS监控：
- 写入延迟（Write Latency）
- Oplog滞后时间
- 连接数使用情况

### 6.2 常见问题解决

**问题1**：写入速度突然下降

- 检查MongoDB磁盘IOPS
- 查看是否触发了流量控制（`emqx_mongo_queries_rate_limited`）
- 确认是否有锁等待（`db.currentOp()`）

**问题2**：连接频繁断开

- 调整心跳间隔（`heartbeat_frequency_ms`）
- 检查网络稳定性（TCP重传率）
- 增加连接池大小

**问题3**：数据格式错误

- 使用`try-catch`处理异常数据：
  ```sql
  SELECT 
      try(payload.temp) as temperature
  FROM "sensor/data"

7. 进阶应用场景

7.1 时序数据处理

结合MongoDB时间序列集合（5.0+版本）：

db.createCollection("sensor_ts", {
  timeseries: {
    timeField: "timestamp",
    metaField: "metadata",
    granularity: "minutes"
  }
})

7.2 数据预处理

在规则引擎中完成计算：

SELECT
    clientid,
    (payload.temp - 32) * 5/9 as temp_c,
    avg(payload.voltage) OVER (PARTITION BY clientid ORDER BY timestamp ROWS 5 PRECEDING) as moving_avg
FROM 
    "sensor/+/data"

7.3 条件存储策略

使用规则引擎分流：

-- 高温数据存MongoDB
SELECT * FROM "sensor/data" WHERE payload.temp > 38

-- 正常数据存InfluxDB
SELECT * FROM "sensor/data" WHERE payload.temp <= 38

8. 安全注意事项

连接安全：
- 始终使用TLS加密MongoDB连接
- 定期轮换数据库凭据

数据安全：

// 启用加密
db.adminCommand({ 
 setClusterParameter: {
   encryptionOptions: {
     keyFile: "/path/to/keyfile"
   }
 }
})

访问控制：
- 限制EMQX服务器的源IP访问
- 使用VPC Peering或私有端点

9. 总结

通过本文的详细指导，您应该已经掌握： - EMQX规则引擎与MongoDB集成的完整流程 - 高性能配置的最佳实践 - 生产环境中的运维技巧

这种组合方案能够支持百万级设备的消息持久化需求，同时保持毫秒级的写入延迟。建议在实际部署前进行压力测试，根据具体硬件配置调整参数。

附录

A. EMQX版本兼容性

EMQX版本	MongoDB驱动版本	关键特性
4.3.x	3.12.x	基础CRUD支持
5.0+	4.4.x	批量写入、聚合管道支持

B. 参考配置模板

# emqx_rule.conf
rule.mongo.rule_1 = {
  sql = "SELECT * FROM \"sensor/data\""
  actions = [{
    name = "mongo_action",
    params = {
      collection = "sensor_data",
      mongo_type = "insert_one"
    }
  }]
}

C. 性能测试数据

测试环境：AWS c5.2xlarge × 3节点

消息大小	吞吐量 (msg/s)	MongoDB CPU使用率
1KB	12,000	65%
5KB	8,200	78%

”`