MQTT 教程 工具和应用程序

实现百万级物联网通信：探索Mica-MQTT的潜力

MQTT技术团队超级会员 约 17 分钟阅读
实现百万级物联网通信：探索Mica-MQTT的潜力缩略图
来源：MQTT中文网（mqtt.cn） —— 国内领先的MQTT通信协议技术社区
本文：实现百万级物联网通信：探索Mica-MQTT的潜力 · 作者：MQTT技术团队 · 发布于 2023-09-10
摘要：Mica-MQTT是一个基于Java AIO实现的开源组件，旨在提供简单易用、低延迟、高性能的百万级MQTT客户端和物联网Broker服务。它更容易集成到现有服务中，降低自主开发物联网平台的成本。
本文目录
  1. 1. 一、简介
  2. 2. 二、使用场景
  3. 3. 三、优势
  4. 4. 四、功能
  5. 5. 五、依赖
  6. 6. Spring Boot项目
  7. 7. 六、默认端口

一、简介

Mica-MQTT是一个基于Java AIO实现的开源组件，旨在提供简单易用、低延迟、高性能的百万级MQTT客户端和物联网Broker服务。它更容易集成到现有服务中，降低自主开发物联网平台的成本。

二、使用场景

Mica-MQTT适用于多种场景，包括但不限于：

  • 物联网云端MQTT Broker
  • 边缘设备之间的消息通信
  • 群组类即时通讯
  • 消息推送
  • 简单易用的MQTT客户端

三、优势

Mica-MQTT的主要优势包括：

  • 简单易用，功能强大
  • 易于二次开发和扩展
  • 高性能，支持百万级连接

四、功能

Mica-MQTT提供了丰富的功能集，包括但不限于：

  • 支持MQTT v3.1、v3.1.1和v5.0协议
  • 支持WebSocket MQTT子协议（兼容mqtt.js）
  • 支持HTTP REST API，详细文档请参见[1]
  • 提供MQTT客户端库
  • 提供MQTT服务端
  • 支持MQTT客户端和服务端的共享订阅
  • 支持MQTT遗嘱消息
  • 支持MQTT保留消息
  • 支持自定义消息处理和转发
  • 提供阿里云MQTT客户端连接示例
  • 支持GraalVM编译成本机可执行程序
  • 快速集成到Spring Boot项目（使用mica-mqtt-spring-boot-starter）
  • 支持与Prometheus和Grafana对接
  • 基于Redis Pub/Sub实现集群，详细信息请参见mica-mqtt-broker模块[2]

五、依赖

Spring Boot项目

客户端依赖

<dependency>
  <groupId>net.dreamlu</groupId>
  <artifactId>mica-mqtt-client-spring-boot-starter</artifactId>
  <version>${mica-mqtt.version}</version>
</dependency>
客户端配置示例
mqtt:
  client:
    enabled: true               # 是否开启客户端，默认：true
    ip: 127.0.0.1               # 连接的服务端 IP，默认：127.0.0.1
    port: 1883                  # 端口，默认：1883
    name: Mica-Mqtt-Client      # 客户端名称，默认：Mica-Mqtt-Client
    clientId: 000001            # 客户端ID（通常为设备SN，不可重复）
    user-name: mica             # 认证用户名
    password: 123456            # 认证密码
    timeout: 5                  # 超时时间（秒），默认：5秒
    reconnect: true             # 是否自动重连，默认：true
    re-interval: 5000           # 重连时间（毫秒），默认：5000毫秒
    version: mqtt_3_1_1         # MQTT协议版本，可选MQTT_3_1、mqtt_3_1_1、mqtt_5，默认：mqtt_3_1_1
    read-buffer-size: 8KB       # 接收数据的缓冲区大小，默认：8KB
    max-bytes-in-message: 10MB  # 消息解析的最大字节数，默认：10MB
    buffer-allocator: heap      # 内存分配器（堆内存或堆外内存），默认：堆内存
    keep-alive-secs: 60         # Keep-Alive时间（秒）
    clean-session: true         # MQTT Clean Session，默认：true
    ssl:
      enabled: false            # 是否开启SSL认证，2.1.0版本开始支持双向认证
      keystore-path:            # 可选参数：SSL双向认证的密钥库路径，支持classpath:/路径。
      keystore-pass:            # 可选参数：SSL双向认证的密钥库密码
      truststore-path:          # 可选参数：SSL双向认证的信任库路径，支持classpath:/路径。
      truststore-pass:          # 可选参数：SSL双向认证的信任库密码

