ApsaraMQ for MQTT:Java SDK を使用した MQTT クライアント間のメッセージ送受信

エンドポイント

クライアントまたはクラウドサービスが ApsaraMQ for MQTT と通信する場合、SDK コードで ApsaraMQ for MQTT インスタンスのエンドポイントを指定する必要があります。エンドポイントのフォーマットは、SDK の種類とネットワークアクセス方法によって異なります。

クライアント SDK のエンドポイントフォーマット

クライアント SDK を介して接続する場合、以下のエンドポイントフォーマットを使用します。

クラウド SDK のエンドポイントフォーマット

クラウド SDK を介して接続する場合、以下のエンドポイントフォーマットを使用します。

クライアント SDK のエンドポイントは、ApsaraMQ for MQTT コンソール の [Instance Details] ページの [Endpoints] タブでも確認できます。インスタンス ID は [Basic Information] セクションに表示されます。

ステップ 1: Maven 依存関係の追加

デモプロジェクト をクローンまたはダウンロードし、pom.xml に以下の依存関係が含まれていることを確認してください。

<dependencies>
    <!-- Eclipse Paho： MQTT 3.1.1 クライアントライブラリ -->
    <dependency>
        <groupId>org.eclipse.paho</groupId>
        <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
        <version>1.2.2</version>
    </dependency>
    <!-- Commons Codec： 認証用の HMAC 署名計算 -->
    <dependency>
        <groupId>commons-codec</groupId>
        <artifactId>commons-codec</artifactId>
        <version>1.10</version>
    </dependency>
    <!-- Apache HttpClient： トークンベース認証用の HTTP リクエスト -->
    <dependency>
        <groupId>org.apache.httpcomponents</groupId>
        <artifactId>httpclient</artifactId>
        <version>4.5.2</version>
    </dependency>
    <!-- Fastjson： API レスポンス用の JSON 解析 -->
    <dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>fastjson</artifactId>
        <version>1.2.83</version>
    </dependency>
    <!-- Alibaba Cloud SDK for MQTT： サーバーサイド API コール -->
    <dependency>
        <groupId>com.aliyun</groupId>
        <artifactId>aliyun-java-sdk-onsmqtt</artifactId>
        <version>1.0.3</version>
    </dependency>
    <!-- Alibaba Cloud Core SDK： 共有 SDK インフラストラクチャ -->
    <dependency>
        <groupId>com.aliyun</groupId>
        <artifactId>aliyun-java-sdk-core</artifactId>
        <version>4.5.0</version>
    </dependency>
</dependencies>

Eclipse Paho Java Client のソースコードとドキュメントについては、「Eclipse Paho Java Client」をご参照ください。

ステップ 2: 環境変数の設定

ソースコード内での認証情報の偶発的な漏洩を防ぐために、AccessKey ペアを環境変数に保存してください。

export MQTT_AK_ENV=<your-access-key-id>
export MQTT_SK_ENV=<your-access-key-secret>

ステップ 3: 接続パラメーターの定義

ApsaraMQ for MQTT コンソール で作成したリソースに基づいて、以下のパラメーターを設定してください。

// コンソールの [Basic Information] セクションのインスタンス ID
String instanceId = "<your-instance-id>";

// コンソールの [Endpoints] タブのエンドポイント
String endPoint = "<your-instance-id>.mqtt.aliyuncs.com";

// 環境変数からの認証情報
String accessKey = System.getenv("MQTT_AK_ENV");
// 署名認証モードには AccessKey シークレットが必要です
String secretKey = System.getenv("MQTT_SK_ENV");

// クライアント ID のフォーマット： GroupID@@@DeviceID (最大 64 文字)
// 各 TCP 接続では、一意のクライアント ID を使用する必要があります。
// クライアント ID が重複すると、接続の競合や予期しない切断が発生します。
String clientId = "<your-group-id>@@@<your-device-id>";

// コンソールで作成した親トピック
final String parentTopic = "<your-parent-topic>";

// メッセージフィルタリング用のサブトピック (合計最大 128 文字)
final String mq4IotTopic = parentTopic + "/" + "testMq4Iot";

