ApsaraMQ for MQTT:MQTT クライアントとバックエンドアプリケーション間のメッセージ送受信

仕組み

ApsaraMQ for MQTT クライアントとバックエンドアプリケーションは、ApsaraMQ for MQTT ブローカーを介して通信します。それぞれ異なる SDK を使用します。

完全なサンプルプロジェクトについては、「デモプロジェクト」または「クラウド SDK デモの実行」をご参照ください。

前提条件

開始する前に、以下を確認してください。

• ApsaraMQ for MQTT インスタンスに、必要なリソース (グループ、トピック) が作成済みであること。「リソースの作成」をご参照ください。

• AccessKey ペアが取得済みであること。「AccessKey ペアの取得」をご参照ください。

• JDK がインストール済みであること

• IntelliJ IDEA または Eclipse がインストール済みであること (このチュートリアルでは IntelliJ IDEA を使用します)

エンドポイント

ApsaraMQ for MQTT は複数のエンドポイントタイプをサポートしています。既存のインターネットエンドポイントと VPC エンドポイントに加えて、ネットワーク環境全体でより安定した安全なアクセスを実現するために、パブリック専用エンドポイントと VPC 専用エンドポイントが追加されています。このトピックでは、各エンドポイントタイプのシナリオ、サポートされるエディション、課金方法、および推奨事項について説明します。

ステップ 1：クライアント SDK プロジェクトの設定

1. サードパーティ製の Java 用オープンソース SDK、Eclipse Paho Java Client をダウンロードします。

2. デモプロジェクト mqtt-java-demo をダウンロードして解凍し、IntelliJ IDEA にインポートします。

pom.xml に以下の依存関係が含まれていることを確認してください。

<dependencies>
    <dependency>
        <groupId>commons-codec</groupId>
        <artifactId>commons-codec</artifactId>
        <version>1.10</version>
    </dependency>
    <dependency>
        <groupId>org.eclipse.paho</groupId>
        <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
        <version>1.2.2</version>
    </dependency>
    <dependency>
        <groupId>org.apache.httpcomponents</groupId>
        <artifactId>httpclient</artifactId>
        <version>4.5.2</version>
    </dependency>
    <dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>fastjson</artifactId>
        <version>1.2.83</version>
    </dependency>
    <dependency>
        <groupId>com.aliyun</groupId>
        <artifactId>aliyun-java-sdk-onsmqtt</artifactId>
        <version>1.0.3</version>
    </dependency>
    <dependency>
        <groupId>com.aliyun</groupId>
        <artifactId>aliyun-java-sdk-core</artifactId>
        <version>4.5.0</version>
    </dependency>
</dependencies>

ステップ 2：認証情報の設定

認証のために以下の環境変数を設定します。詳細については、「アクセス認証情報の設定」をご参照ください。

# AccessKey ID
export MQTT_AK_ENV=<your-access-key-id>
# AccessKey Secret
export MQTT_SK_ENV=<your-access-key-secret>

ステップ 3：クライアント SDK を使用したメッセージの送信

MQ4IoTProducerDemo.java を開き、ApsaraMQ for MQTT インスタンスの値で以下のパラメーターを更新します。これらの値の確認方法については、「リソースの作成」をご参照ください。

ブローカーへの接続

MQTT クライアントを作成し、接続を確立します。平文接続には tcp://endpoint:1883 を、SSL/TLS 接続には ssl://endpoint:8883 を使用します。

String instanceId = "XXXXX";
String endPoint = "XXXXX.mqtt.aliyuncs.com";
String accessKey = System.getenv("MQTT_AK_ENV");
String secretKey = System.getenv("MQTT_SK_ENV");
String clientId = "GID_XXXXX@@@XXXXX";

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);

クライアント ID は {GroupID}@@@{DeviceID} 形式を使用し、TCP 接続ごとにグローバルに一意である必要があります。接続間でクライアント ID を再利用すると、ブローカーは既存のセッションを切断します。

コールバックの設定

セッション再開中のメッセージの欠落を避けるため、接続前にコールバックを設定します。

mqttClient.setCallback(new MqttCallbackExtended() {
    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        System.out.println("connect success");
    }

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

    @Override
    public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
        // 再配信を避けるため、ブローカーのタイムアウト内にメッセージを処理します。
        // べき等な処理を保証するために、重複排除を実装します。
        System.out.println(
            "トピック " + s + " からメッセージを受信しました。本文: "
            + new String(mqttMessage.getPayload()));
    }

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

mqttClient.connect(connectionOptionWrapper.getMqttConnectOptions());

メッセージのパブリッシュ

サブトピックにメッセージをパブリッシュします。サブトピックは親トピックに追加され、最大 128 文字まで指定できます。

final String parentTopic = "XXXXX";
final String mq4IotTopic = parentTopic + "/" + "testMq4Iot";
final int qosLevel = 0;

MqttMessage message = new MqttMessage("hello mq4Iot pub sub msg".getBytes());
message.setQos(qosLevel);
mqttClient.publish(mq4IotTopic, message);