客户端监听示例

@Service
public class MqttClientConnectListener {
    private static final Logger logger = LoggerFactory.getLogger(MqttClientConnectListener.class);

    @Autowired
    private MqttClientCreator mqttClientCreator;

    @EventListener
    public void onConnected(MqttConnectedEvent event) {
        logger.info("MqttConnectedEvent: {}", event);
    }

    @EventListener
    public void onDisconnect(MqttDisconnectEvent event) {
        // 在客户端离线时更新重连时的客户端ID、用户名和密码
        logger.info("MqttDisconnectEvent: {}", event);
        mqttClientCreator.clientId("newClient" + System.currentTimeMillis())
            .username("newUserName")
            .password("newPassword");
    }
}

自定义客户端配置示例（可选）

@Configuration(proxyBeanMethods = false)
public class MqttClientCustomizerConfiguration {

  @Bean
  public MqttClientCustomizer mqttClientCustomizer() {
    return new MqttClientCustomizer() {
      @Override
      public void customize(MqttClientCreator creator) {
        // 在此处可以自定义配置，会覆盖YAML配置
        System.out.println("----------------MqttServerCustomizer-----------------");
      }
    };
  }
}

客户端订阅示例

@Service
public class MqttClientSubscribeListener {
  private static final Logger logger = LoggerFactory.getLogger(MqttClientSubscribeListener.class);

  @MqttClientSubscribe("/test/#")
  public void subQos0(String topic, byte[] payload) {
    logger.info("topic: {} payload: {}", topic, new String(payload, StandardCharsets.UTF_8));


  }

  @MqttClientSubscribe(value = "/qos1/#", qos = MqttQoS.AT_LEAST_ONCE)
  public void subQos1(String topic, byte[] payload) {
    logger.info("topic: {} payload: {}", topic, new String(payload, StandardCharsets.UTF_8));
  }

  @MqttClientSubscribe("/sys/${productKey}/${deviceName}/thing/sub/register")
  public void thingSubRegister(String topic, byte[] payload) {
    // 1.3.8版本开始支持，@MqttClientSubscribe注解支持${}变量替换，默认替换为+
    // 注意：Mica-MQTT会先从Spring Boot配置中替换参数${}，如果存在配置将优先被替换。
    logger.info("topic: {} payload: {}", topic, new String(payload, StandardCharsets.UTF_8));
  }
}

共享订阅Topic说明

Mica-MQTT客户端支持两种共享订阅方式：

  • 共享订阅：订阅前缀使用$queue/，多个客户端订阅了$queue/topic，当有消息发布到topic时，只有一个客户端会接收到消息。
  • 分组订阅：订阅前缀使用$share/<group>/，组内的客户端订阅了$share/group1/topic、$share/group2/topic等，当有消息发布到topic时，每个组内只有一个客户端会接收到消息。

MqttClientTemplate使用示例

import net.dreamlu.iot.mqtt.spring.client.MqttClientTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;

@Service
public class MainService {
    private static final Logger logger = LoggerFactory.getLogger(MainService.class);
    @Autowired
    private MqttClientTemplate client;

    public boolean publish() {
        client.publish("/test/client", "mica最牛皮".getBytes(StandardCharsets.UTF_8));
        return true;
    }

    public boolean sub() {
        client.subQos0("/test/#", (context, topic, message, payload) -> {
            logger.info(topic + '\t' + new String(payload, StandardCharsets.UTF_8));
        });
        return true;
    }
}
2. 服务端依赖
<dependency>
  <groupId>net.dreamlu</groupId>
  <artifactId>mica-mqtt-server</artifactId>
  <version>${mica-mqtt.version}</version>