// QoS レベル： 0 (最大 1 回)、1 (最低 1 回)、2 (正確に 1 回)
final int qosLevel = 0;

ステップ 4: MQTT クライアントの作成とコールバックの登録

接続する 前 にコールバックを登録してください。connect() の後にコールバックを登録すると、特に永続セッションを再開する際にメッセージを見逃す可能性があります。

MQTT クライアントを作成します。

ConnectionOptionWrapper connectionOptionWrapper =
    new ConnectionOptionWrapper(instanceId, accessKey, secretKey, clientId);

final MemoryPersistence memoryPersistence = new MemoryPersistence();

// プロトコルとポートは一致している必要があります：
//   TCP -> tcp://endpoint:1883
//   SSL -> ssl://endpoint:8883
final MqttClient mqttClient = new MqttClient(
    "tcp://" + endPoint + ":1883", clientId, memoryPersistence);

// ブローカーの応答を待機するタイムアウト (ミリ秒)
mqttClient.setTimeToWait(5000);

final ExecutorService executorService = new ThreadPoolExecutor(
    1, 1, 0, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());

接続イベントと受信メッセージを処理するために、イベントコールバックを設定してください。

mqttClient.setCallback(new MqttCallbackExtended() {
    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        // 各接続後 (再接続を含む) に直ちにサブスクライブします
        System.out.println("接続成功");
        executorService.submit(() -> {
            try {
                final String[] topicFilter = {mq4IotTopic};
                final int[] qos = {qosLevel};
                mqttClient.subscribe(topicFilter, qos);
            } catch (MqttException e) {
                e.printStackTrace();
            }
        });
    }

    @Override
    public void connectionLost(Throwable throwable) {
        throwable.printStackTrace();
    }

    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) {
        // 受信メッセージはここで処理します。
        // このコールバックから例外をスローしないでください。ブローカーは、
        // 正常に返されたコールバックをメッセージの確認応答として扱います。
        // べき等な消費を保証するために、重複排除を実装してください。
        System.out.println(
            "トピック " + topic + " からメッセージを受信しました。本文："
            + new String(mqttMessage.getPayload()));
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        System.out.println("メッセージ送信成功。トピック："
            + iMqttDeliveryToken.getTopics()[0]);
    }
});

主要なコールバックの動作：

ステップ 5: 接続とメッセージのパブリッシュ

ブローカーに接続し、メッセージをパブリッシュします。この例では、通常の Pub/Sub メッセージとポイントツーポイント (P2P) メッセージの両方を送信します。

mqttClient.connect(connectionOptionWrapper.getMqttConnectOptions());

for (int i = 0; i < 10; i++) {
    // サブスクライブしたトピックに通常のメッセージをパブリッシュ
    MqttMessage message = new MqttMessage("こんにちは、Pub/Sub メッセージです".getBytes());
    message.setQos(qosLevel);
    mqttClient.publish(mq4IotTopic, message);

    // 特定のクライアントに P2P メッセージを直接パブリッシュ
    // P2P トピックのフォーマット： {parentTopic}/p2p/{targetClientId}
    // ターゲットクライアントは、このトピックにサブスクライブする必要はありません。
    final String p2pSendTopic = parentTopic + "/p2p/" + clientId;
    message = new MqttMessage("こんにちは、P2P メッセージです".getBytes());
    message.setQos(qosLevel);
    mqttClient.publish(p2pSendTopic, message);
}

// メッセージを受信し続けるために、クライアントを実行したままにします
Thread.sleep(Long.MAX_VALUE);

P2P メッセージング： ターゲットクライアントが正確にわかっている場合、トピック {parentTopic}/p2p/{targetClientId} にメッセージを送信します。ターゲットクライアントはサブスクライブすることなくメッセージを受信します。これにより、メッセージが単一の受信者を対象とするシナリオが簡素化されます。

完全なコード

以下は、上記のすべてのステップを単一の実行可能なクラスにまとめたものです。これは GitHub のデモプロジェクト でも入手できます。

package com.aliyun.openservices.lmq.example.demo;

