ApsaraMQ for MQTT：Kirim notifikasi status klien dari ApsaraMQ for MQTT ke 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 {
        /**
         * ID instansi ApsaraMQ for MQTT. Anda dapat memperoleh ID di konsol ApsaraMQ for MQTT setelah Anda membeli instansi.
         */
        String instanceId = "XXXXX";
        /**
         * Endpoint instansi ApsaraMQ for MQTT. Anda dapat memperoleh endpoint di konsol ApsaraMQ for MQTT setelah Anda membeli dan mengonfigurasi instansi. Anda harus menggunakan nama domain yang ditetapkan daripada alamat IP untuk terhubung ke instansi. Jika tidak, pengecualian mungkin terjadi pada objek klien. 
         */
        String endPoint = "XXXXX.mqtt.aliyuncs.com";
        /**
         * ID AccessKey, yang dapat diperoleh di konsol Resource Access Management (RAM).
         * Pasangan AccessKey akun Alibaba Cloud memiliki izin untuk semua operasi API. Untuk mencegah risiko keamanan, kami sarankan Anda menggunakan pengguna RAM untuk memanggil operasi API atau melakukan pemeliharaan rutin. 
         * Kami sangat menyarankan agar Anda tidak menyimpan pasangan AccessKey di kode proyek. Jika tidak, pasangan AccessKey mungkin bocor dan semua sumber daya di akun Anda mungkin terpapar pada risiko keamanan potensial. 
         * Dalam contoh ini, ID AccessKey dan Rahasia AccessKey disimpan dalam variabel lingkungan. Sebelum Anda menjalankan kode sampel, pastikan bahwa variabel lingkungan MQTT_AK_ENV dan MQTT_SK_ENV dikonfigurasi.
         * Contoh: 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, yang dapat diperoleh di konsol RAM. Parameter ini diperlukan hanya jika Anda menggunakan mode otentikasi tanda tangan.
         */
        String secretKey = System.getenv("MQTT_SK_ENV");
        /**
         * ID unik global yang sistem tetapkan untuk klien ApsaraMQ for MQTT. ID klien harus bervariasi berdasarkan koneksi TCP. Jika beberapa koneksi TCP menggunakan ID klien yang sama, pengecualian terjadi dan koneksi ditutup secara tak terduga. 
         * Nilai parameter clientId dalam format GroupID@@@DeviceId. GroupID menentukan ID grup yang Anda buat di konsol ApsaraMQ for MQTT dan DeviceId menentukan ID perangkat kustom. Nilai parameter clientId dapat mencapai hingga 64 karakter panjangnya. 
         */
        String clientId = "GID_XXXXX@@@XXXXX";

        ConnectionOptionWrapper connectionOptionWrapper = new ConnectionOptionWrapper(instanceId, accessKey, secretKey, clientId);
        final MemoryPersistence memoryPersistence = new MemoryPersistence();
        /**
         * Protokol dan port yang digunakan oleh objek klien ApsaraMQ for MQTT harus cocok. Untuk informasi lebih lanjut, lihat https://www.alibabacloud.com/help/doc-detail/44866.htm?spm=a2c4g.11186623.6.552.25302386RcuYFB.
         Jika enkripsi Secure Sockets Layer (SSL) digunakan, tentukan ssl://endpoint:8883 sebagai protokol dan port.
         */
        final MqttClient mqttClient = new MqttClient("tcp://" + endPoint + ":1883", clientId, memoryPersistence);
        /**
         * Periode timeout selama klien ApsaraMQ for MQTT menunggu respons. Periode timeout mencegah klien ApsaraMQ for MQTT menunggu respons untuk periode waktu yang tidak terbatas.
         */
        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("koneksi berhasil");
            }

            @Override
            public void connectionLost(Throwable throwable) {
                throwable.printStackTrace();
            }

            @Override
            public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
                System.out.println("menerima pesan dari topik " + s + " , isi adalah " + new String(mqttMessage.getPayload()));
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
                System.out.println("pesan terkirim sukses topik adalah : " + 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 ApsaraMQ for RocketMQ sebagai penerima. Dalam sebagian besar skenario bisnis, penerima diterapkan di aplikasi backend. 
         */
        Properties properties = new Properties();
        /**
         * ID grup yang Anda buat di konsol ApsaraMQ for RocketMQ. 
         */
        properties.setProperty(PropertyKeyConst.GROUP_ID, "GID_XXXX");
        /**
         * ID AccessKey yang Anda buat di konsol RAM Alibaba Cloud untuk otentikasi identitas. 
         * Pasangan AccessKey akun Alibaba Cloud memiliki izin untuk semua operasi API. Untuk mencegah risiko keamanan, kami sarankan Anda menggunakan pengguna RAM untuk memanggil operasi API atau melakukan pemeliharaan rutin. 
         * Kami sangat menyarankan agar Anda tidak menyimpan pasangan AccessKey di kode proyek. Jika tidak, pasangan AccessKey mungkin bocor dan semua sumber daya di akun Anda mungkin terpapar pada risiko keamanan potensial. 
         * Dalam contoh ini, ID AccessKey dan Rahasia AccessKey disimpan dalam variabel lingkungan. 
         */
        properties.put(PropertyKeyConst.AccessKey, System.getenv("MQTT_AK_ENV"));
        /**
         * Rahasia AccessKey yang Anda buat di konsol RAM Alibaba Cloud untuk otentikasi identitas. Parameter ini diperlukan hanya jika Anda menggunakan mode otentikasi tanda tangan. 
         */
        properties.put(PropertyKeyConst.SecretKey, System.getenv("MQTT_SK_ENV"));
        /**
         * Endpoint TCP yang digunakan untuk mengakses instansi ApsaraMQ for RocketMQ. Anda dapat memperoleh endpoint TCP di halaman Detail Instansi di konsol ApsaraMQ for RocketMQ. 
         */
        properties.put(PropertyKeyConst.NAMESRV_ADDR, "XXXX");
        /**
         * Topik ApsaraMQ for RocketMQ yang ingin Anda gunakan untuk memproses notifikasi status klien dari ApsaraMQ for MQTT. 
         */
        final String parentTopic = "GID_XXXX_MQTT";
        /**
         * Data status klien. Dalam lingkungan produksi, kami sarankan Anda menggunakan sistem penyimpanan persisten eksternal, seperti database atau sistem Redis, untuk menyimpan data status dan mencegah hilangnya data status saat aplikasi dimulai ulang. Dalam contoh ini, data status disimpan di mesin lokal. 
         */
        MqttClientStatusStore mqttClientStatusStore = new MemoryHashMapStoreImpl();
        Consumer consumer = ONSFactory.createConsumer(properties);
        /**
         * Dalam contoh ini, hanya data terkait apakah klien online diproses. Oleh karena itu, Anda hanya perlu memperhatikan acara 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 yang digunakan untuk memproses notifikasi status klien. 
     * Selama proses penyebaran aktual, aplikasi yang memproses notifikasi status mungkin diterapkan di beberapa mesin. Oleh karena itu, data status klien dapat dipertahankan di penyimpanan bersama eksternal seperti database atau sistem Redis. 
     * Jika mesin negara menerima pesan berulang kali, lakukan idempotensi pada pesan untuk mencegah kesalahan yang mungkin terjadi. 
     */
    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 acara baru terlebih dahulu. 
                 */
                mqttClientStatusStore.addEvent(clientId, channelId, eventType, event);
                /**
                 * Baca daftar acara saluran saat ini. 
                 */
                Set<ClientStatusEvent> events = mqttClientStatusStore.getEvent(clientId, channelId);
                if (events == null || events.isEmpty()) {
                    return Action.CommitMessage;
                }
                /**
                 * Jika semua acara online dan offline dalam daftar diterima dan saluran saat ini ditutup, Anda dapat menghapus data saluran. 
                 */
                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 klien ApsaraMQ for MQTT memiliki koneksi TCP aktif berdasarkan tabel saluran. 
     * 1. Jika tabel saluran kosong, klien ApsaraMQ for MQTT offline. 
     * 2. Jika tabel saluran tidak kosong, periksa apakah hanya acara online yang diterima dalam koneksi. Jika ya, koneksi aktif ada dan klien ApsaraMQ for MQTT online. 
     * Jika acara offline diterima di semua saluran, klien ApsaraMQ for MQTT 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;
    }
}