ApsaraMQ for MQTT:Aturan notifikasi status klien

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 {
        /**
         * ID instans ApsaraMQ for MQTT. Dapatkan ini dari konsol setelah Anda membeli instans.
         */
        String instanceId = "XXXXX";
        /**
         * Titik akhir. Dapatkan titik akhir setelah Anda membeli dan mengonfigurasi instans ApsaraMQ for MQTT. 
         * Anda harus menggunakan nama domain yang ditetapkan untuk titik akhir. Menggunakan alamat IP langsung dapat menyebabkan exception pada klien.
         */
        String endPoint = "XXXXX.mqtt.aliyuncs.com";
        /**
         * ID AccessKey Akun Alibaba Cloud Anda. Dapatkan ini dari konsol.
         * Pasangan Kunci Akses akun Alibaba Cloud memiliki izin atas semua operasi API. Kami menyarankan agar Anda menggunakan Pengguna RAM untuk akses API atau O&M rutin guna mencegah risiko keamanan.
         * Kami sangat menyarankan agar Anda tidak menyematkan ID AccessKey dan Rahasia AccessKey secara langsung dalam kode proyek Anda. Hal ini dapat menimbulkan risiko keamanan jika kredensial bocor.
         * Contoh ini menunjukkan cara menyimpan ID AccessKey dan Rahasia AccessKey dalam variabel lingkungan. Sebelum menjalankan kode ini, konfigurasikan variabel lingkungan MQTT_AK_ENV dan MQTT_SK_ENV.
         * Misalnya: export MQTT_AK_ENV=<access_key_id>
         *      export MQTT_SK_ENV=<access_key_secret>
         * Ganti <access_key_id> dengan ID AccessKey Anda dan <access_key_secret> dengan Rahasia AccessKey Anda.
         */
        String accessKey = System.getenv("MQTT_AK_ENV");
        /**
         * Rahasia AccessKey Akun Alibaba Cloud Anda. Dapatkan ini dari konsol. Ini hanya diperlukan untuk otentikasi berbasis signature.
         */
        String secretKey = System.getenv("MQTT_SK_ENV");
        /**
         * ID klien untuk ApsaraMQ for MQTT, yang ditetapkan oleh sistem bisnis. Setiap koneksi TCP harus memiliki ID klien yang unik. 
         * Jika objek klien berbeda (koneksi TCP) menggunakan ID klien yang sama, akan terjadi exception koneksi.
         * ID klien terdiri dari dua bagian dalam format GroupID@@@DeviceId. Group ID diperoleh dari konsol ApsaraMQ for MQTT, dan DeviceId ditetapkan oleh sistem bisnis Anda. Panjang total ID klien tidak boleh melebihi 64 karakter.
         */
        String clientId = "GID_XXXXX@@@XXXXX";
        ConnectionOptionWrapper connectionOptionWrapper = new ConnectionOptionWrapper(instanceId, accessKey, secretKey, clientId);
        final MemoryPersistence memoryPersistence = new MemoryPersistence();
        /**
         * Protokol dan port yang digunakan klien harus sesuai. Untuk informasi selengkapnya, lihat https://www.alibabacloud.com/help/en/message-queue-for-mqtt/latest/limits
         * Untuk enkripsi SSL, atur endpoint menjadi ssl://endpoint:8883.
         */
        final MqttClient mqttClient = new MqttClient("tcp://" + endPoint + ":1883", clientId, memoryPersistence);
        /**
         * Tetapkan timeout untuk klien guna mencegahnya terblokir tanpa batas waktu.
         */
        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) {
        /**
         * Inisialisasi klien penerima ApsaraMQ for RocketMQ. Dalam skenario nyata, ini biasanya diterapkan di aplikasi backend.
         */
        Properties properties = new Properties();
        /**
         * Tetapkan group ID ApsaraMQ for RocketMQ. Buat ini di konsol ApsaraMQ for RocketMQ.
         */
        properties.setProperty(PropertyKeyConst.GROUP_ID, "GID_XXXX");
        /**
         * ID AccessKey untuk otentikasi Alibaba Cloud. Buat ini di konsol RAM Alibaba Cloud.
         * Pasangan Kunci Akses akun Alibaba Cloud memiliki izin atas semua operasi API. Kami menyarankan agar Anda menggunakan Pengguna RAM untuk akses API atau O&M rutin guna mencegah risiko keamanan.
         * Kami sangat menyarankan agar Anda tidak menyematkan ID AccessKey dan Rahasia AccessKey secara langsung dalam kode proyek Anda. Hal ini dapat menimbulkan risiko keamanan jika kredensial bocor.
         * Contoh ini menunjukkan cara menyimpan ID AccessKey dan Rahasia AccessKey dalam variabel lingkungan.
         */
        properties.put(PropertyKeyConst.AccessKey, System.getenv("MQTT_AK_ENV"));
        /**
         * Rahasia AccessKey untuk otentikasi Alibaba Cloud. Buat ini di konsol RAM Alibaba Cloud. Ini hanya diperlukan untuk otentikasi berbasis signature.
         */
        properties.put(PropertyKeyConst.SecretKey, System.getenv("MQTT_SK_ENV"));
        /**
         * Tetapkan titik akhir TCP untuk instans ApsaraMQ for RocketMQ. Dapatkan ini dari halaman detail instans di konsol ApsaraMQ for RocketMQ.
         */
        properties.put(PropertyKeyConst.NAMESRV_ADDR, "XXXX");
        /**
         * Saat menggunakan konsumen ApsaraMQ for RocketMQ untuk memproses notifikasi status klien MQTT, berlanggananlah ke topik notifikasi status.
         */
        final String parentTopic = "GID_XXXX_MQTT";
        /**
         * Data status klien. Di lingkungan produksi, kami menyarankan menggunakan penyimpanan persisten eksternal seperti database atau Redis untuk menyimpan informasi ini.
         * Hal ini mencegah kehilangan data jika aplikasi dimulai ulang. Contoh ini menggunakan implementasi in-memory sederhana.
         */
        MqttClientStatusStore mqttClientStatusStore = new MemoryHashMapStoreImpl();
        Consumer consumer = ONSFactory.createConsumer(properties);
        /**
         * Contoh ini hanya memproses apakah klien sedang online, sehingga kami hanya perlu memperhatikan event 'connect' dan 'tcpclean'.
         */
        consumer.subscribe(parentTopic, "connect||tcpclean", new MqttClientStatusNoticeListener(mqttClientStatusStore));
        consumer.start();
        String clientId = "GID_XXXXX@@@XXXXX";
        while (true) {
            System.out.println("ClientStatus :" + checkClientOnline(clientId, mqttClientStatusStore));
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    /**
     * Logika untuk memproses notifikasi koneksi dan pemutusan.
     * Dalam penerapan nyata, aplikasi yang mengonsumsi notifikasi status mungkin diterapkan di beberapa mesin.
     * Oleh karena itu, data status online klien harus dipertahankan di penyimpanan eksternal bersama seperti database atau Redis.
     * Selain itu, Anda harus menangani idempotensi pesan untuk mencegah error mesin keadaan akibat 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"));
                /**
                 * Pertama, simpan event baru tersebut.
                 */
                mqttClientStatusStore.addEvent(clientId, channelId, eventType, event);
                /**
                 * Baca daftar event untuk channel saat ini.
                 */
                Set<ClientStatusEvent> events = mqttClientStatusStore.getEvent(clientId, channelId);
                if (events == null || events.isEmpty()) {
                    return Action.CommitMessage;
                }
                /**
                 * Jika daftar event telah menerima event online dan offline, maka channel saat ini telah terputus.
                 * Anda kemudian dapat menghapus data untuk channel ini.
                 */
                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;
        }
    }
    /**
     * Periksa apakah ID klien memiliki koneksi TCP aktif berdasarkan tabel status.
     * 1. Jika tidak ada peta channel, klien pasti offline.
     * 2. Jika peta channel tidak kosong, periksa apakah data channel hanya berisi event online. Jika ya, artinya ada koneksi aktif dan klien sedang online.
     * Jika semua channel memiliki event pemutusan, klien pasti offline.
     *
     * @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;
    }
}