Besser als Fussball: MQTT Java Client mit Paho – Wenzlaff.de

Fussball? Wie kann man per Java einen MQTT Client mit Paho schreiben, der z.B. die Anzahl Flugzeuge in Hannover anzeigt? Und das innerhalb der ersten Halbzeit, da nur 2 Klassen benötigt werden.

Mein Raspberry Pi sendet an den öffentlichen Testserver test.mosquitto.org unter den Topic /Anzahl/Flugzeuge/Hannover alle 5 Minuten die aktuelle Anzahl, so das dieses Beispiel jeder nachvollziehen kann.

Hier die Architektur:

Wir brauchen zuerst ein Eclipse Maven Projekt mit Java 1.8 mit dieser pom.xml und einen Eintrag des Paho Repo, da es in Maven Central nicht vorhanden ist:

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
	<modelVersion>4.0.0</modelVersion>
	<groupId>de.wenzlaff.mqtt.client</groupId>
	<artifactId>de.wenzlaff.mqtt.client</artifactId>
	<version>0.0.1-SNAPSHOT</version>

	<repositories>
		<repository>
			<id>Eclipse Paho Repo</id>
			<url>https://repo.eclipse.org/content/repositories/paho-releases/</url>
		</repository>
	</repositories>

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

		<dependency>
			<groupId>org.slf4j</groupId>
			<artifactId>slf4j-api</artifactId>
			<version>1.7.13</version>
		</dependency>
		<dependency>
			<groupId>org.slf4j</groupId>
			<artifactId>slf4j-log4j12</artifactId>
			<version>1.7.13</version>
		</dependency>
	</dependencies>

</project>

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

<groupId>de.wenzlaff.mqtt.client</groupId>

<artifactId>de.wenzlaff.mqtt.client</artifactId>

<version>0.0.1-SNAPSHOT</version>

<id>Eclipse Paho Repo</id>

<url>https://repo.eclipse.org/content/repositories/paho-releases/</url>

</repository>

</repositories>

<groupId>org.eclipse.paho</groupId>

<artifactId>org.eclipse.paho.client.mqttv3</artifactId>

</dependency>

</dependency>

</dependency>

</dependencies>

</project>

Dann eine FlugClient.java Klasse:

package de.wenzlaff.mqtt.client;

import java.util.UUID;

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * MQTT Flug Client.
 * 
 * @author Thomas Wenzlaff
 *
 */
public class FlugClient {

	private static final Logger LOG = LoggerFactory.getLogger(FlugClient.class);

	/** Die URI des Servers der abgefragt wird. */
	private static final String SERVER_URI = "tcp://test.mosquitto.org";

	/** Der Topic auf den die Flugdaten ankommen. */
	private static final String FLUG_TOPIC = "/Anzahl/Flugzeuge/Hannover";

	public static void main(String[] args) throws Exception {

		startClientEndlos();
	}

	private static void startClientEndlos() throws MqttException {

		LOG.info("Starte MQTT Flug Client ...");		

		MqttClient client = new MqttClient(SERVER_URI, generateClientId());

		client.setCallback(new LogCallback());

		client.connect();

		client.subscribe(FLUG_TOPIC);
	}

	private static String generateClientId() {
		return UUID.randomUUID().toString();
	}
}

package de.wenzlaff.mqtt.client;

import java.util.UUID;

import org.eclipse.paho.client.mqttv3.MqttClient;

import org.eclipse.paho.client.mqttv3.MqttException;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

/**

* MQTT Flug Client.

* @author Thomas Wenzlaff

public class FlugClient {

private static final Logger LOG = LoggerFactory.getLogger(FlugClient.class);

/** Die URI des Servers der abgefragt wird. */

private static final String SERVER_URI = "tcp://test.mosquitto.org";

/** Der Topic auf den die Flugdaten ankommen. */

private static final String FLUG_TOPIC = "/Anzahl/Flugzeuge/Hannover";

public static void main(String[] args) throws Exception {

startClientEndlos();

}

private static void startClientEndlos() throws MqttException {

LOG.info("Starte MQTT Flug Client ...");

MqttClient client = new MqttClient(SERVER_URI, generateClientId());

client.setCallback(new LogCallback());

client.connect();

client.subscribe(FLUG_TOPIC);

}

private static String generateClientId() {

return UUID.randomUUID().toString();

}

