ApsaraMQ for MQTT:ApsaraMQ for RocketMQ から 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 {
        /**
         * 送信者として ApsaraMQ for RocketMQ クライアントを初期化します。ほとんどのビジネスシナリオでは、送信者はバックエンドアプリケーションにデプロイされます。
         */
        Properties properties = new Properties();
        /**
         * ApsaraMQ for RocketMQ コンソールで作成したグループの ID。
         */
        properties.setProperty(PropertyKeyConst.GROUP_ID, "GID-XXXXX");
        /**
         * ID 認証のために Alibaba Cloud Resource Access Management (RAM) コンソールで作成した AccessKey ID。
         * Alibaba Cloud アカウントの AccessKey ペアは、すべての API 操作に対する権限を持っています。セキュリティリスクを防ぐため、RAM ユーザーを使用して API 操作を呼び出したり、日常的な O&M を実行することをお勧めします。
         * AccessKey ペアをプロジェクトコードに保存しないことを強くお勧めします。そうしないと、AccessKey ペアが漏洩し、アカウント内のすべてのリソースが潜在的なセキュリティリスクにさらされる可能性があります。
         * この例では、AccessKey ID と AccessKey シークレットは環境変数に格納されています。
         */
        properties.put(PropertyKeyConst.AccessKey, System.getenv("MQTT_AK_ENV"));
        /**
         * ID 認証のために Alibaba Cloud RAM コンソールで作成した AccessKey シークレット。AccessKey シークレットは、署名認証モードを使用する場合にのみ必要です。
         */
        properties.put(PropertyKeyConst.SecretKey, System.getenv("MQTT_SK_ENV"));
        /**
         * ApsaraMQ for RocketMQ インスタンスにアクセスするために使用される TCP エンドポイント。TCP エンドポイントは、ApsaraMQ for RocketMQ コンソールの [インスタンスの詳細] ページで取得できます。
         */
        properties.put(PropertyKeyConst.NAMESRV_ADDR, "XXXX");
        /**
         * ApsaraMQ for RocketMQ トピック。ApsaraMQ for RocketMQ コンソールでトピックを作成できます。
         * ApsaraMQ for RocketMQ と ApsaraMQ for MQTT 間でデータを交換する場合、ApsaraMQ for RocketMQ クライアントは親トピックのみ使用できます。
         */
        final String parentTopic = "XXXXX";
        Producer producer = ONSFactory.createProducer(properties);
        producer.start();
        //////////////////////////////////////////////////////////////////////////////////////////////////////////////////
        /**
         * 受信者として ApsaraMQ for MQTT クライアントを初期化します。ほとんどのビジネスシナリオでは、受信者はモバイル端末にデプロイされます。
         */

        /**
         * ApsaraMQ for MQTT コンソールで作成した ApsaraMQ for MQTT インスタンスの ID。
         */
        String instanceId = "XXXXX";
        /**
         * ApsaraMQ for MQTT インスタンスにアクセスするために使用するエンドポイント。エンドポイントは、ApsaraMQ for MQTT コンソールの [インスタンスの詳細] ページで取得できます。
         */
        String endPoint = "XXXXXX.mqtt.aliyuncs.com";
        /**
         * ID 認証のために Alibaba Cloud RAM コンソールで作成した AccessKey ID。
         * Alibaba Cloud アカウントの AccessKey ペアは、すべての API 操作に対する権限を持っています。セキュリティリスクを防ぐため、RAM ユーザーを使用して API 操作を呼び出したり、日常的な O&M を実行することをお勧めします。
         * AccessKey ペアをプロジェクトコードに保存しないことを強くお勧めします。そうしないと、AccessKey ペアが漏洩し、アカウント内のすべてのリソースが潜在的なセキュリティリスクにさらされる可能性があります。
         * この例では、AccessKey ID と AccessKey シークレットは環境変数に格納されています。
         */
        String accessKey = System.getenv("MQTT_AK_ENV");
        /**
         * ID 認証のために Alibaba Cloud RAM コンソールで作成した AccessKey シークレット。このパラメーターは、署名認証モードを使用する場合にのみ必要です。
         */
        String secretKey = System.getenv("MQTT_SK_ENV");
        /**
         * システムが ApsaraMQ for MQTT クライアントに割り当てるグローバルに一意の ID。クライアント ID は、TCP 接続ごとに異なる必要があります。複数の TCP 接続が同じクライアント ID を使用すると、例外が発生し、接続が予期せず閉じられます。
         * クライアント ID は、グループ ID とデバイス ID で構成され、GroupID@@@DeviceID 形式です。グループ ID は、ApsaraMQ for MQTT コンソールで作成したグループの ID です。デバイス ID は、指定するカスタム ID です。クライアント ID は 64 文字以下にする必要があります。
         */
        String clientId = "GID_XXXX@@@XXXXX";
        /**
         * ApsaraMQ for MQTT では、サブトピックを使用してメッセージをフィルタリングできます。サブトピックの名前として文字列を指定できます。
         * mq4IotTopic パラメーターの値は、最大 128 文字まで指定できます。
         */
        final String subTopic = "/testMq4Iot";
        final String mq4IotTopic = parentTopic + subTopic;
        /**
         * メッセージ送信のサービス品質 (QoS) レベル。有効な値: 0、1、2。詳細については、「用語」をご参照ください。
         */
        final int qosLevel = 0;
        ConnectionOptionWrapper connectionOptionWrapper = new ConnectionOptionWrapper(instanceId, accessKey, secretKey, clientId);
        final MemoryPersistence memoryPersistence = new MemoryPersistence();
        /**
         * ApsaraMQ for MQTT クライアントが使用するプロトコルとポート。ApsaraMQ for MQTT クライアントが使用するプロトコルとポートは一致する必要があります。Secure Sockets Layer (SSL) 暗号化を使用する場合は、プロトコルとポートとして ssl://endpoint:8883 を指定します。
         */
        final MqttClient mqttClient = new MqttClient("tcp://" + endPoint + ":1883", clientId, memoryPersistence);
        /**
         * ApsaraMQ for MQTT クライアントが応答を待つタイムアウト期間。タイムアウト期間を設定することで、ApsaraMQ for MQTT クライアントが無期限に応答を待つことを防ぎます。
         */
        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");
                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 {
                /**
                 * パブリッシュされたメッセージを消費するために呼び出されるコールバック。コールバックが例外をスローしないようにしてください。コールバックに対して応答が返された場合、メッセージは消費されます。
                 * メッセージは特定の期間内に消費される必要があります。ApsaraMQ for MQTT ブローカーによって指定されたタイムアウト期間内にメッセージが消費されない場合、ブローカーはメッセージの再送信を試みる可能性があります。メッセージ消費の冪等性を確保するために、重複排除が実行されていることを確認してください。
                 */
                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());
        for (int i = 0; i < 10; i++) {
            /**
             * ApsaraMQ for RocketMQ クライアントを使用して ApsaraMQ for MQTT クライアントにメッセージを送信する場合は、トピックとして親トピックを、タグとして MQ2MQTT を指定します。
             */
            Message msg = new Message(parentTopic, "MQ2MQTT", "hello mq send mqtt msg".getBytes());
            /**
             * ApsaraMQ for RocketMQ クライアントを使用して ApsaraMQ for MQTT クライアントにメッセージを送信する場合は、MqttSecondTopic パラメーターを使用してサブトピックを指定できます。
             */
            msg.putUserProperties(PropertyKeyConst.MqttSecondTopic, subTopic);
            SendResult result = producer.send(msg);
            System.out.println(result);
//            /**
//             * P2P メッセージを送信し、サブトピックを指定します。
//             */
//            msg.putUserProperties(PropertyKeyConst.MqttSecondTopic, "/p2p/" + clientId);
//            result = producer.send(msg);
//            System.out.println(result);
        }
        Thread.sleep(Long.MAX_VALUE);

    }

}