ApsaraMQ for MQTT:MQTT 版 ApsaraMQ クライアントのステータスイベントをエクスポートする

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

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) {
        /**
         * レシーバーとして RocketMQ 版 ApsaraMQ クライアントを初期化します。ほとんどのビジネスシナリオでは、コードはバックエンドアプリケーションにデプロイされます。
         */
        Properties properties = new Properties();
        /**
         * RocketMQ 版 ApsaraMQ クライアントのグループ ID を指定します。このパラメーターに指定するグループ ID は、MQTT 版 ApsaraMQ インスタンスのグループの ID とは異なることに注意してください。対応するサービスの説明に従ってグループ ID を指定する必要があります。
         */
        properties.setProperty(PropertyKeyConst.GROUP_ID, "GID_XXXX");
        /**
         * Alibaba Cloud アカウントの AccessKey ID。AccessKey ID は、Resource Access Management（RAM）コンソールから取得できます。
         */
        properties.put(PropertyKeyConst.AccessKey, "XXXX");
        /**
         * Alibaba Cloud アカウントの AccessKey シークレット。AccessKey シークレットは、RAM コンソールから取得できます。AccessKey シークレットは、署名認証モードを使用する場合にのみ必要です。
         */
        properties.put(PropertyKeyConst.SecretKey, "XXXX");
        /**
         * TCP エンドポイントを指定します。
         */
        properties.put(PropertyKeyConst.NAMESRV_ADDR, "http://XXXX");
        /**
         * RocketMQ 版 ApsaraMQ コンシューマーを使用して MQTT 版 ApsaraMQ クライアントのステータス通知を処理する場合、コンシューマーはステータス通知を受信するために使用されるトピックをサブスクライブする必要があります。RocketMQ 版 ApsaraMQ コンソールのドキュメントを参照して、事前にトピックを作成する必要があります。
         */
        final String parentTopic = "GID_XXXX_MQTT";
        /**
         * クライアントステータスデータ。本番環境では、アプリケーションの再起動時にステータスデータが失われないように、データベースや Redis システムなどの外部永続ストレージシステムを使用してステータスデータを格納することをお勧めします。この例では、ステータスデータはオンプレミスコンピューターに格納されます。
         */
        MqttClientStatusStore mqttClientStatusStore = new MemoryHashMapStoreImpl();
        Consumer consumer = ONSFactory.createConsumer(properties);
        /**
         * 次のコードは、クライアントがオンラインかどうかのみを判断します。したがって、connect イベントと tcpclean イベントにのみ注意する必要があります。
         */
        consumer.subscribe(parentTopic, "connect||tcpclean", new MqttClientStatusNoticeListener(mqttClientStatusStore));
        consumer.start();
        String clientId = "GID_XXXXxXX@@@XXXXX";
        while (true) {
            System.out.println("ClientStatus :" + checkClientOnline(clientId, mqttClientStatusStore));
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * クライアントステータス通知を処理するために使用されるロジック。
     * 実際のデプロイプロセスでは、ステータス通知メッセージを消費するアプリケーションが複数のサーバーにデプロイされる場合があります。したがって、クライアントのステータスデータは、データベースや Redis システムなどの外部共有ストレージに保持できます。
     * ステートマシンがメッセージを繰り返し受信する場合は、メッセージに対して消費冪等処理を実行して、発生する可能性のあるエラーを防ぎます。
     */
    static class MqttClientStatusNoticeListener implements MessageListener {
        private MqttClientStatusStore mqttClientStatusStore;

        public MqttClientStatusNoticeListener(
            MqttClientStatusStore mqttClientStatusStore) {
            this.mqttClientStatusStore = mqttClientStatusStore;
        }

        @Override
        public Action consume(Message message, ConsumeContext context) {
            try {
                JSONObject msgBody = JSON.parseObject(new String(message.getBody()));
                System.out.println(msgBody);
                String eventType = msgBody.getString("eventType");
                String clientId = msgBody.getString("clientId");
                String channelId = msgBody.getString("channelId");
                ClientStatusEvent event = new ClientStatusEvent();
                event.setChannelId(channelId);
                event.setClientIp(msgBody.getString("clientIp"));
                event.setEventType(eventType);
                event.setTime(msgBody.getLong("time"));
                /**
                 * 最初に新しいイベントを格納します。
                 */
                mqttClientStatusStore.addEvent(clientId, channelId, eventType, event);
                /**
                 * 現在の接続のイベントリストを読み取ります。
                 */
                Set<ClientStatusEvent> events = mqttClientStatusStore.getEvent(clientId, channelId);
                if (events == null || events.isEmpty()) {
                    return Action.CommitMessage;
                }
                /**
                 * リスト内のすべてのオンラインイベントとオフラインイベントが受信され、現在の接続が閉じられた場合、接続のデータをクリアできます。
                 */
                boolean findOnlineEvent = false;
                boolean findOfflineEvent = false;
                for (ClientStatusEvent clientStatusEvent : events) {
                    if (clientStatusEvent.isOnlineEvent()) {
                        findOnlineEvent = true;
                    } else {
                        findOfflineEvent = true;
                    }
                }
                if (findOnlineEvent && findOfflineEvent) {
                    mqttClientStatusStore.deleteEvent(clientId, channelId);
                }
                return Action.CommitMessage;
            } catch (Throwable e) {
                e.printStackTrace();
            }
            return Action.ReconsumeLater;
        }
    }

    /**
     * チャネルテーブルに基づいて、MQTT 版 ApsaraMQ クライアントにアクティブな TCP 接続が存在するかどうかを確認します。
     * 1. チャネルテーブルが空の場合、MQTT 版 ApsaraMQ クライアントはオフラインです。
     * 2. チャネルテーブルが空でない場合は、接続でオンラインイベントのみが受信されているかどうかを確認します。はいの場合、アクティブな接続が存在し、MQTT 版 ApsaraMQ クライアントはオンラインです。
     * すべてのチャネルに切断イベントがある場合、クライアントはオフラインです。
     *
     * @param clientId
     * @param mqttClientStatusStore
     * @return
     */
    public static boolean checkClientOnline(String clientId,
        MqttClientStatusStore mqttClientStatusStore) {
        Map<String, Set<ClientStatusEvent>> channelMap = mqttClientStatusStore.getEventsByClientId(clientId);
        if (channelMap == null) {
            return false;
        }
        for (Set<ClientStatusEvent> events : channelMap.values()) {
            boolean findOnlineEvent = false;
            boolean findOfflineEvent = false;
            for (ClientStatusEvent event : events) {
                if (event.isOnlineEvent()) {
                    findOnlineEvent = true;
                } else {
                    findOfflineEvent = true;
                }
            }
            if (findOnlineEvent & !findOfflineEvent) {
                return true;
            }
        }
        return false;
    }

}