mica-mqtt client,最好用的 java mqtt 客户端

JAVA架构日记

共 16003字,需浏览 33分钟

 ·

2022-07-27 06:49

简介

mica-mqtt 基于 t-io 实现的简单低延迟高性能 的 mqtt 物联网开源组件。

mica-mqtt server 更加易于集成到已有服务和二次开发,降低自研物联网平台开发成本。

mica-mqtt client 是简单、易用的 java mqtt 客户端,更加容易集成到自己的业务代码中。今天笔者主要要介绍的就是 mica-mqtt client 的使用。

使用

mica-mqtt-client Spring boot starter

添加依赖

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

配置项说明

mqtt:
  client:
    enabled: true               # 是否开启客户端,默认:false 使用到的场景有限,非必要请不要启用
    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_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
    use-ssl: false              # 是否启用 ssl,默认:false

连接状态监听

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

}

自定义配置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-----------------");
   }
  };
 }

}

订阅示例

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

 @MqttClientSubscribe("/test/#")
 public void subQos0(String topic, ByteBuffer payload) {
  logger.info("topic:{} payload:{}", topic, ByteBufferUtil.toString(payload));
 }

 @MqttClientSubscribe(value = "/qos1/#", qos = MqttQoS.AT_LEAST_ONCE)
 public void subQos1(String topic, ByteBuffer payload) {
  logger.info("topic:{} payload:{}", topic, ByteBufferUtil.toString(payload));
 }

}

MqttClientTemplate 使用示例

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

    public boolean publish() {
        // 发布消息示例
        client.publish("/test/client", ByteBuffer.wrap("mica最牛皮".getBytes(StandardCharsets.UTF_8)));
        return true;
    }

    public boolean sub() {
        // 订阅消息示例
        client.subQos0("/test/#", (topic, payload) -> {
            logger.info(topic + '\t' + ByteBufferUtil.toString(payload));
        });
        return true;
    }

}

共享订阅 topic 说明

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

  • 共享订阅:订阅前缀 $queue/,多个客户端订阅了 $queue/topic,发布者发布到topic,则只有一个客户端会接收到消息。

  • 分组订阅:订阅前缀 $share/<group>/,组客户端订阅了$share/group1/topic$share/group2/topic..,发布者发布到topic,则消息会发布到每个group中,但是每个group中只有一个客户端会接收到消息。

jfinal mica-mqtt client(1.3.7 开始支持)

添加依赖

<dependency>
 <groupId>net.dreamlu</groupId>
 <artifactId>jfinal-mica-mqtt-client</artifactId>
 <version>1.3.7</version>
</dependency>

删除 jfinal-demo 中的 slf4j-nop 依赖

添加 slf4j-log4j12

<dependency>
 <groupId>org.slf4j</groupId>
 <artifactId>slf4j-log4j12</artifactId>
 <version>1.7.33</version>
</dependency>

在 jfinal Config configPlugin 中添加 mica-mqtt client 插件

