保姆级教程：用Docker快速搭建双EMQX集群，实现跨集群数据同步

零代码实战：基于Docker的EMQX双集群数据同步方案

在物联网和实时消息处理领域，跨集群数据同步是确保系统高可用性和数据一致性的关键需求。想象一下，你正在开发一个智能家居控制系统，需要将用户指令同时分发到位于不同地理区域的服务器集群，或者你正在构建一个金融交易平台，要求所有订单数据必须实时同步到备份集群以防单点故障。传统实现这类需求往往需要编写复杂的中间件代码，而今天我们将展示如何利用EMQX强大的数据集成功能，完全通过配置实现跨集群数据同步。

1. 环境准备：单机模拟双集群架构

1.1 Docker化EMQX集群部署

在一台开发机上模拟生产环境的多集群架构，我们需要解决端口冲突和网络隔离问题。以下是精心设计的部署方案：

# 本地集群emqx-01（使用标准端口） docker run -d --name emqx-01 \ -p 1883:1883 -p 8883:8883 \ -p 8083:8083 -p 8084:8084 \ -p 18083:18083 \ -e EMQX_NODE_NAME=emqx@node1.local \ emqx/emqx:5.4.1 # 远程集群emqx-02（端口偏移10000避免冲突） docker run -d --name emqx-02 \ -p 11883:1883 -p 18883:8883 \ -p 18083:8083 -p 18084:8084 \ -p 28083:18083 \ -e EMQX_NODE_NAME=emqx@node2.remote \ emqx/emqx:5.4.1

关键提示：通过环境变量EMQX_NODE_NAME为每个节点设置唯一标识符，这是后续集群扩展的基础。端口映射采用"基础端口+10000"的偏移策略，既保持规律又避免冲突。

1.2 网络连通性验证

部署完成后，使用以下命令验证两个集群的运行状态：

# 检查容器状态 docker ps --filter "name=emqx" --format "table {{.Names}}\t{{.Status}}" # 测试MQTT服务连通性 mosquitto_pub -h localhost -p 1883 -t test -m "ping" # 本地集群 mosquitto_pub -h localhost -p 11883 -t test -m "ping" # 远程集群

如果一切正常，你应该能在两个集群的Web控制台（分别访问http://localhost:18083和http://localhost:28083）看到连接成功的客户端。

2. 数据集成核心组件解析

EMQX的数据集成功能由两大核心组件构成，理解它们的协作机制是配置成功的关键。

2.1 连接器(Connector)：数据通道的桥梁

连接器负责建立与外部系统的通信通道，支持包括：

• MQTT桥接（集群间同步）
• 数据库写入（MySQL、PostgreSQL等）
• 消息队列（Kafka、RabbitMQ）
• 云服务（AWS IoT、阿里云IoT）

在跨集群同步场景中，我们选择MQTT类型的连接器，其核心参数包括：

2.2 规则(Rule)：数据流转的逻辑引擎

规则定义了数据处理的全流程：

1. 数据输入：通过SQL语句筛选MQTT主题或消息内容
2. 数据处理：使用内置函数进行消息转换、过滤
3. 数据输出：将结果发送到指定的连接器

一个典型的规则SQL示例：

SELECT payload.temperature as temp, payload.humidity as hum, clientid FROM "t/#" WHERE temp > 30 AND hum < 60

3. 实战配置：从零构建数据同步流

3.1 创建MQTT桥接连接器

通过本地集群(emqx-01)的Web控制台（端口18083）进行操作：

1. 导航到"数据集成" → "连接器"
2. 点击"创建"，选择"MQTT桥接"类型
3. 填写远程集群连接信息：
   • 名称：remote_emqx_bridge
   • 服务地址：host.docker.internal:11883
   • 用户名/密码：如有认证需填写
   • 高级设置：
     • 心跳间隔：30秒
     • 自动重连：开启
     • 清洁会话：关闭

特别注意：在Docker环境中使用host.docker.internal访问宿主机网络，这是解决容器间通信的便捷方式。

