ApsaraMQ for MQTT:ApsaraMQ for MQTT から ApsaraMQ for RocketMQ にクライアントステータス通知を送信する

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.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

public class MQ4IoTSendMessageToMQ4IoTUseSignatureMode {
    public static void main(String[] args) throws Exception {
        /**
         * ApsaraMQ for MQTT インスタンスの ID です。インスタンスを購入した後、ApsaraMQ for MQTT コンソールで ID を取得できます。
         */
        String instanceId = "XXXXX";
        /**
         * ApsaraMQ for MQTT インスタンスのエンドポイントです。インスタンスを購入して構成した後、ApsaraMQ for MQTT コンソールでエンドポイントを取得できます。インスタンスに接続するには、割り当てられたドメイン名を使用する必要があります。 IP アドレスは使用しないでください。そうしないと、クライアントオブジェクトで例外が発生する可能性があります。
         */
        String endPoint = "XXXXX.mqtt.aliyuncs.com";
        /**
         * AccessKey ID です。これは、Resource Access Management (RAM) コンソールで取得できます。
         * Alibaba Cloud アカウントの AccessKey ペアは、すべての API 操作に対する権限を持っています。セキュリティリスクを防ぐため、RAM ユーザーを使用して API 操作を呼び出したり、日常の O&M を実行することをお勧めします。
         * AccessKey ペアをプロジェクトコードに保存しないことを強くお勧めします。そうしないと、AccessKey ペアが漏洩し、アカウント内のすべてのリソースが潜在的なセキュリティリスクにさらされる可能性があります。
         * この例では、AccessKey ID と AccessKey シークレットは環境変数に格納されています。サンプルコードを実行する前に、MQTT_AK_ENV および MQTT_SK_ENV 環境変数が構成されていることを確認してください。
         * 例：export MQTT_AK_ENV=<access_key_id>
         *      export MQTT_SK_ENV=<access_key_secret>
         * <access_key_id> を AccessKey ID に、<access_key_secret> を AccessKey シークレットに置き換えます。
         */
        String accessKey = System.getenv("MQTT_AK_ENV");
        /**
         * AccessKey シークレットです。これは、RAM コンソールで取得できます。このパラメーターは、署名認証モードを使用する場合にのみ必要です。
         */
        String secretKey = System.getenv("MQTT_SK_ENV");
        /**
         * システムが ApsaraMQ for MQTT クライアントに割り当てるグローバルに一意な ID です。クライアント ID は、TCP 接続に基づいて変化する必要があります。複数の TCP 接続が同じクライアント ID を使用すると、例外が発生し、接続が予期せず閉じられます。
         * clientId パラメーターの値は、GroupID@@@DeviceId 形式です。 GroupID は、ApsaraMQ for MQTT コンソールで作成したグループの ID を指定し、DeviceId はデバイスのカスタム ID を指定します。 clientId パラメーターの値は、最大 64 文字の長さにすることができます。
         */
        String clientId = "GID_XXXXX@@@XXXXX";

        ConnectionOptionWrapper connectionOptionWrapper = new ConnectionOptionWrapper(instanceId, accessKey, secretKey, clientId);
        final MemoryPersistence memoryPersistence = new MemoryPersistence();
        /**
         * ApsaraMQ for MQTT クライアントオブジェクトで使用されるプロトコルとポートは一致する必要があります。詳細については、https://www.alibabacloud.com/help/doc-detail/44866.htm?spm=a2c4g.11186623.6.552.25302386RcuYFB を参照してください。
         * Secure Sockets Layer (SSL) 暗号化を使用する場合は、ssl://endpoint:8883 をプロトコルとポートとして指定します。
         */
        final MqttClient mqttClient = new MqttClient("tcp://" + endPoint + ":1883", clientId, memoryPersistence);
        /**
         * ApsaraMQ for MQTT クライアントが応答を待つタイムアウト期間です。タイムアウト期間により、ApsaraMQ for MQTT クライアントが無期限に応答を待つことがなくなります。
         */
        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());
        Thread.sleep(Long.MAX_VALUE);
    }
}

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Consumer;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.MessageListener;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import java.util.Map;
import java.util.Properties;
import java.util.Set;

public class MQTTClientStatusNoticeProcessDemo {
    public static void main(String[] args) {
        /**
         * レシーバーとして ApsaraMQ for RocketMQ クライアントを初期化します。ほとんどのビジネスシナリオでは、レシーバーはバックエンドアプリケーションにデプロイされます。
         */
        Properties properties = new Properties();
        /**
         * ApsaraMQ for RocketMQ コンソールで作成したグループ ID です。
         */
        properties.setProperty(PropertyKeyConst.GROUP_ID, "GID_XXXX");
        /**
         * ID 認証のために Alibaba Cloud RAM コンソールで作成した AccessKey ID です。
         * Alibaba Cloud アカウントの AccessKey ペアは、すべての API 操作に対する権限を持っています。セキュリティリスクを防ぐため、RAM ユーザーを使用して API 操作を呼び出したり、日常の O&M を実行することをお勧めします。
         * AccessKey ペアをプロジェクトコードに保存しないことを強くお勧めします。そうしないと、AccessKey ペアが漏洩し、アカウント内のすべてのリソースが潜在的なセキュリティリスクにさらされる可能性があります。
         * この例では、AccessKey ID と AccessKey シークレットは環境変数に格納されています。
         */
        properties.put(PropertyKeyConst.AccessKey, System.getenv("MQTT_AK_ENV"));
        /**
         * ID 認証のために Alibaba Cloud RAM コンソールで作成した AccessKey シークレットです。このパラメーターは、署名認証モードを使用する場合にのみ必要です。
         */
        properties.put(PropertyKeyConst.SecretKey, System.getenv("MQTT_SK_ENV"));
        /**
         * ApsaraMQ for RocketMQ インスタンスにアクセスするために使用される TCP エンドポイントです。 TCP エンドポイントは、ApsaraMQ for RocketMQ コンソールの [インスタンスの詳細] ページで取得できます。
         */
        properties.put(PropertyKeyConst.NAMESRV_ADDR, "XXXX");
        /**
         * ApsaraMQ for MQTT からのクライアントステータス通知の処理に使用する ApsaraMQ for RocketMQ トピックです。
         */
        final String parentTopic = "GID_XXXX_MQTT";
        /**
         * クライアントステータスデータです。本番環境では、データベースや Redis システムなどの外部永続ストレージシステムを使用してステータスデータを格納し、アプリケーションの再起動時にステータスデータが失われないようにすることをお勧めします。この例では、ステータスデータはオンプレミスマシンに格納されます。
         */
        MqttClientStatusStore mqttClientStatusStore = new MemoryHashMapStoreImpl();
        Consumer consumer = ONSFactory.createConsumer(properties);
        /**
         * この例では、クライアントがオンラインかどうか