Как использовать MQTT в Java

MQTT — это стандартный протокол обмена сообщениями OASIS для Интернета вещей (IoT). Он разработан как чрезвычайно легкий транспорт обмена сообщениями публикации/подписки, который идеально подходит для подключения удаленных устройств с небольшим объемом кода и минимальной пропускной способностью сети. MQTT сегодня используется в самых разных отраслях, таких как автомобилестроение, производство, телекоммуникации, нефть и газ и т. д.

В этой статье рассказывается, как использовать MQTT в проекте Java для реализации функций подключения, подписки, отмены подписки, публикации и получения сообщений между клиентом и брокером.

Добавить зависимость

Среда разработки для этой статьи:

Мы будем использовать Java-клиент Eclipse Paho в качестве клиента, который является наиболее широко используемой клиентской библиотекой 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. Информация о доступе к серверу выглядит следующим образом:

  • Маклер: 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: представляет постоянное хранилище данных, используемое для хранения исходящих и входящих сообщений, пока они находятся в пути, обеспечивая доставку с указанным качеством обслуживания.
  • MqttConnectOptions: содержит набор параметров, управляющих подключением клиента к серверу. Вот несколько распространенных методов:
    • setUserName: устанавливает имя пользователя, которое будет использоваться для соединения.
    • setPassword: устанавливает пароль для подключения.
    • setCleanSession: устанавливает, должны ли клиент и сервер запоминать состояние при перезапуске и повторном подключении.
    • setKeepAliveInterval: устанавливает интервал «поддержания активности».
    • setConnectionTimeout: устанавливает значение времени ожидания соединения.
    • setAutomaticReconnect: устанавливает, будет ли клиент автоматически пытаться повторно подключиться к серверу, если соединение потеряно.

Подключение с помощью TLS/SSL

Если вы хотите использовать самозаверяющий сертификат для соединений TLS/SSL, добавьте bcpkix-jdk15on к pom.xml файл.

<!--  -->
<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();
      }
  }
}

Обратный вызов Mqtt:

connectionLost (причина Throwable): этот метод вызывается при потере соединения с сервером.

messageArrived (раздел строки, сообщение MqttMessage): этот метод вызывается, когда с сервера приходит сообщение.

deliveryComplete(IMqttDeliveryToken token): вызывается после завершения доставки сообщения и получения всех подтверждений.

Тест

Далее запустите SubscribeSample подписаться на mqtt/test тема. Затем запустите PublishSample опубликовать сообщение в mqtt/test тема. Мы увидим, что издатель успешно публикует сообщение, а подписчик его получает.

Java MQTT

Теперь мы закончили использовать Paho Java Client в качестве клиента MQTT для подключения к публичный MQTT-сервер и реализовать публикацию сообщений и подписку.

Полный код доступен по адресу: https://github.com/emqx/MQTT-Client-Examples/tree/master/mqtt-client-Java.

Первоначально опубликовано по адресу:https://www.emqx.com.

Похожие записи

Добавить комментарий

Ваш адрес email не будет опубликован. Обязательные поля помечены *