import com.aliyun.openservices.lmq.example.util.ConnectionOptionWrapper;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

public class MQ4IoTSendMessageToMQ4IoTUseSignatureMode {
    public static void main(String[] args) throws Exception {
        // --- 接続パラメーター ---
        // コンソールの [Basic Information] セクションのインスタンス ID
        String instanceId = "<your-instance-id>";
        // コンソールの [Endpoints] タブのエンドポイント
        String endPoint = "<your-instance-id>.mqtt.aliyuncs.com";
        // 環境変数からの認証情報
        String accessKey = System.getenv("MQTT_AK_ENV");
        // 署名認証モードには AccessKey シークレットが必要です
        String secretKey = System.getenv("MQTT_SK_ENV");
        // クライアント ID のフォーマット： GroupID@@@DeviceID (最大 64 文字)
        // 各 TCP 接続では、一意のクライアント ID を使用する必要があります。
        String clientId = "<your-group-id>@@@<your-device-id>";
        // コンソールで作成した親トピック
        final String parentTopic = "<your-parent-topic>";
        // メッセージフィルタリング用のサブトピック
        final String mq4IotTopic = parentTopic + "/" + "testMq4Iot";
        // QoS レベル： 0 (最大 1 回)、1 (最低 1 回)、2 (正確に 1 回)
        final int qosLevel = 0;

        // --- クライアントの作成 ---
        ConnectionOptionWrapper connectionOptionWrapper =
            new ConnectionOptionWrapper(instanceId, accessKey, secretKey, clientId);
        final MemoryPersistence memoryPersistence = new MemoryPersistence();
        final MqttClient mqttClient = new MqttClient(
            "tcp://" + endPoint + ":1883", clientId, memoryPersistence);
        mqttClient.setTimeToWait(5000);

        final ExecutorService executorService = new ThreadPoolExecutor(
            1, 1, 0, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());

        // --- 接続前にコールバックを登録 ---
        mqttClient.setCallback(new MqttCallbackExtended() {
            @Override
            public void connectComplete(boolean reconnect, String serverURI) {
                System.out.println("接続成功");
                executorService.submit(() -> {
                    try {
                        final String[] topicFilter = {mq4IotTopic};
                        final int[] qos = {qosLevel};
                        mqttClient.subscribe(topicFilter, qos);
                    } catch (MqttException e) {
                        e.printStackTrace();
                    }
                });
            }

            @Override
            public void connectionLost(Throwable throwable) {
                throwable.printStackTrace();
            }

            @Override
            public void messageArrived(String topic, MqttMessage mqttMessage) {
                // 受信メッセージはここで処理します。
                // このコールバックから例外をスローしないでください。ブローカーは、
                // 正常に返されたコールバックをメッセージの確認応答として扱います。
                // べき等な消費を保証するために、重複排除を実装してください。
                System.out.println(
                    "トピック " + topic + " からメッセージを受信しました。本文："
                    + new String(mqttMessage.getPayload()));
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
                System.out.println("メッセージ送信成功。トピック："
                    + iMqttDeliveryToken.getTopics()[0]);
            }
        });

        // --- 接続とパブリッシュ ---
        mqttClient.connect(connectionOptionWrapper.getMqttConnectOptions());
        for (int i = 0; i < 10; i++) {
            MqttMessage message = new MqttMessage("こんにちは、Pub/Sub メッセージです".getBytes());
            message.setQos(qosLevel);
            mqttClient.publish(mq4IotTopic, message);

            final String p2pSendTopic = parentTopic + "/p2p/" + clientId;
            message = new MqttMessage("こんにちは、P2P メッセージです".getBytes());
            message.setQos(qosLevel);
            mqttClient.publish(p2pSendTopic, message);
        }
        Thread.sleep(Long.MAX_VALUE);
    }
}

結果の検証

アプリケーションを実行した後、ApsaraMQ for MQTT コンソール でメッセージトレースを照会し、メッセージが送受信されたことを確認してください。詳細な手順については、「メッセージトレースの照会」をご参照ください。

次のステップ

• MQTT クライアントとバックエンドサービスアプリケーション間のメッセージ送受信

• 用語