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

クライアント SDK のエンドポイント

ApsaraMQ for MQTT コンソールの エンドポイント タブ（インスタンスの詳細 ページ内）で、クライアント SDK のエンドポイントを確認できます。ApsaraMQ for MQTT コンソール

クラウド SDK のエンドポイント

ApsaraMQ for MQTT コンソールの 基本情報 セクション（インスタンスの詳細 ページ内）で、インスタンス ID を確認できます。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(
            "receive msg from topic " + s + " , body is "
            + new String(mqttMessage.getPayload()));
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        System.out.println("send msg succeed topic is : "
            + 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(
                    "receive msg from topic " + s + " , body is "
                    + new String(mqttMessage.getPayload()));
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
                System.out.println("send msg succeed topic is : "
                    + iMqttDeliveryToken.getTopics()[0]);
            }
        });

        mqttClient.connect(connectionOptionWrapper.getMqttConnectOptions());

        for (int i = 0; i < 10; i++) {
            // サブトピックへのパブリッシュ（標準のパブリッシュ／サブスクライブ）
            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
send msg succeed topic is : XXXXX/testMq4Iot
send msg succeed topic is : XXXXX/p2p/xxx
...

プロジェクトのセットアップ

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("Receive:" + 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("Receive:" + msgId + ","
                    + JSONObject.toJSONString(messageProperties) + ","
                    + new String(payload));
            }
        });
    }
}

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

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

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 を使用します。