Java 中如何使用 MQTT 进行物联网消息传输

Java中的MQTT：深入实践与探索指南

随着物联网的快速发展，需要一个轻量级、高效和可靠的消息传递机制，而MQTT（Message Queuing Telemetry Transport）正好满足了这些需求。这篇文章将为您解开在Java中使用MQTT的奥秘，确保您能够快速入门并深入实践。

1. MQTT简介

MQTT是一种发布/订阅模式的消息传输协议，尤其适合低功耗设备和不稳定网络环境。它可以轻松处理设备间的消息通信，使物联网应用更为简洁。

2. 选择Java作为MQTT客户端的优势

Java是一个跨平台、稳定和安全的编程语言。其丰富的库和社区资源使得开发MQTT应用变得相对简单。

3. 开始之前

确保您的系统上已经安装了JDK 1.8或更高版本，并且配置了合适的Java环境。

4. 选择合适的库：Eclipse Paho

Eclipse Paho Java Client是一个非常受欢迎的库，支持MQTT协议。为了在你的Maven项目中使用它，请添加以下依赖：

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

5. 构建MQTT客户端

连接到MQTT broker

首先，我们需要连接到一个MQTT broker。为了简化，这里使用一个公开的broker：broker.emqx.io。

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: 设置是否自动重连

6. 强化安全性：SSL/TLS连接

保证数据安全是非常关键的，建议使用SSL/TLS来连接到MQTT broker。要实现这一点，你需要确保broker支持SSL，并在客户端配置适当的证书和加密设置。

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 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);
           // 连接
           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 主题。我们将会看到发布端成功发布消息，同时订阅端接收到消息。

8. 总结

MQTT在Java中的实现变得更为简单，感谢丰富的库和工具。通过本指南，您应该能够构建一个基本的MQTT客户端，并对其进行扩展以满足更复杂的需求。

希望这篇文章为您提供了宝贵的知识和实践指南！

打赏赞(2)

Java 中如何使用 MQTT 进行物联网消息传输

1. MQTT简介

2. 选择Java作为MQTT客户端的优势

3. 开始之前

4. 选择合适的库：Eclipse Paho

5. 构建MQTT客户端

连接到MQTT broker

6. 强化安全性：SSL/TLS连接

TLS/SSL 连接

发布 MQTT 消息

订阅 MQTT 主题

测试

8. 总结

HiveMQ将 Modbus 数据转换为 MQTT 消息

Golang与MQTT：轻松构建物联网应用

Java 中如何使用 MQTT 进行物联网消息传输

1. MQTT简介

2. 选择Java作为MQTT客户端的优势

3. 开始之前

4. 选择合适的库：Eclipse Paho

5. 构建MQTT客户端

连接到MQTT broker

6. 强化安全性：SSL/TLS连接

TLS/SSL 连接

发布 MQTT 消息

订阅 MQTT 主题

测试

8. 总结

HiveMQ将 Modbus 数据转换为 MQTT 消息

Golang与MQTT：轻松构建物联网应用

猜你喜欢

改善基础设施：HiveMQ如何推动智能城市发展

为什么企业选择全托管HiveMQ云进行MQTT部署

MQTT 赋能工业 PLC 数据采集与应用