MqttClientPlugin mqttClientPlugin = new MqttClientPlugin();
mqttClientPlugin.config(mqttClientCreator -> {
 // 设置 mqtt 连接配置信息
 mqttClientCreator
   .clientId("clientId"// 按需配置,相同的会互踢
   .ip("mqtt.dreamlu.net")
   .port(1883)
   .connectListener(Aop.get(MqttClientConnectListener.class));
});
me.add(mqttClientPlugin);

在 jfinal Config onStart 启动完成之后添加 mqtt 订阅

@Override
public void onStart() {
    IMqttClientMessageListener clientMessageListener = Aop.get(TestMqttClientMessageListener.class);
    MqttClientKit.subQos0("#", clientMessageListener);
}

使用 MqttClientKit 发送消息

MqttClientKit.publish("mica""hello".getBytes(StandardCharsets.UTF_8));

示例代码 MqttClientConnectListener

public class MqttClientConnectListener implements IMqttClientConnectListener {

    @Override
    public void onConnected(ChannelContext channelContext, boolean isReconnect) {
        if (isReconnect) {
            System.out.println("重连 mqtt 服务器重连成功...");
        } else {
            System.out.println("连接 mqtt 服务器成功...");
        }
    }

    @Override
    public void onDisconnect(ChannelContext channelContext, Throwable throwable, String remark, boolean isRemove) {
        System.out.println("mqtt 链接断开 remark:" + remark + " isRemove:" + isRemove);
    }
}

示例 TestMqttClientMessageListener

public class TestMqttClientMessageListener implements IMqttClientMessageListener {
    @Override
    public void onMessage(String topic, MqttPublishMessage mqttPublishMessage, ByteBuffer byteBuffer) {
        System.out.println("收到消息 topic:" + topic + "内容:\n" + ByteBufferUtil.toString(byteBuffer));
    }
}

其它 java 项目

添加依赖

<dependency>
    <groupId>net.dreamlu</groupId>
    <artifactId>mica-mqtt-core</artifactId>
    <version>1.3.7</version>
    <exclusions>
        <exclusion>
            <groupId>org.t-io</groupId>
            <artifactId>tio-websocket-server</artifactId>
        </exclusion>
        <exclusion>
            <groupId>net.dreamlu</groupId>
            <artifactId>mica-mqtt-model</artifactId>
        </exclusion>
        <exclusion>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
        </exclusion>
    </exclusions>
</dependency>

使用

// 初始化 mqtt 客户端
MqttClient client = MqttClient.create()
    .ip("127.0.0.1")                // mqtt 服务端 ip 地址
    .port(1883)                     // 默认:1883
    .username("admin")              // 账号
    .password("123456")             // 密码
    .version(MqttVersion.MQTT_5)    // 默认:3_1_1
    .clientId("xxxxxx")             // 非常重要务必手动设置,一般设备 sn 号,默认:MICA-MQTT- 前缀和 36进制的纳秒数
    .bufferAllocator(ByteBufferAllocator.DIRECT) // 堆内存和堆外内存,默认:堆内存
    .readBufferSize(512)            // 消息一起解析的长度,默认:为 8092 (mqtt 消息最大长度)
    .maxBytesInMessage(1024 * 10)   // 最大包体长度,如果包体过大需要设置此参数,默认为:10M (10*1024*1024)
    .keepAliveSecs(120)             // 默认:60s
    .timeout(10)                    // 超时时间,t-io 配置,可为 null,为 null 时,t-io 默认为 5
    .reconnect(true)                // 是否重连,默认:true
    .reInterval(5000)               // 重连重试时间,reconnect 为 true 时有效,t-io 默认为:5000
    .willMessage(builder -> {
        builder.topic("/test/offline").messageText("down");    // 遗嘱消息
    })
    .connectListener(new IMqttClientConnectListener() {
        @Override
        public void onConnected(ChannelContext context, boolean isReconnect) {
            logger.info("链接服务器成功...");
        }
        
        @Override
        public void onDisconnect(ChannelContext channelContext, Throwable throwable, String remark, boolean isRemove) {
            logger.info("与链接服务器断开连接...");
        }
    })
    .properties()                   // mqtt5 properties
    .connect();                     // 异步连接,还支持同步 connectSync()

    // 消息订阅,同类方法 subxxx
    client.subQos0("/test/#", (topic, payload) -> {
        logger.info(topic + '\t' + ByteBufferUtil.toString(payload));
    });
    // 取消订阅
    client.unSubscribe("/test/#");

    // 发送消息
    client.publish("/test/client", ByteBuffer.wrap("mica-mqtt 牛皮".getBytes(StandardCharsets.UTF_8)));

    // 断开连接
    client.disconnect();
    // 重连
    client.reconnect();

鸣谢

mica-mqtt 从一个试验性的项目逐渐完善,目前 gitee 上已有 800 多颗星。

mica-mqtt 的成长也离不开大伙使用和积极反馈,感谢 @冷月宫主@willianfu@hjkJOJO@Symous@hongfeng11@胡萝博@杨钊@一醉化千愁@toskeyfine 、@亡羊补牛等同学,谢谢大家!!!


浏览 61
点赞
评论
收藏
分享

手机扫一扫分享

分享
举报
评论
图片
表情
推荐
点赞
评论
收藏
分享

手机扫一扫分享

分享
举报