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

以下の例では、ApsaraMQ for RocketMQ コンシューマーを使用してクライアントステータス通知を受信し、処理します。コンシューマーは、指定された RocketMQ トピックの connect および tcpclean タグをサブスクライブします。

完全なソースコードについては、「MQTTClientStatusNoticeProcessDemo.java」をご参照ください。

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) {
        // ステータス通知を受信するための ApsaraMQ for RocketMQ コンシューマーを初期化します。
        Properties properties = new Properties();

        // RocketMQ のコンシューマーグループ ID (ApsaraMQ for MQTT のグループ ID とは異なります)。
        properties.setProperty(PropertyKeyConst.GROUP_ID, "<your-rocketmq-group-id>");

        // AccessKey ペア。RAM コンソールから取得します。
        properties.put(PropertyKeyConst.AccessKey, "<your-access-key>");
        properties.put(PropertyKeyConst.SecretKey, "<your-secret-key>");

        // ApsaraMQ for RocketMQ インスタンスの TCP エンドポイント。
        properties.put(PropertyKeyConst.NAMESRV_ADDR, "<your-rocketmq-endpoint>");

        // MQTT クライアントステータス通知を受信する RocketMQ トピック。
        // このトピックは事前に ApsaraMQ for RocketMQ コンソールで作成しておきます。
        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("Client online: " + 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;
                }

                // この channelId に connect と tcpclean の両方のイベントが存在する場合、
                // 接続はクローズされています。エントリをクリーンアップします。
                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;
        }
    }

    /**
     * クライアントがアクティブな TCP 接続を持っているかどうかを確認します。
     *
     * ロジック：
     * 1. チャネルマップが空の場合、クライアントはオフラインです。
     * 2. いずれかの channelId に connect イベントのみ (tcpclean なし) がある場合、クライアントはオンラインです。
     * 3. すべての channelId に connect と tcpclean の両方のイベントがある場合、クライアントはオフラインです。
     */
    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;
    }

}

以下のプレースホルダーを実際の値に置き換えてください。