共计 7231 个字符,预计需要花费 19 分钟才能阅读完成。
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 服务器的连贯、音讯公布和订阅。
版权申明:本文为 EMQ 原创,转载请注明出处。
原文链接:https://www.emqx.com/zh/blog/how-to-use-mqtt-in-java