EMQX 5.0 企业版三节点集群实战：从Docker Compose部署到千万级连接压测

1. 企业级MQTT集群架构设计

在物联网爆发式增长的今天，单节点MQTT代理已无法满足大规模设备连接需求。EMQX 5.0企业版作为专为物联网设计的分布式MQTT消息服务器，其集群架构具有以下核心优势：

• 无单点故障 ：采用无主(leaderless)架构，所有节点对等
• 线性扩展 ：支持动态增加节点提升处理能力
• 数据分区 ：通过分布式哈希表(DHT)实现主题空间分片
• 会话持久化 ：集群共享会话状态，确保设备重连不丢失

集群节点间通信采用 Mria 分布式数据库(基于Raft改进)进行元数据同步，消息转发则通过 直接路由 机制避免中间跳转。这种混合架构在保证一致性的同时，将消息延迟控制在毫秒级。

2. Docker Compose集群部署

2.1 环境准备

部署三节点集群需要准备：

• 至少4GB内存的Linux服务器
• Docker 20.10+和Docker Compose 2.0+
• 开放端口：1883(MQTT)、4369(EPMD)、5370(集群RPC)

# 验证Docker环境
docker --version && docker-compose version

# 创建专用网络
docker network create emqx-net

2.2 编写docker-compose.yml

version: '3'

services:
  emqx1:
    image: emqx/emqx-enterprise:5.0.0
    container_name: emqx-node1
    environment:
      - EMQX_NODE_NAME=emqx@node1
      - EMQX_CLUSTER__DISCOVERY_STRATEGY=static
      - EMQX_CLUSTER__STATIC__SEEDS=emqx@node1,emqx@node2,emqx@node3
    networks:
      - emqx-net
    ports:
      - "1883:1883"
      - "8081:8081"
      - "8083:8083"
      - "8084:8084"
      - "8883:8883"
      - "18083:18083"
    volumes:
      - ./node1_data:/opt/emqx/data

  emqx2:
    image: emqx/emqx-enterprise:5.0.0
    container_name: emqx-node2
    environment:
      - EMQX_NODE_NAME=emqx@node2
      - EMQX_CLUSTER__DISCOVERY_STRATEGY=static
      - EMQX_CLUSTER__STATIC__SEEDS=emqx@node1,emqx@node2,emqx@node3
    networks:
      - emqx-net
    volumes:
      - ./node2_data:/opt/emqx/data

  emqx3:
    image: emqx/emqx-enterprise:5.0.0
    container_name: emqx-node3
    environment:
      - EMQX_NODE_NAME=emqx@node3
      - EMQX_CLUSTER__DISCOVERY_STRATEGY=static
      - EMQX_CLUSTER__STATIC__SEEDS=emqx@node1,emqx@node2,emqx@node3
    networks:
      - emqx-net
    volumes:
      - ./node3_data:/opt/emqx/data

networks:
  emqx-net:
    external: true

关键配置说明：

• EMQX_NODE_NAME ：节点唯一标识，格式为 emqx@hostname
• CLUSTER__DISCOVERY_STRATEGY ：集群发现机制，生产环境推荐DNS或K8S
• CLUSTER__STATIC__SEEDS ：静态种子节点列表

2.3 启动与验证集群

# 启动集群
docker-compose up -d

# 查看节点状态
docker exec -it emqx-node1 emqx_ctl cluster status

# 预期输出
Cluster status: #{running_nodes => ['emqx@node1','emqx@node2','emqx@node3']}

