使用Java实现MQTT

1 年 ago
韵, 科
5 minutes
image.png

目录

    • 依存関係の追加

 

    • MQTT接続の作成

 

    • MQTTメッセージのパブリッシュ

 

    • 購読

 

    • テスト

 

    まとめ

MQTT是为物联网（IoT）设计的OASIS标准消息传输协议。MQTT被设计为一种非常轻量级的发布/订阅消息传输方式，适用于连接代码足迹小且网络带宽较低的远程设备。如今，MQTT在汽车、制造、通信、石油和天然气等各个行业得到广泛应用。

本文将介绍在Java项目中如何使用MQTT实现客户端与代理之间的连接、订阅、取消订阅、发布以及接收消息等功能。

添加依赖关系

这篇文章的开发环境是以下的:

    • ビルドツール: Maven

IDE: IntelliJ IDEA

Java: JDK 1.8.0

作为客户端，我们将使用Eclipse Paho Java Client这个在Java语言中最广泛使用的MQTT客户端库。

在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代理服务器

在本文中，我們將使用基於EMQX Cloud創建的免費公共MQTT代理。以下是服務器訪問信息：

    • ブローカー: http://broker.emqx.io

TCPポート: 1883

SSL/TLSポート: 8883

MQTT经纪人选定实践指南

下载这个实用指南，学习选择 MQTT 代理时要考虑的事项。

下载指南

连接

设置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连接中使用自签名证书，可以在pom.xml文件中添加bcpkix-jdk15on库。

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

将选项设定为以下内容。

String broker = "ssl://broker.emqx.io:8883";
// ソケットファクトリを設定
String caFilePath = "/cacert.pem";
String clientCrtFilePath = "/client.pem";
String clientKeyFilePath = "/client.key";
SSLSocketFactory socketFactory = getSocketFactory(caFilePath, clientCrtFilePath, clientKeyFilePath, "");
options.setSocketFactory(socketFactory);

发布MQTT消息

创建一个PublishSample类，向mqtt/test主题发布Hello MQTT消息。

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:

MQTT回调函数

    • connectionLost(Throwable cause): これはサーバーへの接続が失われたときに呼び出されるメソッドです。

 

    • messageArrived(String topic, MqttMessage message): これはサーバーからメッセージが到着したときに呼び出されるメソッドです。

 

    deliveryComplete(IMqttDeliveryToken token): メッセージの配信が完了し、すべての確認応答が受信されたときに呼び出されます。

考试 shì)

接下来，执行SubscribeSample，订阅mqtt/test主题。然后，执行PublishSample，发布一条消息到mqtt/test主题。您会看到发布者成功发布了消息，并且订阅者接收到了该消息。

image.png

总结

通过使用Paho Java Client作为MQTT客户端，我们成功地连接到公共MQTT服务器，并实现了消息的发布和订阅。

可在此处获取完整的代码：https://github.com/emqx/MQTT-Client-Examples/tree/master/mqtt-client-Java。

接下来，您可以查看EMQ提供的一系列易于理解的MQTT协议指南文章，通过学习MQTT协议的功能，进一步探索MQTT的高级应用，以及开始进行MQTT应用和服务开发。

相关资源：

    • WebSocketを使用したMQTTのクイックスタートガイド

 

    • ESP32でのMQTT: 初心者向けガイド

 

    [PythonでのPahoクライアントを使用したMQTTの使い方](
✖
广告
将在 10 秒后关闭
bannerAds