ApsaraMQ for MQTT:ApsaraMQ for MQTT から ApsaraMQ for RocketMQ へのデータエクスポート

サンプルコードは、トピックをサブスクライブする RocketMQ コンシューマーを初期化し、サブトピックに 10 件のテストメッセージをパブリッシュする MQTT クライアントを作成します。その後、データアウトバウンドルールによってこれらのメッセージが 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 {
        // ============================================================
        // 1. RocketMQ コンシューマー (バックエンドアプリケーション) を初期化
        // ============================================================
        Properties properties = new Properties();

        // RocketMQ グループ ID
        properties.setProperty(PropertyKeyConst.GROUP_ID, "GID-XXXXX");

        // 環境変数から認証情報を読み込み
        properties.put(PropertyKeyConst.AccessKey, System.getenv("MQTT_AK_ENV"));
        properties.put(PropertyKeyConst.SecretKey, System.getenv("MQTT_SK_ENV"));

        // RocketMQ NameServer アドレス (Instance Details ページから取得)
        properties.put(PropertyKeyConst.NAMESRV_ADDR,
            "http://xxxxx.XXXXX.mq-internet.aliyuncs.com");

        // RocketMQ トピック (MQTT-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();

        // ============================================================
        // 2. MQTT クライアント (デバイスまたはモバイルアプリ) を初期化
        // ============================================================

        // MQTT インスタンス ID
        String instanceId = "XXXXX";

        // MQTT エンドポイント (Instance Details ページから取得)
        String endPoint = "XXXXXX.mqtt.aliyuncs.com";

        // 環境変数から認証情報を読み込み
        String accessKey = System.getenv("MQTT_AK_ENV");
        String secretKey = System.getenv("MQTT_SK_ENV");

        // クライアント ID：GroupID@@@DeviceID (TCP 接続ごとに一意である必要があります)
        String clientId = "GID_XXXX@@@XXXXX";

        // サブトピック：parentTopic + "/" + サブトピック名 (最大 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();

        // プロトコルとポート：tcp://endpoint:1883 または ssl://endpoint:8883
        final MqttClient mqttClient =
            new MqttClient("tcp://" + endPoint + ":1883", clientId, memoryPersistence);

        // クライアントのタイムアウト (ミリ秒)
        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());

        // ============================================================
        // 3. 10 件のテストメッセージをパブリッシュ
        // ============================================================
        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);
    }
}