The most surprising thing about MQTT is that it’s a publish-subscribe protocol, not a point-to-point one, meaning your client doesn’t know or care who receives its messages; it just broadcasts them to a central broker.
Let’s see this in action. We’ll set up a simple Java Paho client to connect to a public MQTT broker and publish a message.
First, you need the Paho Java client library. Add this to your pom.xml if you’re using Maven:
<dependency>
<groupId>org.eclipse.paho</groupId>
<artifactId>org.eclipse.paho.client.mqttv3</artifactId>
<version>1.2.5</version>
</dependency>
Now, here’s the Java code. We’ll connect to tcp://mqtt.eclipseprojects.io:1883, which is a free, public broker.
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
public class MqttPublisher {
public static void main(String[] args) {
String broker = "tcp://mqtt.eclipseprojects.io:1883";
String clientId = MqttClient.generateClientId();
String topic = "my/test/topic";
String messageString = "Hello from Paho Java Client!";
int qos = 0; // Quality of Service level
boolean retained = false; // Retained message flag
try {
// Create a MemoryPersistence object to store messages temporarily
MemoryPersistence persistence = new MemoryPersistence();
// Create an MqttClient instance
MqttClient client = new MqttClient(broker, clientId, persistence);
// Set connection options
MqttConnectOptions connOpts = new MqttConnectOptions();
connOpts.setCleanSession(true); // Start with a clean session
connOpts.setConnectionTimeout(60); // Connection timeout in seconds
connOpts.setKeepAliveInterval(60); // Keep alive interval in seconds
// Connect to the broker
System.out.println("Connecting to broker: " + broker);
client.connect(connOpts);
System.out.println("Connected");
// Create the message payload
MqttMessage message = new MqttMessage(messageString.getBytes());
message.setQos(qos);
message.setRetained(retained);
// Publish the message
System.out.println("Publishing message: " + messageString);
System.out.println("Topic: " + topic);
MqttDeliveryToken token = client.publish(topic, message);
// Wait for the message to be delivered to the broker
token.waitForCompletion();
System.out.println("Message delivered");
// Disconnect from the broker
client.disconnect();
System.out.println("Disconnected");
} catch (MqttException me) {
System.out.println("reason " + me.getReasonCode());
System.out.println("msg " + me.getMessage());
System.out.println("loc " + me.getLocalizedMessage());
System.out.println("cause " + me.getCause());
System.out.println("excep " + me);
me.printStackTrace();
}
}
}
This code does a few key things:
- Client ID: It generates a unique client ID. This is crucial for the broker to distinguish between different clients. If you use the same client ID from multiple places simultaneously, the broker will typically disconnect the older connection.
- Broker Address: It specifies the broker’s address, including the protocol (
tcp://) and port (1883is the default for unencrypted MQTT). - Persistence:
MemoryPersistencemeans messages are held in RAM. For critical applications, you’d useMqttDefaultFilePersistenceto save messages to disk, ensuring they aren’t lost if the client crashes. - Connection Options:
MqttConnectOptionsallows you to configure things likecleanSession. SettingcleanSessiontotruemeans the broker will discard any session information for this client upon disconnection. Iffalse, the broker will store subscriptions and QoS 1/2 messages for the client, resuming the session upon reconnection.connectionTimeoutandkeepAliveIntervalare important for network stability. - Publishing: You create an
MqttMessageobject, set its payload (the actual data as bytes), and its Quality of Service (QoS). QoS 0 is "at most once" delivery (fire and forget), QoS 1 is "at least once" (guarantees delivery, but might arrive multiple times), and QoS 2 is "exactly once" (guarantees delivery exactly once, but is more complex and slower).retainedis a flag for the broker: if true, the broker will store the last message published to this topic and send it to any new subscriber that joins after the message was published.
The core of the MQTT mental model revolves around three components: the Client, the Broker, and the Topic. Clients don’t talk directly to each other. Instead, they send messages to the broker, which then filters and distributes them to other clients that have subscribed to specific topics. Topics are hierarchical strings, like file paths, e.g., building/floor/room/sensor. A client can publish to a topic and subscribe to one or more topics.
When a client publishes a message, it sends it to the broker along with the topic name. The broker then looks at all its connected clients and checks their subscriptions. If a client has subscribed to a topic that matches the published topic (or a wildcard pattern), the broker forwards the message to that client.
The token.waitForCompletion() is important. For QoS 1 and 2 messages, publishing is an asynchronous operation. The publish method returns a MqttDeliveryToken immediately, but the message might not have actually reached the broker yet. waitForCompletion() blocks until the broker acknowledges receipt of the message (for QoS 1) or the entire QoS 2 handshake is complete.
A common point of confusion is topic wildcard usage. Subscriptions can use wildcards: + matches any single level, and # matches any number of levels from that point down. For example, subscribing to sensors/+/temperature would receive messages from sensors/room1/temperature and sensors/room2/temperature, but not sensors/hallway/temperature. Subscribing to sensors/# would receive messages from sensors/room1/temperature, sensors/hallway/humidity, and anything else under the sensors parent topic.
The final step is client.disconnect(). This cleanly closes the connection to the broker. If cleanSession was true, all temporary subscriptions and message queues associated with this client are removed from the broker.
After successfully publishing messages, the next logical step is to handle incoming messages, which involves setting up a MqttCallback to process messages that are subscribed to.