QoS レベル： 0 (最大 1 回)、1 (最低 1 回)、2 (正確に 1 回)。

ポイントツーポイント (P2P) メッセージの送信

P2P メッセージングは、特定のクライアントに直接メッセージを送信します。ターゲットクライアントはトピックをサブスクライブする必要はありません。{parentTopic}/p2p/{targetClientId} にパブリッシュします。

String receiverId = "GID_test@@@device_002";
final String p2pSendTopic = parentTopic + "/p2p/" + receiverId;
MqttMessage p2pMessage = new MqttMessage("hello mq4Iot p2p msg".getBytes());
p2pMessage.setQos(qosLevel);
mqttClient.publish(p2pSendTopic, p2pMessage);

プロデューサーの完全なコード

以下のコードは、上記のすべてのステップを組み合わせたものです。MQ4IoTProducerDemo.java を開き、パラメーターを更新してください。

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 MQ4IoTProducerDemo {
    public static void main(String[] args) throws Exception {
        // ApsaraMQ for MQTT インスタンス ID
        String instanceId = "XXXXX";

        // クライアント SDK エンドポイント。IP アドレスではなくドメイン名を使用してください。
        String endPoint = "XXXXX.mqtt.aliyuncs.com";

        // 環境変数から認証情報を読み込みます
        String accessKey = System.getenv("MQTT_AK_ENV");
        String secretKey = System.getenv("MQTT_SK_ENV");

        // クライアント ID のフォーマット：{GroupID}@@@{DeviceID}
        // TCP 接続ごとに一意である必要があります。最大 64 文字。
        String clientId = "GID_XXXXX@@@XXXXX";

        // ApsaraMQ for MQTT コンソールで作成された親トピック。
        // 無効なトピックや権限のないトピックを指定すると、ブローカーは接続を閉じます。
        final String parentTopic = "XXXXX";

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

        // QoS レベル：0、1、または 2
        final int qosLevel = 0;

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

        // ポート 1883 で TCP 経由で接続します。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("connect success");
            }

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

            @Override
            public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
                // 再配信を避けるため、ブローカーのタイムアウト内にメッセージを処理します。
                // べき等な処理を保証するために、重複排除を実装します。
                System.out.println(
                    "トピック " + s + " からメッセージを受信しました。本文: "
                    + 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++) {
            // サブトピックにパブリッシュします (標準的な pub/sub)
            MqttMessage message = new MqttMessage("hello mq4Iot pub sub msg".getBytes());
            message.setQos(qosLevel);
            mqttClient.publish(mq4IotTopic, message);

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

プロデューサーの検証

MQ4IoTProducerDemo.java を実行します。接続に成功すると、以下のような出力が表示されます。

connect success
メッセージの送信に成功しました。トピック: XXXXX/testMq4Iot
メッセージの送信に成功しました。トピック: XXXXX/p2p/xxx
...

ステップ 4：クラウド SDK を使用したメッセージの受信

クラウド SDK は、バックエンドアプリケーションを ApsaraMQ for MQTT ブローカーに接続します。クライアント SDK とは異なり、クラウド SDK は親トピックのみをサブスクライブし (サブトピックはサブスクライブしません)、アプリケーションがトピック以下のすべてのメッセージを収集・分析するバックエンドのシナリオに適しています。

プロジェクトの設定

1. クラウド SDK をダウンロードします。最新バージョンについては、「リリースノート」をご参照ください。

2. デモプロジェクト mqtt-server-sdk-demo をダウンロードします。

3. パッケージを解凍し、IntelliJ IDEA にインポートします。

4. pom.xml に以下の依存関係が含まれていることを確認してください。

<dependencies>
    <dependency>
        <groupId>com.alibaba.mqtt</groupId>
        <artifactId>server-sdk</artifactId>
        <version>1.0.0.Final</version>
    </dependency>
    <dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>fastjson</artifactId>
        <version>1.2.83</version>
    </dependency>
</dependencies>

コンシューマーの設定と実行

MQTTConsumerDemo.java を開き、以下のパラメーターを更新します。

認証情報は、ステップ 2 で設定したのと同じ環境変数 (MQTT_AK_ENV と MQTT_SK_ENV) から読み込まれます。

接続とサブスクライブ

String domain = "post-cn-jaj3h8i****.mqtt.aliyuncs.com";
int port = 5672;
String instanceId = "post-cn-jaj3h8i****";
String accessKey = System.getenv("MQTT_AK_ENV");
String secretKey = System.getenv("MQTT_SK_ENV");
String firstTopic = "firstTopic";

ChannelConfig channelConfig = new ChannelConfig();
channelConfig.setDomain(domain);
channelConfig.setPort(port);
channelConfig.setInstanceId(instanceId);
channelConfig.setAccessKey(accessKey);
channelConfig.setSecretKey(secretKey);

ServerConsumer serverConsumer = new ServerConsumer(channelConfig, new ConsumerConfig());
serverConsumer.start();

serverConsumer.subscribeTopic(firstTopic, new MessageListener() {
    @Override
    public void process(String msgId, MessageProperties messageProperties,
                        byte[] payload) {
        System.out.println("受信:" + msgId + ","
            + JSONObject.toJSONString(messageProperties) + ","
            + new String(payload));
    }
});

コンシューマーの完全なコード

package com.aliyun.openservices.lmq.example;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.mqtt.server.ServerConsumer;
import com.alibaba.mqtt.server.callback.MessageListener;
import com.alibaba.mqtt.server.config.ChannelConfig;
import com.alibaba.mqtt.server.config.ConsumerConfig;
import com.alibaba.mqtt.server.model.MessageProperties;

public class MQTTConsumerDemo {
    public static void main(String[] args) throws Exception {
        // クラウド SDK エンドポイント。IP アドレスではなくドメイン名を使用してください。
        String domain = "post-cn-jaj3h8i****.mqtt.aliyuncs.com";

        // クラウド SDK ポート。5672 である必要があります。
        int port = 5672;

        // ApsaraMQ for MQTT インスタンス ID
        String instanceId = "post-cn-jaj3h8i****";

        // 環境変数から認証情報を読み込みます
        String accessKey = System.getenv("MQTT_AK_ENV");
        String secretKey = System.getenv("MQTT_SK_ENV");

        // サブスクライブする親トピック。
        // クラウド SDK は親トピックのみをサブスクライブします (サブトピックは不可)。
        // 無効なトピックや権限のないトピックを指定すると、ブローカーは接続を閉じます。
        String firstTopic = "firstTopic";

        ChannelConfig channelConfig = new ChannelConfig();
        channelConfig.setDomain(domain);
        channelConfig.setPort(port);
        channelConfig.setInstanceId(instanceId);
        channelConfig.setAccessKey(accessKey);
        channelConfig.setSecretKey(secretKey);

        ServerConsumer serverConsumer = new ServerConsumer(channelConfig, new ConsumerConfig());
        serverConsumer.start();

        // 受信メッセージをサブスクライブして処理します
        serverConsumer.subscribeTopic(firstTopic, new MessageListener() {
            @Override
            public void process(String msgId, MessageProperties messageProperties,
                                byte[] payload) {
                System.out.println("受信:" + msgId + ","
                    + JSONObject.toJSONString(messageProperties) + ","
                    + new String(payload));
            }
        });
    }
}

よくある質問

自動再接続を有効にするにはどうすればよいですか？

接続前に MqttConnectOptions で setAutomaticReconnect(true) を設定します。

MqttConnectOptions options = connectionOptionWrapper.getMqttConnectOptions();
options.setAutomaticReconnect(true);
mqttClient.connect(options);

SSL/TLS 経由で接続するにはどうすればよいですか？

tcp:// を ssl:// に置き換え、ポート 8883 を使用します。

final MqttClient mqttClient =
    new MqttClient("ssl://" + endPoint + ":8883", clientId, memoryPersistence);

バックエンドアプリケーションから MQTT クライアントにメッセージを送信できますか？

はい。逆方向のメッセージング (バックエンドからクライアント) には、クラウド SDK のデモプロジェクトにある MQTTProducerDemo.java を使用してください。

次のステップ

• デモプロジェクトで他のメッセージングパターンを確認する

• クラウド SDK アクセスがサポートされているリージョンを確認する

Step 6 Change Summary

Optimization type: Structure and expression optimization (Step 6 optimize-architecture)

Changes applied:

1. Note/Important format standardization: Changed all > **Note**: and > **Important**: patterns to > **Note:** and > **Important:** (colon inside bold) per en.md section 3.4 format consistency rules.

2. Removed "You can" filler patterns: Changed "You can find the client SDK endpoint on..." to "Find the client SDK endpoint on..." and "You can find the instance ID in..." to "Find the instance ID in..." per en.md section 2.3 (descriptive content should omit "you").

3. Verified structural integrity: The document follows proper progressive disclosure: How it works (context) -> Prerequisites -> Endpoints (reference) -> Step 1-4 (hands-on) -> FAQ (troubleshooting) -> What's next. This structure aligns with the feature-docs knowledge type.

4. Verified expression quality: All sentences use active voice, terminology is consistent throughout ("ApsaraMQ for MQTT", "client SDK", "cloud SDK", "VPC-dedicated endpoint"), no translation artifacts detected, no tautology found.

5. Preserved all technical content: All code blocks, parameter values, endpoint formats, SDK class names, and technical details remain unchanged. No facts were modified.

Items intentionally not changed: - All code blocks and inline code preserved exactly - All parameter tables preserved exactly - All endpoint formats and domain patterns preserved exactly - Document structure (section order) preserved -- the existing progressive flow is already well-organized - The draft was already high quality from Step 5, so changes were minimal and focused on format consistency and minor expression tightening