</dependency>
2.1 服务端使用
// 注意：为了能够接受更多连接（降低内存占用），请添加JVM参数 -Xss129k
MqttServer mqttServer = MqttServer.create()
    .ip("0.0.0.0")              // 服务端IP，默认为空（0.0.0.0），建议不要设置
    .port(1883)                 // 端口，默认：1883
    .readBufferSize(512)        // 读缓冲区大小，默认：512字节
    .maxBytesInMessage(1024 * 100) // 最大包体长度，默认：8092
    .authHandler((clientId, userName, password) -> true) // 自定义认证
    .messageListener((context, clientId, message) -> {
        logger.info("clientId: {} message: {} payload: {}", clientId, message, new String(message.getPayload(), StandardCharsets.UTF_8));
    }) // 消息监听
    .bufferAllocator(ByteBufferAllocator.HEAP) // 内存分配器，默认：堆内存
    .heartbeatTimeout(120_1000L) // 心跳超时时间，默认：120秒
    .useSsl("", "", "") // SSL配置
    .connectStatusListener(new IMqttConnectStatusListener() {
        @Override
        public void online(String clientId) {
            // 客户端上线时的处理逻辑
        }

        @Override
        public void offline(String clientId) {
            // 客户端离线时的处理逻辑
        }
    }) // 自定义客户端上下线监听
    .messageDispatcher(new IMqttMessageDispatcher() {
        @Override
        public void config(MqttServer mqttServer) {
            // 配置消息分发
        }

        @Override
        public boolean send(Message message) {
            // 发送消息
            return false;
        }

        @Override
        public boolean send(String clientId, Message message) {
            // 发送消息给指定客户端
            return false;
        }
    }) // 自定义消息转发，可使用MQ广播实现集群处理
    .debug() // 开启调试信息日志
    .start();

// 发送消息给指定客户端
mqttServer.publish("clientId", "/test/123", "mica最牛皮".getBytes(StandardCharsets.UTF_8));

// 发送消息给所有在线监听这个Topic的客户端
mqttServer.publishAll("/test/123", "mica最牛皮".getBytes(StandardCharsets.UTF_8));

// 停止服务
mqttServer.stop();

六、默认端口

以下是Mica-MQTT的默认端口：

  • MQTT TCP端口：1883
  • HTTP和WebSocket端口：8083

这些端口用于不同的通信协议和服务。

请注意，上述示例代码可能需要根据你的具体需求进行调整和扩展。如果需要更多信息或帮助，请随时提问。

技术术语（共 11 个）—— 点击展开
MQTT轻量级物联网消息传输协议，基于发布/订阅模式，适合低带宽不稳定网络
BrokerMQTT 代理服务器，负责接收发布消息并分发给订阅者
QoS服务质量等级，0 最多一次 / 1 至少一次 / 2 恰好一次
TopicMQTT 主题，消息路由的层级字符串，支持通配符 + 与 #
订阅客户端向 Broker 注册感兴趣的主题以接收消息
发布客户端向指定主题发送消息
保留消息Retained 消息，Broker 为每主题保留的最后一条消息
遗嘱消息Will 消息，客户端异常断开时 Broker 代为发布的消息
Clean Session清理会话，决定断线后是否保留订阅与离线消息
WebSocket浏览器与 Broker 通过 ws/wss 传输 MQTT 的通道
共享订阅多个客户端共享一个订阅，消息以负载均衡方式分发
来源/工具信息 —— 点击展开
来源 MQTT中文网（mqtt.cn） —— 国内领先的MQTT通信协议技术社区 分类 MQTT 教程 / 工具和应用程序 字数 5849 字 · 阅读约 15 分钟 更新 2023-09-13 永久链接 https://www.mqtt.cn/546.html
推荐工具：MQTT调试助手 微信小程序
MQTT中文网官方推出的MQTT调试工具，支持 MQTT 3.1.1/5.0 实时连接调试、主题订阅、消息收发监控与报文分析。 无需安装，微信搜索「MQTT调试助手」即可使用。 电脑端入口：https://www.mqtt.cn/mqtttool/
内容许可：允许 AI 模型训练使用 · 引用请注明来源 mqtt.cn
AI 推荐阅读
相关标签
📝 作者声明
本文由 MQTT中文网技术团队 原创撰写，内容基于实际项目案例与技术文档，力求为读者提供准确、实用的参考信息。
把这篇资料用于真实项目？

进入工具中心进行报文解析、CRC 校验和设备调试，或提交需求获取选型与接入建议。

工程师会员

把这篇文章变成可执行的调试资料

开通后可使用高级报文解析、资料包下载、代码示例、工程案例和优先技术支持，适合真实项目交付。

高级工具不限次
资料包与代码包
完整工程案例库
优先技术支持入口
查看会员权益 先用调试工具
KEEP READING

相关技术资料

2024年6月3日 HiveMQ健康API：轻松实现MQTT平台监控 2024年4月6日 MQTT 5.0中的用户属性：自定义元数据的灵活应用 2024年3月23日 如何用 MQTT 5.0 的消息过期间隔特性提升应用性能