本文目录
- 1. 1. MQTT简介
- 2. 2. 选择Java作为MQTT客户端的优势
- 3. 3. 开始之前
- 4. 4. 选择合适的库：Eclipse Paho
- 5. 5. 构建MQTT客户端
- 6. 6. 强化安全性：SSL/TLS连接
- 7. TLS/SSL 连接
- 8. 发布 MQTT 消息
- 9. 订阅 MQTT 主题
- 10. 测试
- 11. 8. 总结
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客户端，并对其进行扩展以满足更复杂的需求。
希望这篇文章为您提供了宝贵的知识和实践指南！
