MQTT — это стандартный протокол обмена сообщениями OASIS для Интернета вещей (IoT). Он разработан как чрезвычайно легкий транспорт для передачи сообщений по принципу «публикация/подписка», который идеально подходит для подключения удаленных устройств с небольшим кодовым отпечатком и минимальной пропускной способностью сети. Сегодня MQTT используется в самых разных отраслях, таких как автомобилестроение, производство, телекоммуникации, нефть и газ и т.д.
Эта статья рассказывает о том, как использовать MQTT в проекте Java для реализации функций подключения, подписки, отписки, публикации и получения сообщений между клиентом и брокером.
Добавление зависимостей
Среда разработки для этой статьи:
- Инструмент сборки: Maven
- IDE: IntelliJ IDEA
- Java: JDK 1.8.0
В качестве клиента мы будем использовать Eclipse Paho Java Client, который является наиболее широко используемой клиентской библиотекой MQTT на языке Java.
Добавьте следующие зависимости в файл pom.xml.
<dependencies>
<dependency>
<groupId>org.eclipse.paho</groupId>
<artifactId>org.eclipse.paho.client.mqttv3</artifactId>
<version>1.2.5</version>
</dependency>
</dependencies>
Создайте MQTT-соединение
Брокер MQTT
В этой статье будет использоваться публичный MQTT-брокер, созданный на базе EMQX Cloud. Информация о доступе к серверу выглядит следующим образом:
- Брокер: broker.emqx.io
- TCP-порт: 1883
- SSL/TLS порт: 8883
Подключение
Установите основные параметры подключения MQTT. Имя пользователя и пароль необязательны.
String broker = "tcp://broker.emqx.io:1883";
// TLS/SSL
// String broker = "ssl://broker.emqx.io:8883";
String username = "emqx";
String password = "public";
String clientid = "publish_client";
Затем создайте MQTT-клиент и подключитесь к брокеру.
MqttClient client = new MqttClient(broker, clientid, new MemoryPersistence());
MqttConnectOptions options = new MqttConnectOptions();
options.setUserName(username);
options.setPassword(password.toCharArray());
client.connect(options);
Инструкции
- MqttClient: MqttClient предоставляет набор методов, которые блокируют и возвращают управление прикладной программе после завершения действия MQTT.
- MqttClientPersistence: Представляет собой постоянное хранилище данных, используемое для хранения исходящих и входящих сообщений, пока они находятся в полете, обеспечивая доставку с заданным QoS.
- MqttConnectOptions: Содержит набор опций, которые управляют тем, как клиент подключается к серверу. Вот некоторые общие методы:
- setUserName: Устанавливает имя пользователя, которое будет использоваться для подключения.
- setPassword: устанавливает пароль для подключения.
- setCleanSession: Устанавливает, должны ли клиент и сервер запоминать состояние при перезагрузках и повторных подключениях.
- setKeepAliveInterval: Устанавливает интервал «поддержания жизни».
- setConnectionTimeout: Устанавливает значение таймаута соединения.
- setAutomaticReconnect: Устанавливает, будет ли клиент автоматически пытаться переподключиться к серверу в случае потери соединения.
Подключение с помощью TLS/SSL
Если вы хотите использовать самоподписанный сертификат для TLS/SSL соединений, добавьте bcpkix-jdk15on в файл pom.xml.
<!-- https://mvnrepository.com/artifact/org.bouncycastle/bcpkix-jdk15on -->
<dependency>
<groupId>org.bouncycastle</groupId>
<artifactId>bcpkix-jdk15on</artifactId>
<version>1.70</version>
</dependency>
Затем создайте файл SSLUtils.java со следующим кодом.
package io.emqx.mqtt;
import org.bouncycastle.jce.provider.BouncyCastleProvider;
import org.bouncycastle.openssl.PEMKeyPair;
import org.bouncycastle.openssl.PEMParser;
import org.bouncycastle.openssl.jcajce.JcaPEMKeyConverter;
import javax.net.ssl.KeyManagerFactory;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLSocketFactory;
import javax.net.ssl.TrustManagerFactory;
import java.io.BufferedInputStream;
import java.io.FileInputStream;
import java.io.FileReader;
import java.security.KeyPair;
import java.security.KeyStore;
import java.security.Security;
import java.security.cert.CertificateFactory;
import java.security.cert.X509Certificate;
public class SSLUtils {
public static SSLSocketFactory getSocketFactory(final String caCrtFile,
final String crtFile, final String keyFile, final String password)
throws Exception {
Security.addProvider(new BouncyCastleProvider());
// load CA certificate
X509Certificate caCert = null;
FileInputStream fis = new FileInputStream(caCrtFile);
BufferedInputStream bis = new BufferedInputStream(fis);
CertificateFactory cf = CertificateFactory.getInstance("X.509");
while (bis.available() > 0) {
caCert = (X509Certificate) cf.generateCertificate(bis);
}
// load client certificate
bis = new BufferedInputStream(new FileInputStream(crtFile));
X509Certificate cert = null;
while (bis.available() > 0) {
cert = (X509Certificate) cf.generateCertificate(bis);
}
// load client private key
PEMParser pemParser = new PEMParser(new FileReader(keyFile));
Object object = pemParser.readObject();
JcaPEMKeyConverter converter = new JcaPEMKeyConverter().setProvider("BC");
KeyPair key = converter.getKeyPair((PEMKeyPair) object);
pemParser.close();
// CA certificate is used to authenticate server
KeyStore caKs = KeyStore.getInstance(KeyStore.getDefaultType());
caKs.load(null, null);
caKs.setCertificateEntry("ca-certificate", caCert);
TrustManagerFactory tmf = TrustManagerFactory.getInstance("X509");
tmf.init(caKs);
// client key and certificates are sent to server so it can authenticate
KeyStore ks = KeyStore.getInstance(KeyStore.getDefaultType());
ks.load(null, null);
ks.setCertificateEntry("certificate", cert);
ks.setKeyEntry("private-key", key.getPrivate(), password.toCharArray(),
new java.security.cert.Certificate[]{cert});
KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory
.getDefaultAlgorithm());
kmf.init(ks, password.toCharArray());
// finally, create SSL socket factory
SSLContext context = SSLContext.getInstance("TLSv1.2");
context.init(kmf.getKeyManagers(), tmf.getTrustManagers(), null);
return context.getSocketFactory();
}
}
Установите options следующим образом.
String broker = "ssl://broker.emqx.io:8883";
// Set socket factory
String caFilePath = "/cacert.pem";
String clientCrtFilePath = "/client.pem";
String clientKeyFilePath = "/client.key";
SSLSocketFactory socketFactory = getSocketFactory(caFilePath, clientCrtFilePath, clientKeyFilePath, "");
options.setSocketFactory(socketFactory);
Публикация сообщений MQTT
Создайте класс PublishSample, который будет публиковать сообщение Hello MQTT в тему mqtt/test.
package io.emqx.mqtt;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
public class PublishSample {
public static void main(String[] args) {
String broker = "tcp://broker.emqx.io:1883";
String topic = "mqtt/test";
String username = "emqx";
String password = "public";
String clientid = "publish_client";
String content = "Hello MQTT";
int qos = 0;
try {
MqttClient client = new MqttClient(broker, clientid, new MemoryPersistence());
MqttConnectOptions options = new MqttConnectOptions();
options.setUserName(username);
options.setPassword(password.toCharArray());
options.setConnectionTimeout(60);
options.setKeepAliveInterval(60);
// connect
client.connect(options);
// create message and setup QoS
MqttMessage message = new MqttMessage(content.getBytes());
message.setQos(qos);
// publish message
client.publish(topic, message);
System.out.println("Message published");
System.out.println("topic: " + topic);
System.out.println("message content: " + content);
// disconnect
client.disconnect();
// close client
client.close();
} catch (MqttException e) {
throw new RuntimeException(e);
}
}
}
Подписка
Создайте класс SubscribeSample, который будет подписываться на тему mqtt/test.
package io.emqx.mqtt;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
public class SubscribeSample {
public static void main(String[] args) {
String broker = "tcp://broker.emqx.io:1883";
String topic = "mqtt/test";
String username = "emqx";
String password = "public";
String clientid = "subscribe_client";
int qos = 0;
try {
MqttClient client = new MqttClient(broker, clientid, new MemoryPersistence());
// connect options
MqttConnectOptions options = new MqttConnectOptions();
options.setUserName(username);
options.setPassword(password.toCharArray());
options.setConnectionTimeout(60);
options.setKeepAliveInterval(60);
// setup callback
client.setCallback(new MqttCallback() {
public void connectionLost(Throwable cause) {
System.out.println("connectionLost: " + cause.getMessage());
}
public void messageArrived(String topic, MqttMessage message) {
System.out.println("topic: " + topic);
System.out.println("Qos: " + message.getQos());
System.out.println("message content: " + new String(message.getPayload()));
}
public void deliveryComplete(IMqttDeliveryToken token) {
System.out.println("deliveryComplete---------" + token.isComplete());
}
});
client.connect(options);
client.subscribe(topic, qos);
} catch (Exception e) {
e.printStackTrace();
}
}
}
MqttCallback：
-
connectionLost(Throwable cause): Этот метод вызывается при потере соединения с сервером.
-
messageArrived(String topic, MqttMessage message): Этот метод вызывается, когда от сервера приходит сообщение.
-
deliveryComplete(IMqttDeliveryToken token): Вызывается, когда доставка сообщения завершена, и все подтверждения получены.
Тест
Далее выполните команду SubscribeSample, чтобы подписаться на тему mqtt/test. Затем выполните PublishSample для публикации сообщения в тему mqtt/test. Мы увидим, что издатель успешно опубликовал сообщение, а подписчик получил его.
Теперь мы закончили использование Paho Java Client в качестве MQTT клиента для подключения к публичному MQTT серверу и реализации публикации и подписки сообщений.
Полный код доступен по адресу: https://github.com/emqx/MQTT-Client-Examples/tree/master/mqtt-client-Java.
Первоначально опубликовано на https://www.emqx.com.