3.2 配置数据同步规则

继续在emqx-01控制台操作：

1. 转到"规则引擎" → "规则"
2. 点击"创建"，输入以下SQL语句：

   SELECT *, now() as sync_time FROM "t/#"

3. 添加动作：
   • 动作类型：消息转发
   • 使用连接器：选择刚创建的remote_emqx_bridge
   • 目标主题：保持与原始主题相同（可通过${topic}变量）

高级配置技巧：

• 使用payload_encoding字段处理二进制数据
• 通过qos参数控制消息质量等级
• 设置retain标志位管理保留消息

3.3 验证数据流

打开两个终端窗口，分别运行：

# 终端1：订阅远程集群的测试主题 mosquitto_sub -h localhost -p 11883 -t "t/test" -v # 终端2：向本地集群发布消息 mosquitto_pub -h localhost -p 1883 -t "t/test" -m '{"temp":25,"hum":45}'

如果配置正确，你将在终端1立即看到来自远程集群的消息。

4. 生产环境进阶配置

4.1 性能优化参数

在docker run命令中添加以下环境变量优化EMQX性能：

-e EMQX_LISTENERS__TCP__DEFAULT__MAX_CONNECTIONS=100000 \ -e EMQX_ZONE__EXTERNAL__MQTT_MAX_PACKET_SIZE=10MB \ -e EMQX_ZONE__EXTERNAL__RATE_LIMIT__CONN_MESSAGES_IN=1000,10s \

关键性能指标监控建议：

4.2 高可用架构设计

对于生产环境，建议采用以下架构：

1. 集群部署：每个EMQX集群至少3个节点
2. 负载均衡：使用Nginx或HAProxy分发MQTT连接
3. 持久化存储：配置Redis或MySQL作为规则和连接器的后端存储

示例的Docker Compose片段：

services: emqx1: image: emqx/emqx:5.4.1 environment: - EMQX_CLUSTER__DISCOVERY_STRATEGY=static - EMQX_CLUSTER__STATIC__SEEDS=emqx1@node1,emqx2@node2,emqx3@node3 networks: - emqx_net emqx2: image: emqx/emqx:5.4.1 environment: - EMQX_CLUSTER__DISCOVERY_STRATEGY=static - EMQX_CLUSTER__STATIC__SEEDS=emqx1@node1,emqx2@node2,emqx3@node3 networks: - emqx_net networks: emqx_net: driver: bridge

4.3 安全加固措施

1. TLS加密传输：

   -p 8883:8883 -e EMQX_LISTENERS__SSL__DEFAULT__ENABLED=true

2. ACL访问控制：

   -e EMQX_AUTHORIZATION__SOURCES="[\"file\"]" \ -e EMQX_AUTHORIZATION__CACHE__ENABLE=true

3. 审计日志：

   -e EMQX_LOG__LEVEL=warning \ -e EMQX_LOG__FILE="/var/log/emqx/audit.log"

5. 故障排查与调试技巧

当同步失败时，按照以下步骤排查：

1. 检查连接器状态：

   docker exec emqx-01 emqx_ctl bridges list

2. 查看规则执行日志：

   docker logs emqx-01 --tail 100 | grep "rule_engine"

3. 网络连通性测试：

   docker exec -it emqx-01 ping emqx-02

常见问题解决方案：

• 端口冲突：使用netstat -tulnp | grep 1883确认端口占用情况
• 认证失败：检查/etc/emqx/acl.conf文件权限和内容
• 消息堆积：调整EMQX_ZONE__EXTERNAL__MAX_TOPIC_ALIAS参数

在最近的一个智慧园区项目中，我们发现当同步大量设备状态时，会出现消息延迟。通过调整EMQX_BRIDGES__MQTT__AWS__MAX_SEND_QUEUE_LEN=5000参数并增加EMQX_BRIDGES__MQTT__AWS__RESOURCE_OPTS__WORKER_POOL_SIZE=8，最终将同步延迟控制在100ms以内。