ApsaraMQ for MQTT：Impor data dari ApsaraMQ for RocketMQ ke ApsaraMQ for MQTT

import com.aliyun.openservices.lmq.example.util.ConnectionOptionWrapper;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.SendResult;
import java.util.Properties;
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.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

public class RocketMQSendMessageToMQ4IoT {
    public static void main(String[] args) throws Exception {
        /**
         * Inisialisasi klien ApsaraMQ for RocketMQ sebagai pengirim. Dalam sebagian besar skenario bisnis, pengirim ditempatkan di aplikasi backend. 
         */
        Properties properties = new Properties();
        /**
         * ID grup yang Anda buat di konsol ApsaraMQ for RocketMQ. 
         */
        properties.setProperty(PropertyKeyConst.GROUP_ID, "GID-XXXXX");
        /**
         * ID AccessKey yang Anda buat di konsol Resource Access Management (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 Anda untuk 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 di variabel lingkungan. 
         */
        properties.put(PropertyKeyConst.AccessKey, System.getenv("MQTT_AK_ENV"));
        /**
         * Rahasia AccessKey yang Anda buat di konsol RAM Alibaba Cloud untuk otentikasi identitas. Rahasia AccessKey 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. Anda dapat membuat topik di konsol ApsaraMQ for RocketMQ. 
         * Saat bertukar data antara ApsaraMQ for RocketMQ dan ApsaraMQ for MQTT, hanya topik induk yang dapat digunakan oleh klien ApsaraMQ for RocketMQ. 
         */
        final String parentTopic = "XXXXX";
        Producer producer = ONSFactory.createProducer(properties);
        producer.start();
        //////////////////////////////////////////////////////////////////////////////////////////////////////////////////
        /**
         * Inisialisasi klien ApsaraMQ for MQTT sebagai penerima. Dalam sebagian besar skenario bisnis, penerima ditempatkan di terminal seluler. 
         */

        /**
         * ID instansi ApsaraMQ for MQTT yang Anda buat di konsol ApsaraMQ for MQTT. 
         */
        String instanceId = "XXXXX";
        /**
         * Endpoint yang ingin Anda gunakan untuk mengakses instansi ApsaraMQ for MQTT. Anda dapat memperoleh endpoint di halaman Detail Instansi di konsol ApsaraMQ for MQTT. 
         */
        String endPoint = "XXXXXX.mqtt.aliyuncs.com";
        /**
         * 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 Anda untuk 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 di variabel lingkungan. 
         */
        String 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. 
         */
        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 akan terjadi dan koneksi akan ditutup secara tak terduga. 
         * ID klien terdiri dari ID grup dan ID perangkat dan dalam format GroupID@@@DeviceID. ID grup adalah ID grup yang Anda buat di konsol ApsaraMQ for MQTT. ID perangkat adalah ID kustom yang Anda tentukan. ID klien tidak boleh melebihi 64 karakter panjangnya. 
         */
        String clientId = "GID_XXXX@@@XXXXX";
        /**
         * ApsaraMQ for MQTT memungkinkan Anda menggunakan subtopik untuk memfilter pesan. Anda dapat menentukan string sebagai nama subtopik. 
         * Nilai parameter mq4IotTopic dapat mencapai hingga 128 karakter panjangnya. 
         */
        final String subTopic = "/testMq4Iot";
        final String mq4IotTopic = parentTopic + subTopic;
        /**
         * Tingkat kualitas layanan (QoS) untuk transmisi pesan. Nilai valid: 0, 1, dan 2. Untuk informasi lebih lanjut, lihat Istilah. 
         */
        final int qosLevel = 0;
        ConnectionOptionWrapper connectionOptionWrapper = new ConnectionOptionWrapper(instanceId, accessKey, secretKey, clientId);
        final MemoryPersistence memoryPersistence = new MemoryPersistence();
        /**
         * Protokol dan port yang digunakan oleh klien ApsaraMQ for MQTT. Protokol dan port yang digunakan oleh klien ApsaraMQ for MQTT harus cocok. 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 saat 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) {
                /**
                 * Topik yang harus dilanggan konsumen sesegera mungkin setelah koneksi klien didirikan. 
                 */
                System.out.println("koneksi berhasil");
                executorService.submit(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            final String topicFilter[] = {mq4IotTopic};
                            final int[] qos = {qosLevel};
                            mqttClient.subscribe(topicFilter, qos);
                        } catch (MqttException e) {
                            e.printStackTrace();
                        }
                    }
                });
            }

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

            @Override
            public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
                /**
                 * Callback yang dipanggil untuk mengonsumsi pesan yang diterbitkan. Pastikan callback tidak melempar pengecualian. Jika respons dikembalikan untuk callback, pesan dikonsumsi. 
                 * Pesan harus dikonsumsi dalam periode waktu tertentu. Jika pesan tidak dikonsumsi dalam periode timeout yang ditentukan oleh broker ApsaraMQ for MQTT, broker mungkin mencoba mengirim ulang pesan. Pastikan deduplikasi dilakukan untuk memastikan idempotensi untuk konsumsi pesan. 
                 */
                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());
        for (int i = 0; i < 10; i++) {
            /**
             * Tentukan topik induk sebagai topik dan MQ2MQTT sebagai tag saat menggunakan klien ApsaraMQ for RocketMQ untuk mengirim pesan ke klien ApsaraMQ for MQTT. 
             */
            Message msg = new Message(parentTopic, "MQ2MQTT", "hello mq send mqtt msg".getBytes());
            /**
             * Anda dapat menggunakan parameter MqttSecondTopic untuk menentukan subtopik saat menggunakan klien ApsaraMQ for RocketMQ untuk mengirim pesan ke klien ApsaraMQ for MQTT. 
             */
            msg.putUserProperties(PropertyKeyConst.MqttSecondTopic, subTopic);
            SendResult result = producer.send(msg);
            System.out.println(result);
//            /**
//             * Kirim pesan P2P dan tentukan subtopik. 
//             */
//            msg.putUserProperties(PropertyKeyConst.MqttSecondTopic, "/p2p/" + clientId);
//            result = producer.send(msg);
//            System.out.println(result);
        }
        Thread.sleep(Long.MAX_VALUE);

    }

}