Unlock High‑Performance IoT Messaging with mica‑mqtt: Low‑Latency Java MQTT Framework

1. Introduction

mica-mqtt is a simple, low‑latency, high‑performance MQTT open‑source component built on t‑io . See the source on Gitee and the

mica-mqtt-example

module for usage. It integrates easily into existing services and reduces IoT platform development cost.

2. Features

Supports MQTT v3.1, v3.1.1 and v5.0.

Supports WebSocket MQTT sub‑protocol (compatible with mqtt.js).

Provides HTTP REST API.

Offers MQTT client and server implementations.

Supports MQTT will messages and retained messages.

Custom message handling and cluster forwarding.

Includes an Alibaba Cloud MQTT connection demo.

Can be compiled to native executables with GraalVM.

Spring Boot starter for rapid integration.

Starter integrates with Prometheus + Grafana.

Cluster support via Redis pub/sub.

3. Use Cases

IoT cloud

IoT edge

Group chat IM

Message push

4. TODO

Optimize MQTT session handling and add some MQTT v5 features.

Implement rule engine using easy‑rule and druid SQL parsing.

5. Changelog

Added onDisconnect method to IMqttClientConnectListener.

Adjusted IMqttMessageListener interface (incompatible with older versions).

Modified broker message channels and added node management.

Changed default Message serialization (incompatible with older versions).

Optimized device online/offline handling for node shutdown.

Extracted mica‑mqtt-model module for future message bridging.

Improved Message model for clustering.

Fixed SSL configuration bug in MqttClient.

Fixed websocket packet splitting issue.

Updated README and documentation.

Upgraded tio to 3.7.5.v20211028‑RELEASE.

6.1 Client onDisconnect method

<code>public interface IMqttClientConnectListener {
    void onConnected(ChannelContext context, boolean isReconnect);
    void onDisconnect(ChannelContext channelContext, Throwable throwable, String remark, boolean isRemove);
}</code>

6.2 Server onMessage parameter adjustment

<code>@FunctionalInterface
public interface IMqttMessageListener {
    void onMessage(ChannelContext context, String clientId, Message message);
}</code>

6.3 Message object enhancements

The Message class now includes additional attributes and uses

ByteBuffer

based serialization for faster processing and smaller packets.

<code>byte[] data = DefaultMessageSerializer.INSTANCE.serialize(message);
Message message1 = DefaultMessageSerializer.INSTANCE.deserialize(data);
</code>

7. Spring Boot Quick Start

7.1 Add dependency

<code>&lt;dependency&gt;
    &lt;groupId&gt;net.dreamlu&lt;/groupId&gt;
    &lt;artifactId&gt;mica-mqtt-spring-boot-starter&lt;/artifactId&gt;
    &lt;version&gt;1.2.0&lt;/version&gt;
&lt;/dependency&gt;
</code>

7.2 Server configuration example

<code>mqtt:
  server:
    enabled: true
    ip: 0.0.0.0
    port: 5883
    name: Mica-Mqtt-Server
    buffer-allocator: HEAP
    heartbeat-timeout: 120000
    read-buffer-size: 8092
    max-bytes-in-message: 8092
    debug: true
    websocket-enable: true
    websocket-port: 8083
</code>

7.3 Server interfaces (register as Spring beans)

IMqttServerUniqueIdService (optional): custom unique ID when clientId is not unique.

IMqttServerAuthHandler (required): server authentication.

IMqttServerSubscribeValidator (required): subscription validation (added in 1.1.3).

IMqttMessageListener (required): message listener.

IMqttConnectStatusListener (required): connection status listener.

IMqttSessionManager (optional): session management.

IMqttMessageStore (required for clustering): stores will and retained messages.

AbstractMqttMessageDispatcher (required for clustering): forwards messages, including will and retained.

IpStatListener (optional): t‑io IP status listener.

7.4 Prometheus + Grafana monitoring

Thanks to t‑io design, metrics are exposed via t‑iostat. Supported metrics include connection counts, message packets and bytes handled, received, and sent.

For more details, see the mica‑mqtt‑spring‑boot‑starter documentation at https://gitee.com/596392912/mica-mqtt/tree/master/mica-mqtt-spring-boot-starter.

8. Plain Java Project Integration

8.1 Maven dependency

<code>&lt;dependency&gt;
    &lt;groupId&gt;net.dreamlu&lt;/groupId&gt;
    &lt;artifactId&gt;mica-mqtt-core&lt;/artifactId&gt;
    &lt;version&gt;1.2.0&lt;/version&gt;
&lt;/dependency&gt;
</code>

8.2 mica‑mqtt client example

<code>// Initialize MQTT client
MqttClient client = MqttClient.create()
    .ip("127.0.0.1")
    .port(1883)
    .username("admin")
    .password("123456")
    .version(MqttVersion.MQTT_5)
    .clientId("xxxxxx")
    .connect();

// Subscribe
client.subQos0("/test/#", (topic, payload) -> {
    logger.info(topic + '\t' + ByteBufferUtil.toString(payload));
});
client.unSubscribe("/test/#");

// Publish
client.publish("/test/client", ByteBuffer.wrap("mica最牛皮".getBytes(StandardCharsets.UTF_8)));

// Disconnect, reconnect, stop
client.disconnect();
client.reconnect();
client.stop();
</code>

8.3 mica‑mqtt server example

<code>// Reduce stack size for many connections: -Xss129k
MqttServer mqttServer = MqttServer.create()
    .ip("0.0.0.0")
    .port(1883)
    .readBufferSize(512)
    .messageListener((context, clientId, message) -> {
        logger.info("clientId:{} message:{} payload:{}", clientId, message, ByteBufferUtil.toString(message.getPayload()));
    })
    .debug()
    .start();

// Publish to a client
mqttServer.publish("clientId", "/test/123", ByteBuffer.wrap("mica最牛皮".getBytes()));
// Publish to all online listeners of a topic
mqttServer.publishAll("/test/123", ByteBuffer.wrap("mica最牛皮".getBytes()));
// Stop server
mqttServer.stop();
</code>

9. Follow Us

More exciting content is recommended daily!