MQTT 是一种基于公布/订阅模式的 轻量级物联网音讯传输协定 ，可在重大受限的硬件设施和低带宽、高提早的网络上实现稳固传输。它凭借简略易实现、反对 QoS、报文小等特点，占据了物联网协定的半壁江山。

本文次要介绍如何在 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 提供的 收费公共 MQTT 服务器，该服务基于 EMQX 的 MQTT 云平台 创立。服务器接入信息如下：

  • Broker: broker.emqx.io（中国用户能够应用 broker-cn.emqx.io）
  • TCP Port: 1883
  • SSL/TLS Port: 8883

一般 TCP 连贯

设置 MQTT Broker 根本连贯参数，用户名、明码为非必选参数。

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: 同步调用客户端，应用阻塞办法通信。
  • 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。

// 设置 SSL/TLS 连贯地址String broker = "ssl://broker.emqx.io:8883";// 设置 socket factoryString 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);           // 连贯           client.connect(options);           // 创立音讯并设置 QoS           MqttMessage message = new MqttMessage(content.getBytes());           message.setQos(qos);           // 公布音讯           client.publish(topic, message);           System.out.println("Message published");           System.out.println("topic: " + topic);           System.out.println("message content: " + content);           // 敞开连贯           client.disconnect();           // 敞开客户端           client.close();      } catch (MqttException e) {           throw new RuntimeException(e);      }  }}

订阅 MQTT 主题

创立一个订阅客户端类 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());           // 连贯参数           MqttConnectOptions options = new MqttConnectOptions();           options.setUserName(username);           options.setPassword(password.toCharArray());           options.setConnectionTimeout(60);      options.setKeepAliveInterval(60);           // 设置回调           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 主题。 咱们将会看到公布端胜利公布音讯，同时订阅端接管到音讯。

至此，咱们实现了在 Java 中应用 Paho Java Client 来作为 MQTT 客户端连贯到 公共 MQTT 服务器，并实现了测试客户端与 MQTT 服务器的连贯、音讯公布和订阅。

版权申明： 本文为 EMQ 原创，转载请注明出处。

原文链接：https://www.emqx.com/zh/blog/how-to-use-mqtt-in-java