Unlock Low-Latency IoT Messaging with mica-mqtt: A Java AIO MQTT Solution

1. Introduction

mica-mqtt is a simple, low‑latency, high‑performance MQTT IoT open‑source component implemented with java aio . It is easy to integrate into existing services and reduces development cost of custom IoT platforms.

2. Features

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

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

Provides HTTP REST API (documentation omitted).

Provides MQTT client.

Provides MQTT server.

Supports MQTT will messages.

Supports MQTT retained messages.

Custom message (mq) processing and forwarding for clustering.

Demo for Alibaba Cloud MQTT connection.

Supports GraalVM compilation to native executable.

Spring Boot starter for quick integration (mica-mqtt-spring-boot-starter).

Starter integrates with Prometheus + Grafana.

Cluster implementation based on Redis pub/sub (see mica-mqtt-broker module).

3. Use Cases

IoT cloud‑side MQTT broker.

IoT edge‑side message communication.

Group chat IM.

Message push services.

Simple, easy‑to‑use MQTT client.

4. Release Notes

v2.2.12 – 2024‑04‑16

✨ Simplified MqttClientTemplate construction in mica-mqtt-client-spring-boot-starter.

✨ Optimized Spring event MQTT client connection listener.

✨ Optimized annotation subscription.

🐛 Fixed encoding issue when configuring both MQTT5 props and will message.

🐛 Fixed will‑message sending condition in mica-mqtt-server (issue introduced in 2.2.7).

5. Custom MqttClientTemplate

In some scenarios multiple MQTT clients are needed to handle and forward data between several brokers.

For pure data processing and forwarding, launching two mica‑mqtt clients directly from a main method is simpler and results in a ~1 MB jar that starts very quickly.

<code>/**
 * 2 MQTT services communicate using 2 clients for data transfer
 *
 * @author L.cm
 */
public class MqttClientProxy {

    public static void main(String[] args) {
        MqttClient client1 = MqttClient.create()
            .ip("ip1")
            .port(1883)
            .clientId("clientI")
            .username("mica")
            .password("mica")
            .debug()
            .connectSync();

        MqttClient client2 = MqttClient.create()
            .ip("ip2")
            .port(1883)
            .clientId("client2")
            .username("mica")
            .password("mica")
            .debug()
            .connectSync();

        String[] topics = new String[]{
            "$share/test/link/product1/+/event/+/post",
            "$share/test/link/product2/+/event/+/post",
            "$share/test/link/product3/+/event/+/post"
        };
        client1.subscribe(topics, MqttQoS.AT_MOST_ONCE, (context, topic, message, payload) -> {
            client2.publish(topic, payload);
        });
    }

}
</code>

Custom MqttClientTemplate Bean

a. Custom bean (simplified since 2.2.11)

<code>@Configuration
public class OtherMqttClientConfiguration {

    @Bean("mqttClientTemplate1")
    public MqttClientTemplate mqttClientTemplate1() {
        MqttClientCreator mqttClientCreator1 = MqttClient.create()
            .ip("mqtt.dreamlu.net")
            .username("mica")
            .password("mica");
        return new MqttClientTemplate(mqttClientCreator1);
    }

}
</code>

b. Modify starter’s default MqttClientTemplate bean injection

Because a new bean named

mqttClientTemplate1

is added, the original starter’s MqttClientTemplate bean also needs a bean name.

<code>@Autowired
@Qualifier(MqttClientTemplate.DEFAULT_CLIENT_TEMPLATE_BEAN)
private MqttClientTemplate mqttClientTemplate;
</code>

c. Inject the new mqttClientTemplate1 bean

<code>@Autowired
@Qualifier("mqttClientTemplate1")
private MqttClientTemplate mqttClientTemplate;
</code>

d. Annotation subscription with the new bean

Note: The @MqttClientSubscribe annotation defaults to

MqttClientTemplate.DEFAULT_CLIENT_TEMPLATE_BEAN

; therefore the new bean must be specified.

<code>@MqttClientSubscribe(
    value = "/#",
    clientTemplateBean = "mqttClientTemplate1"
)
public void sub1(String topic, byte[] payload) {
    logger.info("topic:{} payload:{}", topic, ByteBufferUtil.toString(payload));
}
</code>