支持亿级连接并开源的分布式MQTT消息服务器分享

今天给各位分享一款开源的分布式MQTT消息服务器EMQX，此消息服务器几乎是物联网系统的标配同时也适合做即时通知和推送服务场景，在作者之前参与的项目中主要用于做物联网系统边缘设备信息采集、以及交易所行情数据推送使用，下面是EMQX 相关介绍。

什么是 EMQX

EMQX 是一款开源的大规模分布式 MQTT 消息服务器，功能丰富，专为物联网和实时通信应用而设计。EMQX 5.0 单集群支持 MQTT 并发连接数高达 1 亿条，单服务器的传输与处理吞吐量可达每秒百万级 MQTT 消息，同时保证毫秒级的低时延。

EMQX 支持多种协议，包括 MQTT (3.1、3.1.1 和 5.0)、HTTP、QUIC 和 WebSocket 等，保证各种网络环境和硬件设备的可访问性。EMQX 还提供了全面的 SSL/TLS 功能支持，比如双向认证以及多种身份验证机制，为物联网设备和应用程序提供可靠和高效的通信基础设施。

图片

内置基于 SQL 的规则引擎，EMQX 可以实时提取、过滤、丰富和转换物联网数据。此外，EMQX 采用了无主分布式架构，以确保高可用性和水平扩展性，并提供操作友好的用户体验和出色的可观测性。

EMQX 提供了开源版和商业版两种方式，用户可以基于自己需求进行选择。

官网地址：https://www.emqx.io

github 地址：https://github.com/emqx/emqx

为什么说专为物联网和实时通信设计？

物联网方面

以下是几个理由说明为什么MQTT适合物联网：

实时通信设方面

分布式集群设计原理

MQX 本身支持分布式集群架构，能够在保证高可用性、容错性和可扩展性的同时，处理大量的客户端和消息。通过使用 EMQX 集群，您可以在一个或多个节点发生故障时仍然保持集群运行，从而享受到容错和高可用性的好处。

以下是一个四个节点组成的EMQ集群，每个节点都运行一个 EMQX 实例，并且与集群中的其他节点通信，共享客户端连接、订阅、发布消息等信息。这允许集群在节点之间自动分配负载并在节点出现故障时提供高可用性

图片

在集群架构下，我们可以随着业务的增长向集群添加新节点，从而提供可扩展性。这样可以处理越来越多的客户端和消息，而不必担心单个代理的限制。

消息转发设计

EMQX 分布式集群的基本功能是转发和发布消息到订阅者，如下图所示。

图片

为了实现这一目标，EMQX 在 嵌入式数据库 Mria 中维护着与之相关的几个数据表：

• 订阅表
• 路由表
• 主题树

订阅表：主题-订阅者

EMQX 会维护一个订阅表，用于存储主题->订阅者之间的映射关系，从而确保能将传入消息正确路由到对应的客户端。该数据只存在于订阅者所在的 EMQX 节点上，类似的结构如下：

bash

node1:

    topic1 -> client1, client2
    topic2 -> client3

node2:

    topic1 -> client4

路由表：Topic-Node

路由表记录了 主题->节点 之间的映射，它存储每个节点上客户端订阅的主题列表，并用于将消息路由到对应的节点。该数据会在同一集群中的所有节点复制一份，类似结构如下：

bash

topic1 -> node1, node2
topic2 -> node3
topic3 -> node2, node4

主题树：主题匹配通配符

主题树是一种分层数据结构，它存储有关主题层次结构的信息，并用于消息与订阅客户端的匹配。

主题树会在同一集群中的所有节点复制一份，下面是一个 主题-订阅关系 的例子:

当所有的订阅完成后，EMQX 会维护以下主题树和路由表。

图片

消息分发流程

当一个 MQTT 客户端发布消息时，它所在的节点会查找路由表，并根据消息主题将消息转发到对应的节点（可能是多个节点）。

然后，接收到消息的节点会查找本地订阅表，并将消息发送至对应的订阅者。

例如，当客户端 1 发布一条消息到主题 t/a 时，消息在节点之间的路由和分发如下:

节点 2 上有客户端订阅了 t/# 主题；

节点 3 上有客户端订阅了 t/a 主题；因此节点 1 会将消息同时转发给节点 2 和节点 3。

连接数测试

5.0支持并发连接数高达 1 亿条测试报告：https://www.emqx.com/zh/blog/reaching-100m-mqtt-connections-with-emqx-5-0

快速体验

安装

容器化部署是体验 EMQX 的最快方式，因此本节将以容器化部署为例，在命令行工具中输入如下命令，下载并运行最新版 EMQX。