Dann noch eine MQTT Callback Klasse mit Namen LogCallback.java. Hier ein einfaches Beispiel das nur per Log4j auf der Konsole loggt:

package de.wenzlaff.mqtt.client;

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Callback Klasse die nur die Nachrichten loggt.
 * 
 * @author Thomas Wenzlaff
 *
 */
class LogCallback implements MqttCallback {

	private static final Logger LOG = LoggerFactory.getLogger(LogCallback.class);

	public void connectionLost(Throwable throwable) {
	}

	public void messageArrived(String t, MqttMessage m) throws Exception {
		String nachricht = m.toString().replaceAll("#", "");

		LOG.info("{}", nachricht);
	}

	public void deliveryComplete(IMqttDeliveryToken t) {
	}

}

package de.wenzlaff.mqtt.client;

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;

import org.eclipse.paho.client.mqttv3.MqttCallback;

import org.eclipse.paho.client.mqttv3.MqttMessage;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

/**

* Callback Klasse die nur die Nachrichten loggt.

* @author Thomas Wenzlaff

class LogCallback implements MqttCallback {

private static final Logger LOG = LoggerFactory.getLogger(LogCallback.class);

public void connectionLost(Throwable throwable) {

}

public void messageArrived(String t, MqttMessage m) throws Exception {

String nachricht = m.toString().replaceAll("#", "");

LOG.info("{}", nachricht);

}

public void deliveryComplete(IMqttDeliveryToken t) {

}

Es kann aber auch leicht erweitert werden, einfach das Interface MqttCallback in einer Callback Klasse implementieren. Die ankommenden Meldungen werden in der messageArrived Methode behandelt.

Wenn wir nun die FlugClient Klasse starten, wird auf der Konsole alle 5 Minuten die Anzahl Flugzeuge in Hannover in einem eigenen Thread ausgegeben:

INFO  FlugClient - Starte MQTT Flug Client ...
INFO  LogCallback - 12 Flugzeuge im Umkreis Flughafen Hannover sichtbar via ADSB am Thu Jul 07 2016 19:57:13 GMT+0200 (CEST) mit Raspberry Pi
INFO  LogCallback - 13 Flugzeuge im Umkreis Flughafen Hannover sichtbar via ADSB am Thu Jul 07 2016 20:02:13 GMT+0200 (CEST) mit Raspberry Pi
INFO  LogCallback - 15 Flugzeuge im Umkreis Flughafen Hannover sichtbar via ADSB am Thu Jul 07 2016 20:07:13 GMT+0200 (CEST) mit Raspberry Pi
INFO  LogCallback - 14 Flugzeuge im Umkreis Flughafen Hannover sichtbar via ADSB am Thu Jul 07 2016 20:12:13 GMT+0200 (CEST) mit Raspberry Pi
...

INFO FlugClient - Starte MQTT Flug Client ...

INFO LogCallback - 12 Flugzeuge im Umkreis Flughafen Hannover sichtbar via ADSB am Thu Jul 07 2016 19:57:13 GMT+0200 (CEST) mit Raspberry Pi

INFO LogCallback - 13 Flugzeuge im Umkreis Flughafen Hannover sichtbar via ADSB am Thu Jul 07 2016 20:02:13 GMT+0200 (CEST) mit Raspberry Pi

INFO LogCallback - 15 Flugzeuge im Umkreis Flughafen Hannover sichtbar via ADSB am Thu Jul 07 2016 20:07:13 GMT+0200 (CEST) mit Raspberry Pi

INFO LogCallback - 14 Flugzeuge im Umkreis Flughafen Hannover sichtbar via ADSB am Thu Jul 07 2016 20:12:13 GMT+0200 (CEST) mit Raspberry Pi

...

Da ich beim Senden der Anzahl die retained messages (aufbewahrte Nachrichten) auf true gesetzt habe, kommt gleich beim Start des Client die letzte Meldung.
D.h. das sind Nachrichten mit gesetztem Retain-Flag. Meldet sich ein neuer Teilnehmer am Topic an, erhält er als Erstes die letzte Nachricht, die mit diesem Flag im Topic eingetroffen ist. Das gilt auch bei Abonnements mit Wildcards, die das betreffende Topic enthalten.

Die Flugdaten werden übrigens von einem eigenen Raspberry Pi mit node-RED von IBM an den Testserver, Twitter, Debug-Konsole und ThinkSpeak gesendet, hier der Flow: