ApsaraMQ for MQTT:ApsaraMQ for MQTT から 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 {
        /**
         * レシーバーとして ApsaraMQ for RocketMQ クライアントを初期化します。ほとんどのビジネスシナリオでは、レシーバーはバックエンドアプリケーションにデプロイされます。
         */
        Properties properties = new Properties();
        /**
         * ApsaraMQ for RocketMQ コンソールで作成したグループの ID。
         */
        properties.setProperty(PropertyKeyConst.GROUP_ID, "GID-XXXXX");
        
        /**
         * アイデンティティ認証のために 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"));
        /**
         * アイデンティティ認証のために Alibaba Cloud RAM コンソールで作成した AccessKey シークレット。このパラメーターは、署名認証モードを使用する場合にのみ必須です。
         */
        properties.put(PropertyKeyConst.SecretKey, System.getenv("MQTT_SK_ENV"));
      
         /**
         * ApsaraMQ for RocketMQ インスタンスへのアクセスに使用する TCP エンドポイント。TCP エンドポイントは、ApsaraMQ for RocketMQ コンソールの [インスタンスの詳細] ページで取得できます。
         */
        properties.put(PropertyKeyConst.NAMESRV_ADDR, "http://xxxxx.XXXXX.mq-internet.aliyuncs.com");
        /**
         * ApsaraMQ for RocketMQ コンソールで作成したトピック。
         * ApsaraMQ for RocketMQ と ApsaraMQ for MQTT 間でデータを交換する場合、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();
        //////////////////////////////////////////////////////////////////////////////////////////////////////////////////
        /**
         * センダーとして ApsaraMQ for MQTT クライアントを初期化します。ほとんどのビジネスシナリオでは、センダーはモバイル端末にデプロイされます。
         */

        /**
         * コンソールで作成した ApsaraMQ for MQTT インスタンスの ID。
         */
        String instanceId = "XXXXX";
         /**
         * ApsaraMQ for MQTT インスタンスへのアクセスに使用するエンドポイント。エンドポイントは、ApsaraMQ for MQTT コンソールの [インスタンスの詳細] ページで取得できます。
         */
        String endPoint = "XXXXXX.mqtt.aliyuncs.com";
        /**
         * アイデンティティ認証のために 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");
        /**
         * アイデンティティ認証のために 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 mq4IotTopic = parentTopic + "/" + "testMq4Iot";
        /**
         * メッセージ送信のサービス品質 (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);
        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 {
            }

            @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++) {
            MqttMessage message = new MqttMessage("hello mq4Iot pub sub msg".getBytes());
            message.setQos(qosLevel);
            /**
             * メッセージの送信先となるトピック。送信されるメッセージが通常のメッセージの場合、トピックはコンシューマーがサブスクライブするトピック、またはワイルドカードを使用して一致させることができるトピックである必要があります。
             */
            mqttClient.publish(mq4IotTopic, message);
        }
        Thread.sleep(Long.MAX_VALUE);

    }

}