Unlock Low-Latency MQTT with mica-mqtt: A High-Performance Java Component

1. Introduction

mica-mqtt is a simple, low‑latency, high‑performance MQTT open‑source component built on t‑io . It is easy to integrate into existing services and reduces IoT platform development cost. See the mica-mqtt Gitee source and the mica-mqtt-example module for usage.

2. Features

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

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

Provides HTTP REST API.

Provides MQTT client and server implementations.

Supports will messages and retained messages.

Custom message processing for clustering.

Demo for Alibaba Cloud MQTT connection.

GraalVM native compilation.

Spring Boot starter (

mica-mqtt-spring-boot-starter

) for quick integration.

Prometheus + Grafana metrics via the starter.

Cluster support based on Redis pub/sub (see

mica-mqtt-broker

module).

3. TODO

Optimize MQTT session handling and add some MQTT v5.0 features.

4. Changelog

1.1.4

Added

IMqttServerUniqueIdService

to handle non‑unique clientId scenarios.

Adjusted

IMqttServerAuthHandler

to include

uniqueId

parameter.

1.1.3

Added

ChannelContext

to

IMqttConnectStatusListener

.

Split subscription validation into

IMqttServerSubscribeValidator

with

ChannelContext

and

clientId

.

Adjusted

IMqttServerAuthHandler

package and added

ChannelContext

.

Improved documentation and examples, added default port description.

Upgraded dependencies.

5. Spring Boot Quick Start

5.1 Add Dependency

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

5.2 Server Configuration Example

<code>mqtt:
  server:
    enabled: true
    ip: 127.0.0.1
    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>

5.3 Server Implementable Interfaces

Register the following interfaces as Spring beans as needed:

IMqttServerUniqueIdService

(optional) – custom unique ID for non‑unique clientId.

IMqttServerAuthHandler

(required) – server‑side authentication.

IMqttServerSubscribeValidator

(required) – subscription validation.

IMqttMessageListener

(required) – message listener.

IMqttConnectStatusListener

(required) – connection status listener.

IMqttSessionManager

(optional) – session management.

IMqttMessageStore

– storage for will and retained messages (clustered).

AbstractMqttMessageDispatcher

– message forwarding (will/retained).

IpStatListener

(optional) – t‑io IP status listener.

5.4 Prometheus + Grafana Monitoring

Thanks to t‑io’s design, metrics are exposed via

t-iostat

. Supported metrics include connection counts, message packets, bytes processed, etc.

6. Plain Java Project Integration

6.1 Maven Dependency

<code><dependency>
    <groupId>net.dreamlu</groupId>
    <artifactId>mica-mqtt-core</artifactId>
    <version>1.1.4</version>
</dependency>
</code>

6.2 mqtt Client Example

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

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

// Unsubscribe
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>

6.3 mqtt Server Example

<code>// Increase stack size for many connections: -Xss129k
MqttServer mqttServer = MqttServer.create()
    .ip("127.0.0.1")
    .port(1883)
    .readBufferSize(512) // adjust as needed
    .messageListener((clientId, topic, mqttQoS, payload) -> {
        logger.info("clientId:{} topic:{} mqttQoS:{} message:{}",
            clientId, topic, mqttQoS, ByteBufferUtil.toString(payload));
    })
    .debug()
    .start();

// Send to a specific client
mqttServer.publish("clientId", "/test/123", ByteBuffer.wrap("mica最牛皮".getBytes()));

// Broadcast to all online listeners of a topic
mqttServer.publishAll("/test/123", ByteBuffer.wrap("mica最牛皮".getBytes()));

// Stop server
mqttServer.stop();
</code>

7. Follow Us

More exciting content recommended daily!