ApsaraMQ for MQTT:Ekspor event status client

Contoh berikut menggunakan konsumen ApsaraMQ for RocketMQ untuk menerima dan memproses notifikasi status klien. Konsumen berlangganan tag connect dan tcpclean pada topik RocketMQ yang ditentukan.

Untuk kode sumber lengkap, lihat 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) {
        // Inisialisasi konsumen ApsaraMQ for RocketMQ untuk menerima notifikasi status.
        Properties properties = new Properties();

        // ID group konsumen RocketMQ (berbeda dari ID group ApsaraMQ for MQTT).
        properties.setProperty(PropertyKeyConst.GROUP_ID, "<your-rocketmq-group-id>");

        // Pasangan AccessKey. Dapatkan dari Konsol RAM.
        properties.put(PropertyKeyConst.AccessKey, "<your-access-key>");
        properties.put(PropertyKeyConst.SecretKey, "<your-secret-key>");

        // Titik akhir TCP instans ApsaraMQ for RocketMQ.
        properties.put(PropertyKeyConst.NAMESRV_ADDR, "<your-rocketmq-endpoint>");

        // Topik RocketMQ yang menerima notifikasi status client MQTT.
        // Buat topik ini terlebih dahulu di Konsol ApsaraMQ for RocketMQ.
        final String parentTopic = "GID_XXXX_MQTT";

        // Di lingkungan produksi, gunakan penyimpanan eksternal (database atau Redis)
        // untuk menyimpan data status secara persisten meskipun aplikasi dimulai ulang.
        MqttClientStatusStore mqttClientStatusStore = new MemoryHashMapStoreImpl();

        Consumer consumer = ONSFactory.createConsumer(properties);

        // Berlangganan hanya event "connect" dan "tcpclean".
        consumer.subscribe(parentTopic, "connect||tcpclean",
            new MqttClientStatusNoticeListener(mqttClientStatusStore));
        consumer.start();

        // Periksa status client tertentu secara berkala.
        String clientId = "GID_XXXXxXX@@@XXXXX";
        while (true) {
            System.out.println("Client online: " + checkClientOnline(clientId, mqttClientStatusStore));
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * Memproses notifikasi status masuk dan memelihara peta channel-event.
     *
     * Di penerapan multi-server, simpan status client di penyimpanan eksternal bersama
     * (database atau Redis). Implementasikan konsumsi idempoten untuk menangani pesan duplikat.
     */
    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"));

                // Simpan event baru.
                mqttClientStatusStore.addEvent(clientId, channelId, eventType, event);

                // Baca daftar event untuk koneksi ini.
                Set<ClientStatusEvent> events = mqttClientStatusStore.getEvent(clientId, channelId);
                if (events == null || events.isEmpty()) {
                    return Action.CommitMessage;
                }

                // Jika terdapat event connect dan tcpclean untuk channelId ini,
                // koneksi tersebut ditutup. Bersihkan entri tersebut.
                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;
        }
    }

    /**
     * Memeriksa apakah client memiliki koneksi TCP aktif.
     *
     * Logika:
     * 1. Jika peta channel kosong, client sedang offline.
     * 2. Jika ada channelId yang hanya memiliki event connect (tanpa tcpclean), client sedang online.
     * 3. Jika semua channelId memiliki event connect dan tcpclean, client sedang offline.
     */
    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;
    }

}

Ganti placeholder berikut dengan nilai aktual: