如何在 Java 中使用 MQTT

Posted 2023-04-07 EMQX

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

至此，我们完成了在 Java 中使用 Paho Java Client 来作为 MQTT 客户端连接到 公共 MQTT 服务器，并实现了测试客户端与 MQTT 服务器的连接、消息发布和订阅。

完整代码请见：https://github.com/emqx/MQTT-Client-Examples/tree/master/mqtt-client-Java。

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

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

如何在 Flask 项目中使用 MQTT

​​Flask​​ 是一个使用 Python 编写的轻量级 Web 应用框架，其被称为 “微框架”，因为它使用简单的核心，用扩展增加其他功能，例如：ORM、窗体验证工具、文件上传、各种开放式身份验证技术等。

​​MQTT​​ 是一种基于发布/订阅模式的 轻量级物联网消息传输协议 ，可以用极少的代码和带宽为联网设备提供实时可靠的消息服务，它广泛应用于物联网、移动互联网、智能硬件、车联网、电力能源等行业。

本文主要介绍如何在 Flask 项目中实现 ​​MQTT 客户端​​的连接、订阅、取消订阅、收发消息等功能。

我们将使用到 ​​Flask-MQTT​​ 的装饰器，用于简化 Flask 应用程序中的 MQTT 集成。

项目初始化

本项目使用 Python 3.8 进行开发测试，读者可用如下命令确认 Python 的版本。

$ python3 --version
Python 3.8.2

使用 Pip 安装 Flask-MQTT 库

pip3 install flask-mqtt

Flask-MQTT 使用

本文将使用 EMQ 提供的​​免费公共 MQTT 服务器​​ 创建。服务器接入信息如下：

• Broker:​​broker.emqx.io​​
• TCP Port:​​1883​​
• Websocket Port:​​8083​​

导入 Flask-MQTT

导入 Flask 库以及 Flask-MQTT 扩展，并创建 Flask 应用

from flask import Flask, request, jsonify
from flask_mqtt import Mqtt

app = Flask(__name__)

配置 Flask-MQTT 扩展

app.config[MQTT_BROKER_URL] = broker.emqx.io
app.config[MQTT_BROKER_PORT] = 1883
app.config[MQTT_USERNAME] =   # 当你需要验证用户名和密码时，请设置该项
app.config[MQTT_PASSWORD] =   # 当你需要验证用户名和密码时，请设置该项
app.config[MQTT_KEEPALIVE] = 5  # 设置心跳时间，单位为秒
app.config[MQTT_TLS_ENABLED] = False  # 如果你的服务器支持 TLS，请设置为 True
topic = /flask/mqtt

mqtt_client = Mqtt(app)

完整的配置项可以参考 ​​Flask-MQTT 配置文档​​。

编写连接回调函数

可以在该回调函数中对 MQTT 连接成功或失败的情况进行处理，本示例将在连接成功后订阅 ​​/flask/mqtt​​ 主题。

@mqtt_client.on_connect()
def handle_connect(client, userdata, flags, rc):
   if rc == 0:
       print(Connected successfully)
       mqtt_client.subscribe(topic) # 订阅主题
   else:
       print(Bad connection. Code:, rc)

编写消息回调函数

该函数将打印 ​​/flask/mqtt​​ 主题接收到的消息。

@mqtt_client.on_message()
def handle_mqtt_message(client, userdata, message):
   data = dict(
       topic=message.topic,
       payload=message.payload.decode()
  )
   print(Received message on topic: topic with payload: payload.format(**data))

创建发布消息接口

我们创建一个简单的 POST 接口实现 MQTT 消息发布。

在实际应用中该接口可能需要进行一些更复杂的业务逻辑处理。

@app.route(/publish, methods=[POST])
def publish_message():
   request_data = request.get_json()
   publish_result = mqtt_client.publish(request_data[topic], request_data[msg])
   return jsonify(code: publish_result[0])

运行 Flask 应用

if __name__ == __main__:
   app.run(host=127.0.0.1, port=5000)

当 Flask 应用启动后，MQTT 客户端将会连接到服务器，并且订阅主题 ​​/flask/mqtt​​。

测试

接下来我们使用 ​​MQTT 客户端 - MQTT X​​ 进行连接、订阅、发布测试。

测试消息接收

1. 在 MQTT X 中创建链接并连接到服务器。

2. 在 MQTT X 中向​​/flask/mqtt​​​ 主题发布消息​​Hello from MQTT X​​。

3. 在 Flask 运行窗口中将能看到 MQTT X 发送的消息。

测试消息发布接口

1. 在 MQTT X 中订阅​​/flask/mqtt​​ 主题。

2. 使用 Postman 调用​​/publish​​​ 接口：发送消息​​Hello from Flask​​​ 至​​/flask/mqtt​​ 主题。

3. 在 MQTT X 中将能看到 Flask 发送过来的消息。

完整代码

from flask import Flask, request, jsonify
from flask_mqtt import Mqtt

app = Flask(__name__)

app.config[MQTT_BROKER_URL] = broker.emqx.io
app.config[MQTT_BROKER_PORT] = 1883
app.config[MQTT_USERNAME] =   # 当你需要验证用户名和密码时，请设置该项
app.config[MQTT_PASSWORD] =   # 当你需要验证用户名和密码时，请设置该项
app.config[MQTT_KEEPALIVE] = 5  # 设置心跳时间，单位为秒
app.config[MQTT_TLS_ENABLED] = False  # 如果你的服务器支持 TLS，请设置为 True
topic = /flask/mqtt

mqtt_client = Mqtt(app)


@mqtt_client.on_connect()
def handle_connect(client, userdata, flags, rc):
   if rc == 0:
       print(Connected successfully)
       mqtt_client.subscribe(topic)
   else:
       print(Bad connection. Code:, rc)


@mqtt_client.on_message()
def handle_mqtt_message(client, userdata, message):
   data = dict(
       topic=message.topic,
       payload=message.payload.decode()
  )
   print(Received message on topic: topic with payload: payload.format(**data))


@app.route(/publish, methods=[POST])
def publish_message():
   request_data = request.get_json()
   publish_result = mqtt_client.publish(request_data[topic], request_data[msg])
   return jsonify(code: publish_result[0])

if __name__ == __main__:
   app.run(host=127.0.0.1, port=5000)

注意事项

Flask-MQTT 目前不适合使用多个工作实例，如果您需要使用 *gevent* 或 *gunicorn* 这样的 WSGI 服务器，请确保只有一个工作实例。

总结

至此，我们使用 Flask-MQTT 完成了简单