Java MQTTクライアントの設定

お問い合わせ
クライアント.disconnect();
System.out.println(「ブローカーから切断」);
お問い合わせ お問い合わせ
System.out.println(「ブローカーから切断するエラー:「+ e.getMessage()」);」);
e.printStackTrace() ;
お問い合わせ

MQTT 3.1.1の特長

//パホ mqttv3
インポートorg.eclipse.paho.client.mqttv3.*;
インポートorg.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
javax.net.sl.SSLSocketFactoryをインポートします。
javax.net.ssl.SSLContext をインポートします。
java.security.cert.Certificate をインポート 工場;
java.security.cert.X509Certificate をインポートします。
java.security.KeyStore をインポートします。
java.io をインポートします。 ファイル入力ストリーム;
java.io.IOException のインポート
java.security.cert.CertificateException のインポート
java.security.NoSuchAlgorithmException をインポートします。
インポート javax.net.sl.TrustManager 工場;
java.security.KeyManagementException のインポート
java.util をインポートします。 ランダム;
java.util をインポートします。 スキャナ;
パブリッククラス MqttClientv3 お問い合わせ
パブリック static void main(String[] args) {
文字列 client_id = "your_client_id";
文字列 hostip = "public-mqtt-broker.bevywise.com";
boolean TLS = false; // TLS を有効にするには true を設定
boolean AUTH = true;
文字列 client_username = "Access-Key";
文字列 client_password = "アクセストークン";
boolean websocket_connection=false; //websocketを有効にするためにtrueを設定
文字列のトピック = "";
int qos = 0;
boolean は = false を保持します。
ストリングブローカーUrl= "";
intの港;
もし(websocket_connection)
ポート= TLS ? 11433:10433;
その他
ポート = TLS ? 8883 : 1883; // SSL/TLS で 8883 を使用

// 新しいMQTTクライアントインスタンスを作成する
MqttClientクライアント
もし(websocket_connection)
BrokerUrl=String.format("%s://%s:%d", TLS ? "ws" : "ws", hostip, port);
その他
ブローカー Url = 文字列.format("%s://%s:%d", TLS ? "ssl" : "tcp", hostip, port);

MemoryPersistence の永続性 = new MemoryPersistence();
お問い合わせ
クライアント = 新しいMqttClient(ブローカー) Url、client_id、永続性;
// コールバックの設定
client.setCallback(新しいMqttCallback() {
@オーバーライド
パブリック void 接続 失われた(恐ろしい原因) {
System.out.println("Connection Lost:" + 原因.getMessage()); );
原因。printStackTrace();
お問い合わせ
@オーバーライド
パブリック void メッセージ 到着(文字列トピック、MqttMessageメッセージ)は例外を投げます お問い合わせ
System.out.println(「トピックに関するメッセージを受け取る」+トピック);
System.out.println("メッセージ:" + new String(message.getPayload()));
お問い合わせ
@オーバーライド
公共 void 配達 完全な(IMqttDeliveryTokenトークン) {
System.out.println(「メッセージが正常に配信されました」)。
お問い合わせ
お問い合わせ
// 接続オプションの設定
MqttConnectOptions connOpts = 新しいMqttConnectOptions();
お問い合わせ
お問い合わせ
connOpts.setSocketFactory(getSocketFactory("/path/to/your/root.crt")); // // // //
ルートの証明書パスに置き換える
お問い合わせ
System.out.println(「SSLソケット工場の設定エラー」+ e.getMessage() );
e.printStackTrace() ;
お問い合わせ
お問い合わせ
お問い合わせ
connOpts.setUserName(client_username) );
connOpts.setPassword(client_password.toCharArray())) );
お問い合わせ
connOpts.setCleanSession(true);
// MQTTブローカーに接続する
System.out.println(「ブローカーに接続する:「+ブローカーUrl」);
client.connect(connOpts);
System.out.println(「コネクテッド」);
// 購読するトピックの例
最後の文字列の購読 = "will_topic/device1";
// トピックへのサブスクライブ
client.subscribe(subscribeTopic)。
// ユーザの入力に基づいてメッセージやサブスクライブ/サブスクライブを投稿
スキャナスキャナー = 新しいスキャナ(System.in)。
時 (true) {
System.out.println("1. Publish\n2.subscribe\n3. Disconnect\n4. 購読解除
int 選択 = スキャナ.nextInt();
Scanner.nextLine(); // 新規ラインを消費する
スイッチ (choice) {
ケース1:
System.out.print("Enter Topic QoS(0/1/2) 保持(0/1) カウント: ");
文字列[] 入力 = スキャナ.nextLine().split(");
トピック = 入力[0];
qos = Integer.parseInt(input[1]); ;
保持 = Integer.parseInt(input[2]) === 1;
int カウント = Integer.parseInt(input[3]);;
(int i = 0; i < count; i++) { 文字列 message="{\"temp\":" + (new)
Random().nextInt(50) + 1) + ",\" 圧力\":" + (新しいRandom().nextInt(100) + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
1 + ",\" ステータス: + (new Random().nextInt(2)) + "}" ; ; ; ; ; ; ; ; ; ; ; ; ;
client.publish(topic, message.getBytes(), qos, 保持);
System.out.println("公開メッセージ:" + メッセージ);
お問い合わせ お問い合わせ
System.out.println(「システムアウト」) エラー公開メッセージ: " + e.getMessage());
e.printStackTrace() ;
お問い合わせ
Thread.sleep(6000); // 6秒
お問い合わせ
ブレイク;
ケース2:
System.out.print(" トピックQoS(0/1/2): ");
文字列[] subInput = Scanner.nextLine().split("););
トピック = subInput[0];
qos = Integer.parseInt(subInput[1]); ;
お問い合わせ
client.subscribe(トピック、qos);
System.out.println(「システムアウト」) トピック: " + トピック;
お問い合わせ お問い合わせ
System.out.println(「システムアウト」) トピックにサブスクライブエラー: " + e.getMessage());
e.printStackTrace() ;
お問い合わせ
ブレイク;
場合3:
お問い合わせ
クライアント.disconnect();
System.out.println(「システムアウト」) ブローカーから切断」。 お問い合わせ { System.out.println("エラー切断")
ブローカーから: " + e.getMessage());
e.printStackTrace() ;
お問い合わせ
ブレイク;
場合4:
System.out.print(「システムアウト」) 退会するトピックを入力してください。
トピック = Scanner.nextLine();
お問い合わせ
client.unsubscribe(トピック);
System.out.println(「システムアウト」) トピックから退会: " + トピック;
お問い合わせ お問い合わせ
System.out.println(「システムアウト」) トピックから購読しないエラー: " + e.getMessage() );
e.printStackTrace() ;
お問い合わせ
ブレイク;
デフォルト:
お問い合わせ _ Invalid Choice お問い合わせ System.out.println("An"An") は、
エラーが発生しました: " + e.getMessage());
e.printStackTrace() ;
お問い合わせ
お問い合わせ
// // // // SSLソケット工場(必要に応じて)を作成する方法
プライベート静的SSLSocket 工場getSocket 工場(String caCrtFile) スロー 例外 {
証明書工場 cf = CertificateFactory.getInstance(" X.509"); ファイル入力ストリーム caInput=new
FileInputStream(caCrtFile)、X509Certificate caCert=(X509Certificate)
cf.generateCertificate(caInput); caInput.close(); キーストア
keyStore=KeyStore.getInstance(KeyStore.getDefaultType()); );
キーストア.load(null、null); キーストア.setCertificateEntry("ca-certificate")
証明書; TrustManager 本社工場
tmf=TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm()));
tmf.init(keystore)、SSLContext=SSLContext.getInstance(「TLS」)。
context.init(null, tmf.getTrustManager(), null); を返します。
お問い合わせ _ socketFactory

MQTT 5の特長

//パホmqttv5
インポートorg.eclipse.paho.mqttv5.client.IMqttClient;
org.eclipse.paho.mqttv5.clientを輸入して下さい。 MqttClient;
インポートorg.eclipse.paho.mqttv5.client.MqttConnectionOptions;
インポートorg.eclipse.paho.mqttv5.client.persist.MemoryPersistence;
インポートorg.eclipse.paho.mqttv5.common.MqttException;
インポートorg.eclipse.paho.mqttv5.common.MqttMessage;
インポートorg.eclipse.paho.mqttv5.common.packet.MqttProperties;
インポートorg.eclipse.paho.mqttv5.client.MqttCallback;
インポートorg.eclipse.paho.mqttv5.client.MqttDisconnectResponse;
インポートorg.eclipse.paho.mqttv5.client.IMqttToken;
javax.net.sl.SSLSocketFactoryをインポートします。
javax.net.ssl.SSLContext をインポートします。
java.security.cert.Certificate をインポート 工場;
java.security.cert.X509Certificate をインポートします。
java.security.KeyStore をインポートします。
java.io をインポートします。 ファイル入力ストリーム;
java.io.IOException のインポート
java.security.cert.CertificateException のインポート
java.security.NoSuchAlgorithmException をインポートします。
インポート javax.net.sl.TrustManager 工場;
javax.net.sl.KeyManagerFactoryをインポートします。
java.security.KeyManagementException のインポート
java.util をインポートします。 リスト;
java.util をインポートします。 ランダム;
java.util をインポートします。 スキャナ;
パブリッククラス MqttClientv5 お問い合わせ
パブリック static void main(String[] args) のスロー Exception{
文字列 client_id = "your_client_id";
文字列 hostip = "public-mqtt-broker.bevywise.com";
boolean AUTH = true;
文字列 client_username = "Access-Key";
文字列 client_password = "アクセストークン";
boolean websocket_connection=false; //websocketを有効にするためにtrueを設定
文字列のトピック = "";
int qos = 0;
boolean は = false を保持します。
ストリングブローカーUrl= "";
intの港;
もし(websocket_connection)
port= 10433;
その他
ポート = 1883; // SSL/TLS で 8883 を使用

// 新しいMQTTクライアントインスタンスを作成する
MqttClientクライアント
もし(websocket_connection)
BrokerUrl=String.format("%s://%s:%d"、"ws"、hostip、ポート);
その他
ブローカー Url = 文字列.format("%s://%s:%d", "tcp", hostip, port); で指定します。

MemoryPersistence の永続性 = new MemoryPersistence();
お問い合わせ
クライアント = 新しいMqttClient(ブローカー) Url、client_id、永続性;
// コールバックの設定
client.setCallback(新しいMqttCallback() {
@オーバーライド
未接続(MqttDisconnectResponse DisconnectResponse) {
System.out.println(「ブローカーから切断」);
お問い合わせ
@オーバーライド
パブリック void mqtt エラー発生(MqttException例外) {
System.out.println("MQTT Error:" +except.getMessage()); );
例外。printStackTrace();
お問い合わせ
@オーバーライド
パブリック void メッセージ 到着(文字列トピック、MqttMessageメッセージ)は例外を投げます お問い合わせ
System.out.println(「トピックに関するメッセージを受け取る」+トピック);
System.out.println("メッセージ:" + new String(message.getPayload()));
お問い合わせ
@オーバーライド
公共 void 配達 完全な(IMqttTokenトークン) {
System.out.println(「メッセージが正常に配信されました」)。
お問い合わせ
@オーバーライド
パブリック void authPacketArrived(int resultCode, MqttProperties プロパティ) お問い合わせ
System.out.println(「Auth パケット」は、理由コード「+ 理由コード」で到着しました)。
お問い合わせ
@オーバーライド
パブリック void connectComplete(boolean reconnect, String serverURI) {
System.out.println("ブローカーに接続:" + serverURI);
お問い合わせ
// 接続オプションの設定
MqttConnectionOptions connOpts = 新しいMqttConnectionOptions();
お問い合わせ
お問い合わせ
connOpts.setSocketFactory(getSocketFactory("/path/to/your/root.crt")); // // // //
ルートの証明書パスに置き換える
お問い合わせ
System.out.println(「SSLソケット工場の設定エラー」+ e.getMessage() );
e.printStackTrace() ;
お問い合わせ
お問い合わせ
お問い合わせ
connOpts.setUserName(client_username) );
connOpts.setPassword(client_password.getBytes())) ); // // // // バイト配列に変換する
お問い合わせ
connOpts.setCleanStart(true); // MQTTv5のクリーンスタート
connOpts.setMaxInflight(50);
// MQTTブローカーに接続する
System.out.println(「ブローカーに接続する:「+ブローカーUrl」);
client.connect(connOpts);
System.out.println(「コネクテッド」);
// 購読するトピックの例
最後の文字列の購読 = "will_topic/device1";
// トピックへのサブスクライブ
client.subscribe(サブスクライブトピック、qos);
// ユーザの入力に基づいてメッセージやサブスクライブ/サブスクライブを投稿
スキャナスキャナー = 新しいスキャナ(System.in)。
時 (true) {
System.out.println("1. Publish\n2.subscribe\n3. Disconnect\n4. 購読解除
int 選択 = スキャナ.nextInt();
Scanner.nextLine(); // 新規ラインを消費する
スイッチ (choice) {
ケース1:
System.out.print("Enter Topic QoS(0/1/2) 保持(0/1) カウント: ");
文字列[] 入力 = スキャナ.nextLine().split(");
トピック = 入力[0];
qos = Integer.parseInt(input[1]); ;
保持 = Integer.parseInt(input[2]) === 1;
int カウント = Integer.parseInt(input[3]);;
(int i = 0; i < count; i++) { //String message="{\"temp\":" + (new)
Random().nextInt(50) + 1) + ",\" 圧力\":" + (新しいRandom().nextInt(100) + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
1 + ",\" ステータス: + (new Random().nextInt(2)==0 ? "ON" : "OFF" ) + "}" ;;;
文字列 message="hello world"; MqttMessage mqttMessage=new
MqttMessage(message.getBytes()); mqttMessage.setQos(qos);
mqttMessage.setRetained(保持); System.out.println("メッセージ: 「+メッセージ」
client.publish(トピック、mqttMessage)
System.out.println(「パケット:+i+」) System.out.print(「入力トピックQoS(0/1/2):」);
入力 = スキャナ.nextLine().split(");
トピック = 入力[0];
qos = Integer.parseInt(input[1]); ;
client.subscribe(トピック、qos);
ブレイク;
場合3:
クライアント.disconnect();
System.out.println("ブローカーから切断")); 戻り値 4: System.out.print(「トピックを入力: ");
トピック = Scanner.nextLine();
client.unsubscribe(トピック);
ブレイク;
デフォルト:
System.out.println(「無効な選択」) お問い合わせ (MqttException e) { System.out.println("エラー")
MQTTブローカーに接続する: " + e.getMessage());
e.printStackTrace() ;
お問い合わせ
お問い合わせ
プライベート静的SSLSocket 工場getSocket 工場(String rootCertPath) スロー 例外 {
証明書工場 cf = CertificateFactory.getInstance(" X.509"); ファイル入力ストリーム fis=new
FileInputStream(rootCertPath)、X509Certificate caCert=(X509Certificate)
cf.generateCertificate(fis); fis.close(); キーストア
ks=KeyStore.getInstance("JKS"); ks.load(null、null);
ks.setCertificateEntry("ca-certificate"、caCert); TrustManager 本社工場
tmf=TrustManagerFactory.getInstance("SunX509"); tmf.init(ks); SSLContext
sslContext=SSLContext.getInstance("TLSv1.2"); sslContext.init(null,
tmf.getTrustManager(), null;sslContext.getSocketFactory() を返します。