docker pull emqx/emqx:5.5.1

docker run -d --name emqx -p 1883:1883 -p 8083:8083 -p 8084:8084 -p 8883:8883 -p 18083:18083 emqx/emqx:5.5.

图片

通过浏览器访问 http://localhost:18083/（localhost 可替换为您的实际 IP 地址）以访问 EMQX Dashboard 管理控制台，进行设备连接与相关指标监控管理，默认用户名及密码：admin/public。

图片

登录成功之后如下图

图片

示例编写

图片

下面我们使用Java 语言，写一个示例，发送消息至主题mytopic ，订阅端分布为Java后端程序和JS订阅

Maven依赖项

创建工程并添加Maven依赖项，这里依赖的paho是 mqtt 的一个工具类

  org.eclipse.paho
  org.eclipse.paho.client.mqttv3
  1.2.5

创建发送消息代码

package cn.g2link.seg.base.mqtt.test;

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttTopic;

public class MqttPublishExample {
    public static void main(String[] args) {
        //emq 的 tcp监听端口
        String broker = "tcp://localhost:1883";
        String clientId = "mqtt_client1";
        //发送的主题
        String topic = "mytopic";
        //消息体
        String message = "Hello, MQTT!";

        try {
            MqttClient mqttClient = new MqttClient(broker, clientId);
            mqttClient.connect();

            MqttTopic mqttTopic = mqttClient.getTopic(topic);
            MqttMessage mqttMessage = new MqttMessage(message.getBytes());
            mqttTopic.publish(mqttMessage);

            mqttClient.disconnect();
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }
}

创建订阅消息代码

Java后端订阅

package cn.g2link.seg.base.mqtt.test;

import org.eclipse.paho.client.mqttv3.*;

public class MqttSubscribeExample {
    public static void main(String[] args) {
      //emq 的 tcp监听端口
        String broker = "tcp://localhost:1883";
        String clientId = "mqtt_subsribe_client1";
      //监听的主题
        String topic = "mytopic";

        try {
            MqttClient mqttClient = new MqttClient(broker, clientId);
            mqttClient.connect();
            System.out.println("connect success" );
            mqttClient.setCallback(new MqttCallback() {
                @Override
                public void connectionLost(Throwable cause) {
                    System.out.println("Connection lost!");
                }

                @Override
                public void messageArrived(String topic, MqttMessage message) throws Exception {
                    String payload = new String(message.getPayload());
                    System.out.println("Received message: " + payload);
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken token) {
                    // Not used in this example
                }
            });

            mqttClient.subscribe(topic);
            System.out.println(String.format("topic:%s subscribe success ", topic));
            // Keep the program running to receive messages
            while (true) {
                // Do nothing
            }
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }
}

浏览器端订阅

我们通过mqtt.min.js，来连接EMQX暴露的 webscoket 为8083端口，同时订阅mytopic主题

    
    
        // 将在全局初始化一个 mqtt 变量
        console.log(mqtt)

        // 创建一个 MQTT 客户端实例
        var client = mqtt.connect('mqtt://localhost:8083/mqtt', {
            clientId: 'web-mqtt-client' // 替换为您的客户端ID
        });

        // 连接成功时的回调函数
        client.on('connect', function () {
            console.log('已连接到 MQTT 服务器');

            // 订阅主题
            client.subscribe('mytopic'); // 替换为您要订阅的主题
        });

        // 接收到消息时的回调函数
        client.on('message', function (topic, message) {
            console.log('收到消息：', message.toString());
            // 在这里处理收到的消息，可以根据需要进行相应的逻辑操作
        });

        // 连接断开时的回调函数
        client.on('close', function () {
            console.log('与 MQTT 服务器的连接已断开');
        });

        // 连接错误时的回调函数
        client.on('error', function (error) {
            console.log('连接发生错误：', error);
        });
    

监控消息（可选）

在主题监控页面添加mytopic，这一步主要为了观察发送和消费的次数

图片

示例验证

订阅端启动

点击MqttSubscribeExample的 main 方法启动订阅

图片

图片

启动成功以后，会在EQMX 控制台，显示客户端连接信息，如下图显示了两个订阅端

图片

消息发送

启动MqttPublishExample的 main 方法，进行消息发送，发送后订阅端会收到以下消息

Java 后端

图片

浏览器端

图片

主题监控

查看EQMX 控制台的主题监控，会看到当前topic 流入和流出条数

图片

总结

以上只是简单介绍了什么是 EMQX 以及它的应用场景介绍，要想更多了解EMQX细节，可以访问官方进行了解。