How to Integrate and Use mica-mqtt Java Client in Spring Boot

Introduction

mica-mqtt is a simple, low‑latency, high‑performance MQTT open‑source component built on t‑io . The server component reduces IoT platform development cost, and the Java client is easy to integrate into business code.

Usage

mica-mqtt-client Spring Boot starter

Add dependency

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

Configuration options

<code>mqtt:
  client:
    enabled: true            # enable client, default false
    ip: 127.0.0.1            # server IP, default 127.0.0.1
    port: 1883               # default 1883
    name: Mica-Mqtt-Client   # default name
    clientId: 000001         # unique client ID, usually device SN
    user-name: mica           # username
    password: 123456         # password
    timeout: 5               # seconds, default 5
    reconnect: true          # default true
    re-interval: 5000        # ms, default 5000
    version: MQTT_5          # default 3.1.1
    read-buffer-size: 8KB
    max-bytes-in-message: 10MB
    buffer-allocator: heap
    keep-alive-secs: 60
    clean-session: true
    use-ssl: false</code>

Connection status listener

<code>@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) {
        // Update credentials on disconnect
        logger.info("MqttDisconnectEvent:{}", event);
        mqttClientCreator.clientId("newClient" + System.currentTimeMillis())
                         .username("newUserName")
                         .password("newPassword");
    }
}</code>

Custom Java configuration (optional)

<code>@Configuration(proxyBeanMethods = false)
public class MqttClientCustomizerConfiguration {
    @Bean
    public MqttClientCustomizer mqttClientCustomizer() {
        return new MqttClientCustomizer() {
            @Override
            public void customize(MqttClientCreator creator) {
                // custom configuration overrides yaml
                System.out.println("----MqttServerCustomizer----");
            }
        };
    }
}</code>

Subscription example

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

MqttClientTemplate usage

<code>@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;
    }
}</code>

Shared subscription topics

mica-mqtt client supports two shared‑subscription modes:

Queue subscription: prefix

$queue/

. Multiple clients subscribe to

$queue/topic

; only one receives each message.

Group subscription: prefix

$share/<group>/

. Clients in the same group share messages; each group receives the message once.

jfinal mica-mqtt client (since 1.3.7)

Add dependency

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

Remove

slf4j-nop

from the demo and add

slf4j-log4j12

version 1.7.33.

Register the plugin in

Config.configPlugin

:

<code>MqttClientPlugin mqttClientPlugin = new MqttClientPlugin();
mqttClientPlugin.config(mqttClientCreator -> {
    // configure connection
    mqttClientCreator.clientId("clientId")
                     .ip("mqtt.dreamlu.net")
                     .port(1883)
                     .connectListener(Aop.get(MqttClientConnectListener.class));
});
me.add(mqttClientPlugin);</code>

Add subscriptions after the application starts:

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

Send a message with

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

Other Java projects

Add dependency

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

Initialize and use the client:

<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")
    .bufferAllocator(ByteBufferAllocator.DIRECT)
    .readBufferSize(512)
    .maxBytesInMessage(1024 * 10)
    .keepAliveSecs(120)
    .timeout(10)
    .reconnect(true)
    .reInterval(5000)
    .willMessage(builder -> {
        builder.topic("/test/offline").messageText("down");
    })
    .connectListener(new IMqttClientConnectListener() {
        @Override
        public void onConnected(ChannelContext context, boolean isReconnect) {
            logger.info("Connected to MQTT server...");
        }
        @Override
        public void onDisconnect(ChannelContext ctx, Throwable t, String remark, boolean isRemove) {
            logger.info("Disconnected: " + remark);
        }
    })
    .properties()
    .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-mqtt 牛皮".getBytes(StandardCharsets.UTF_8)));

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

Acknowledgments

mica-mqtt has gained over 800 stars on Gitee. Thanks to contributors such as @冷月宫主, @willianfu, @hjkJOJO, @Symous, @hongfeng11, @胡萝博, @杨钊, @一醉化千愁, @toskeyfine, @亡羊补牛, and the community.