Official References

• Quick Deployment: https://docs.emqx.com/en/emqx/v5.7/getting-started/getting-started.html
• Development Guide: https://docs.emqx.com/en/emqx/latest/connect-emqx/java.html
• MQTTX Client Download: https://mqttx.app/downloads

Implemantation Details

Maven Dependency

<dependency>
    <groupId>org.eclipse.paho</groupId>
    <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
    <version>1.2.2</version>
</dependency>

Message Callback Handler

Implement the MqttCallback interface to process incoming messages and connection events.

import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import java.util.Base64;

@Slf4j
public class MqttMessageHandler implements MqttCallback {

    @Override
    public void connectionLost(Throwable cause) {
        log.warn("MQTT connection lost", cause);
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        log.info("Message received on topic [{}]", topic);
        byte[] payload = message.getPayload();
        String encoded = Base64.getEncoder().encodeToString(payload);
        log.info("Encoded payload: {}", encoded);
        // Insert custom processing logic here
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        log.debug("Delivery complete for token: {}", token.getMessageId());
    }
}

Configuration Constants

In a production environment these values should be externalised (e.g. through Nacos, Consul, or application properties). For simplicity, they are defined as constants.

public final class MqttConstants {

    private MqttConstants() {}

    public static final String BROKER_URL = "tcp://47.102.199.23:1883";
    public static final String CLIENT_ID = "TEST_MQ";
    public static final String DEFAULT_TOPIC = "test";
    public static final String USERNAME = "test";
    public static final String PASSWORD = "test";
    public static final int QOS_1 = 1;
}

MQTT Client Configuraton

This class initialises a singleton MqttClient bean using double‑checked locking and connects to the broker.

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

@Component
public class MqttClientConfiguration {

    private volatile MqttClient clientInstance;

    @Bean
    public MqttClient mqttClient() {
        if (clientInstance == null) {
            synchronized (MqttClientConfiguration.class) {
                if (clientInstance == null) {
                    try {
                        MemoryPersistence persistence = new MemoryPersistence();
                        MqttClient client = new MqttClient(
                                MqttConstants.BROKER_URL,
                                MqttConstants.CLIENT_ID,
                                persistence);
                        MqttConnectOptions options = new MqttConnectOptions();
                        options.setUserName(MqttConstants.USERNAME);
                        options.setPassword(MqttConstants.PASSWORD.toCharArray());
                        client.connect(options);
                        client.setCallback(new MqttMessageHandler());
                        client.subscribe(MqttConstants.DEFAULT_TOPIC);
                        clientInstance = client;
                    } catch (Exception e) {
                        throw new RuntimeException("Failed to initialize MQTT client", e);
                    }
                }
            }
        }
        return clientInstance;
    }
}

Messaging Service Facade

Exposes convenience methods for subscribing and publishing, suitable for use across services.

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;

@Service
public class MqttMessagingService {

    @Resource
    private MqttClient mqttClient;

    public void subscribe(String topicFilter) throws MqttException {
        mqttClient.subscribe(topicFilter);
    }

    public void publish(String topic, String payload) throws MqttException {
        MqttMessage message = new MqttMessage(payload.getBytes());
        mqttClient.publish(topic, message);
    }

    public void publish(String topic, String payload, int qos) throws MqttException {
        MqttMessage message = new MqttMessage(payload.getBytes());
        message.setQos(qos);
        mqttClient.publish(topic, message);
    }
}

MQTT Quality of Service (QoS)

For a thorough explanation see: https://www.emqx.com/en/blog/introduction-to-mqtt-qos

QoS 0 – At most once

QoS 0 provides no delivery guarantees. The message is sent once and not stored or retried. The receiver never receives duplicates. Message loss can occur if the TCP connection drops while data is still in transit or in socket buffers.

QoS 1 – At least once

To ensure delivery, QoS 1 uses acknowledgements (PUBACK). The sender stores the PUBLISH packet until it receives a PUBACK with the same Packet ID, then deletes the stored message. Duplicates can happen if the PUBACK is lost: the sender retransmits, and the receiver processes the same Packet ID again.

QoS 2 – Exactly once

QoS 2 guarantees no loss and no duplicates using a four‑step handshake (PUBLISH, PUBREC, PUBREL, PUBCOMP). The sender stores the PUBLISH and later the PUBREL until the coresponding acknowledgements arrive. The receiver holds the Packet ID until the PUBREL is received, preventing the same Packet ID from being reused for a different message. This ensures exactly‑once delivery at the cost of higher overhead.

A single topic can have multiple subscribers; any message published to that topic will be delivered to all subscribers. Clients can both publish and subscribe.