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

import com.aliyun.openservices.lmq.example.util.ConnectionOptionWrapper;
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.Properties;
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 MQ4IoTSendMessageToRocketMQ {
    public static void main(String[] args) throws Exception {
        /**
         * Inisialisasi klien ApsaraMQ for RocketMQ sebagai penerima. Dalam sebagian besar skenario bisnis, penerima ditempatkan pada 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 O&M 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 autentikasi tanda tangan. 
         */
        properties.put(PropertyKeyConst.SecretKey, System.getenv("MQTT_SK_ENV"));
      
         /**
         * Titik akhir TCP yang ingin Anda gunakan untuk mengakses instansi ApsaraMQ for RocketMQ. Anda dapat memperoleh titik akhir TCP di halaman Detail Instansi di konsol ApsaraMQ for RocketMQ. 
         */
        properties.put(PropertyKeyConst.NAMESRV_ADDR, "http://xxxxx.XXXXX.mq-internet.aliyuncs.com");
        /**
         * Topik yang Anda buat di konsol ApsaraMQ for RocketMQ. 
         * Saat Anda 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";
        Consumer consumer = ONSFactory.createConsumer(properties);
        consumer.subscribe(parentTopic, "*", new MessageListener() {
            public Action consume(Message message, ConsumeContext consumeContext) {
                System.out.println("recv msg:" + message);
                return Action.CommitMessage;
            }
        });
        consumer.start();
        //////////////////////////////////////////////////////////////////////////////////////////////////////////////////
        /**
         * Inisialisasi klien ApsaraMQ for MQTT sebagai pengirim. Dalam sebagian besar skenario bisnis, pengirim ditempatkan pada terminal seluler. 
         */

        /**
         * ID instansi ApsaraMQ for MQTT yang Anda buat di konsol. 
         */
        String instanceId = "XXXXX";
         /**
         * Titik akhir yang ingin Anda gunakan untuk mengakses instansi ApsaraMQ for MQTT. Anda dapat memperoleh titik akhir 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 O&M 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. 
         */
        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 autentikasi tanda tangan. 
         */
        String secretKey = System.getenv("MQTT_SK_ENV");
        /**
         * ID unik global yang sistem tetapkan ke 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. 
         * ID klien terdiri dari ID grup dan ID perangkat dan berformat 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 mq4IotTopic = parentTopic + "/" + "testMq4Iot";
        /**
         * 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);
        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");
            }

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

            @Override
            public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
            }

            @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++) {
            MqttMessage message = new MqttMessage("hello mq4Iot pub sub msg".getBytes());
            message.setQos(qosLevel);
            /**
             * Topik ke mana pesan dikirim. Jika pesan yang dikirim adalah pesan normal, topik harus merupakan topik yang dilanggan konsumen atau topik yang dapat dicocokkan menggunakan wildcard. 
             */
            mqttClient.publish(mq4IotTopic, message);
        }
        Thread.sleep(Long.MAX_VALUE);

    }

}