mica-mqtt 低延迟、高性能的 mqtt 物联网组件-Toy模板网

这篇具有很好参考价值的文章主要介绍了mica-mqtt 低延迟、高性能的 mqtt 物联网组件。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

一、简介

mica-mqtt 是基于 java aio 实现，简单易用、低延迟、高性能百万级 mqtt client 物联网开源组件和 mqtt broker 服务，更加易于集成到已有服务和二次开发，降低自研物联网平台开发成本。

二、使用场景

物联网（云端 mqtt broker）
物联网（边缘端消息通信）
群组类 IM
消息推送
简单易用的 mqtt 客户端

三、优势

平凡却不单调，简单却不失精彩。
手动档（更加易于二次开发或扩展）。
牛犊初生，无限可能。

四、功能

支持 MQTT v3.1、v3.1.1 以及 v5.0 协议。
支持 websocket mqtt 子协议（支持 mqtt.js）。
支持 http rest api，http api 文档详见[1]。
支持 MQTT client 客户端。
支持 MQTT server 服务端。
支持 MQTT client、server 共享订阅支持（捐助VIP版采用 topic 树存储，跟 topic 数无关，百万 topic 性能依旧）。
支持 MQTT 遗嘱消息。
支持 MQTT 保留消息。
支持自定义消息（mq）处理转发实现集群。
MQTT 客户端阿里云 mqtt 连接 demo。
支持 GraalVM 编译成本机可执行程序。
支持 Spring boot 项目快速接入（mica-mqtt-spring-boot-starter）。
mica-mqtt-spring-boot-starter 支持对接 Prometheus + Grafana。
基于 redis pub/sub 实现集群，详见 mica-mqtt-broker 模块[2]。

五、依赖

Spring boot 项目

1. 客户端


<dependency>
  <groupId>net.dreamlu</groupId>
  <artifactId>mica-mqtt-client-spring-boot-starter</artifactId>
  <version>${mica-mqtt.version}</version>
</dependency>

1.1 客户端配置项示例

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       # 接收数据的 buffer size，默认：8k
    max-bytes-in-message: 10MB  # 消息解析最大 bytes 长度，默认：10M
    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 双向认证 keystore 目录，支持 classpath:/ 路径。
      keystore-pass:            # 可选参数：ssl 双向认证 keystore 密码
      truststore-path:          # 可选参数：ssl 双向认证 truststore 目录，支持 classpath:/ 路径。
      truststore-pass:          # 可选参数：ssl 双向认证 truststore 密码

注意：ssl 存在三种情况

服务端开启ssl	客户端
ClientAuth 为 NONE（不需要客户端验证）	仅仅需要开启 ssl 即可不用配置证书
ClientAuth 为 OPTIONAL（与客户端协商）	需开启 ssl 并且配置 truststore 证书
ClientAuth 为 REQUIRE (必须的客户端验证)	需开启 ssl 并且配置 truststore、 keystore证书

1.2 可实现接口（注册成 Spring Bean 即可）

接口	是否必须	说明
IMqttClientConnectListener	否	客户端连接成功监听

1.3 客户端上下线监听

使用 Spring event 解耦客户端上下线监听，注意： 1.3.4 开始支持。会跟自定义的 IMqttClientConnectListener 实现冲突，取一即可。


/**
 * 示例：客户端连接状态监听
 *
 * @author L.cm
 */
@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) {
        // 离线时更新重连时的密码，适用于类似阿里云 mqtt clientId 连接带时间戳的方式 
        logger.info("MqttDisconnectEvent:{}", event);
        // 在断线时更新 clientId、username、password
        mqttClientCreator.clientId("newClient" + System.currentTimeMillis())
            .username("newUserName")
            .password("newPassword");
    }

}

1.4 客户端自定义 java 配置（可选）


@Configuration(proxyBeanMethods = false)
public class MqttClientCustomizerConfiguration {

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

}

1.5 客户端订阅示例


@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));
  }

}

1.6 共享订阅 topic 说明

mica-mqtt client 支持两种共享订阅方式：

共享订阅：订阅前缀 $queue/，多个客户端订阅了 $queue/topic，发布者发布到topic，则只有一个客户端会接收到消息。
分组订阅：订阅前缀 $share/<group>/，组客户端订阅了$share/group1/topic、$share/group2/topic..，发布者发布到topic，则消息会发布到每个group中，但是每个group中只有一个客户端会接收到消息。

1.7 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;

/**
 * @author wsq
 */
@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")
    // 默认：1883
    .port(1883)
    // 默认为：8092（mqtt 默认最大消息大小），为了降低内存可以减小小此参数，如果消息过大 t-io 会尝试解析多次（建议根据实际业务情况而定）
    .readBufferSize(512)
    // 最大包体长度，如果包体过大需要设置此参数，默认为：8092
    .maxBytesInMessage(1024 * 100)
    // 自定义认证
    .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)
    // 心跳超时时间，默认：120s
    .heartbeatTimeout(120_1000L)
    // ssl 配置
    .useSsl("", "", "")
    // 自定义客户端上下线监听
    .connectStatusListener(new IMqttConnectStatusListener() {
        @Override
        public void online(String clientId) {

        }

        @Override
        public void offline(String clientId) {

        }
    })
    // 自定义消息转发，可用 mq 广播实现集群化处理
    .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;
        }
    })
    .debug() // 开启 debug 信息日志
    .start();

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

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

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