通过Dashboard(http://localhost:18083)可直观查看集群拓扑和节点负载情况。建议配置 负载均衡器 将MQTT流量均匀分发到各节点。

3. 集群调优与生产配置

3.1 关键性能参数

编辑各节点的 /opt/emqx/etc/emqx.conf ：

# 连接层优化
listeners.tcp.default {
  max_connections = 1000000
  acceptors = 16
}

# 会话管理
session {
  max_awaiting_rel = 1000
  upgrade_qos = true
}

# 消息吞吐
zone {
  external {
    mqueue_priorities = "none"
    mqueue_default_priority = highest
  }
}

3.2 持久化与高可用

配置PostgreSQL作为共享订阅数据存储：

-- 创建数据库
CREATE DATABASE emqx_cluster;
CREATE USER emqx WITH PASSWORD 'securepassword';

-- 配置EMQX
bridges.postgresql.1 {
  server = "postgres:5432"
  database = "emqx_cluster"
  pool_size = 8
}

3.3 安全加固

# 生成TLS证书
openssl req -x509 -newkey rsa:4096 -nodes -keyout key.pem -out cert.pem -days 365

# 配置MQTTS
listeners.ssl.default {
  bind = "0.0.0.0:8883"
  certfile = "/opt/emqx/etc/certs/cert.pem"
  keyfile = "/opt/emqx/etc/certs/key.pem"
}

4. 千万级连接压测实战

4.1 压测工具选型

工具	语言	特点	适用场景
emqtt-bench	Erlang	原生支持EMQX，低开销	极限压力测试
JMeter	Java	图形化界面，丰富插件	复杂场景模拟
MQTTx CLI	Go	简单易用	快速验证

推荐使用 emqtt-bench 进行大规模连接测试：

# 启动100万个连接，每秒新增5000个
./emqtt_bench conn -c 1000000 -i 0.2 -h loadbalancer.example.com

4.2 压测场景设计

场景一：纯连接测试

• 目标：验证集群最大连接数
• 参数：100万客户端，QoS 0，1KB心跳间隔
• 监控指标：TCP连接数、内存占用

场景二：消息吞吐测试

• 目标：评估消息处理能力
• 参数：50万发布者，50万订阅者，100字节消息
• 监控指标：消息速率、端到端延迟

场景三：故障转移测试

• 目标：验证集群容错性
• 操作：随机停止1个节点，观察会话恢复

4.3 监控与指标收集

配置Prometheus监控：

# prometheus.yml
scrape_configs:
  - job_name: 'emqx'
    static_configs:
      - targets: ['node1:18083','node2:18083','node3:18083']

关键监控指标：

• emqx_connections_count ：当前连接数
• emqx_messages_received ：消息接收速率
• erlang_vm_memory_total ：内存使用量
• erlang_vm_system_counts_process_count ：Erlang进程数

4.4 压测结果分析

典型性能数据（3节点集群，16核32GB配置）：

场景	连接数	消息速率(msg/s)	平均延迟	CPU负载
纯连接	1,200,000	-	-	65%
发布订阅	500,000	850,000	12ms	85%
故障转移	-	恢复时间<3s	-	-

资源消耗对比：

节点数	内存(GB)	网络吞吐(Mbps)	持久化延迟
1	28	950	单点风险
3	9/节点	2800	<500ms

5. 生产环境最佳实践

5.1 集群扩展策略

• 垂直扩展 ：优先提升单节点资源(CPU/内存)
• 水平扩展 ：当连接数>50万/节点时考虑增加节点
• 分区部署 ：跨可用区部署提高容灾能力

5.2 客户端优化建议

# Python示例：使用持久会话
client = mqtt.Client(client_id="device001", clean_session=False)
client.connect("broker.example.com", keepalive=60)

# 启用自动重连
client.reconnect_delay_set(min_delay=1, max_delay=120)

5.3 常见问题排查

连接不稳定 ：

• 检查Keepalive设置(建议≥60s)
• 验证网络MTU大小(避免分片)
• 监控TCP重传率

消息堆积 ：

• 调整 mqueue_len 参数
• 增加消费者数量
• 检查订阅者QoS等级匹配

内存增长 ：

• 限制 max_inflight 消息数
• 启用 offline_message 清理
• 监控长时间离线的持久会话