快速使用MQTT的Java SDK收发消息（跨产品数据流入）

背景信息

本文以公网环境中的Java SDK为例说明如何将消息队列RocketMQ版数据流入至微消息队列MQTT版。

此场景下可使用多语言的第三方开源SDK来实现消息收发。更多信息，请参见SDK下载。

如上图所示，您部署在公网地域的后端应用数据需要传送至公网环境中的MQTT客户端，后端应用和MQTT客户端均通过Java语言开发。数据从消息队列RocketMQ版流入至微消息队列MQTT版是通过创建相应的数据流入规则实现。两个产品的服务端通过各自产品提供的Java SDK分别与各自的客户端实现消息收发。

说明消息队列RocketMQ版和微消息队列MQTT版的Topic不能跨地域使用，因此，本文中所有资源都应在公网地域创建。详细信息，请参见Topic地域化。

网络访问

微消息队列MQTT版同时提供了公网接入点和VPC 接入点。接入点说明如下：

在物联网和移动互联网的场景中，客户端推荐使用公网接入点接入。
VPC 接入点仅供一些特殊场景使用。因为一般而言，涉及部署在云端服务器上的应用的场景，建议使用服务端消息产品例如消息队列RocketMQ版实现。

注意客户端使用接入点连接服务时务必使用域名接入，不得直接使用域名背后的IP地址直接连接，因为IP地址随时会变化。在以下使用情况中出现的问题微消息队列MQTT版产品方概不负责：

客户端不使用域名接入而是使用IP地址接入，产品方更新了域名解析导致原有IP地址失效。
客户端网络对IP地址设置网络防火墙策略，产品方更新了域名解析后新IP地址被您的防火墙策略拦截。

本文以公网接入点为例。微消息队列MQTT版与消息队列RocketMQ版的应用场景对比和消息属性映射关系请参见以下文档：

使用流程

后端应用消息发送至MQTT客户端的使用流程如下所示。

步骤一：创建MQTT实例并获取接入点

注意轻量版实例不支持数据流入规则、数据流出规则及上下线通知规则，如需使用规则功能，请您在创建实例时选择其他类型的实例。

登录微消息队列MQTT版控制台。
在左侧导航栏单击实例列表。
在顶部菜单栏选择地域。
在实例列表页面左上角单击创建实例。
在实例售卖页面，按需选择您需要购买的实例规格，单击立即购买。
您可购买包年包月或按量付费实例，两种实例的计费信息，请参见计费说明。此处以按量付费实例为例。
在确认订单页面，选中微消息队列for loT （按量付费）服务协议，单击立即开通。
此处以按量付费实例为例。

支付完成后，即可看到以下界面，说明已购买成功。
回到微消息队列MQTT版控制台，在左侧导航栏单击实例列表，并将地域切换为您所购买的实例所对应的地域。
在实例列表页面中，单击您所购买的实例的名称或在其操作列单击详情，进入实例详情页面。
在实例详情页面单击接入点页签，即可看到实例的接入点信息，本文示例以公网接入点为例。

步骤二：创建父级Topic

MQTT协议支持多级Topic，父级Topic需在控制台或调用OpenAPI创建，子级Topic无需创建，Topic详细信息，请参见名词解释。本文以在控制台创建父级Topic为例。

登录微消息队列MQTT版控制台。
在左侧导航栏单击实例列表。
在顶部菜单栏选择地域。
在实例列表中找到目标实例，在其操作列中，选择更多 > Topic 管理。
在Topic 管理页面左上角，单击创建 Topic。
在创建Topic面板中，输入要创建的Topic名称和描述，然后在左下角单击确定。
您可以在Topic 管理页面查看刚创建的Topic。

步骤三：创建Group ID

Group ID详细信息，请参见名词解释。

登录微消息队列MQTT版控制台。
在左侧导航栏单击实例列表。
在顶部菜单栏选择地域。
在实例列表中找到目标实例，在其操作列中，选择更多 > Group 管理。
在Group 管理页面的左上角，单击创建 Group。
在创建Group面板中，输入Group ID，然后在左下角单击确定。
您可以在Group 管理页面查看刚创建的Group。

步骤四：创建数据流入规则

规则中填写的参数需与您创建的资源保持一致。

登录微消息队列MQTT版控制台。
在左侧导航栏单击实例列表。
在顶部菜单栏选择地域。
在实例列表中找到目标实例，在其操作列，选择更多 > 规则管理。
在规则管理页面左上角，单击创建规则。

在创建规则页面完成以下操作。

在配置基本信息配置向导页面，填写规则的基本信息，然后单击下一步。


参数	取值示例	说明
规则ID	111111	规则的全局唯一标识，说明如下：只能包含字母、数字、短划线（-）和下划线（_），至少包含一个字母或数字。名称长度限制在3~64字符之间，长于64字符将被自动截取。创建后无法更新。
描述	migrate from rocketmq	对规则的描述。
状态	启用	是否启用当前规则，取值说明如下：启用停用
规则类型	数据流入	创建的规则类型，取值说明如下：数据流出：用于将微消息队列MQTT版的数据导出至其他阿里云产品。详细信息，请参见跨云产品的数据流出。数据流入：用于将其他阿里云产品的数据导入至微消息队列MQTT版。详细信息，请参见跨云产品数据流入。上下线通知：用于将获取的微消息队列MQTT版客户端上下线事件数据导出至其他阿里云产品。详细信息，请参见MQTT客户端上下线事件数据流出。

在配置规则源配置向导页面，配置数据源，然后单击下一步。


参数	取值示例	说明
源服务类型	消息队列 RocketMQ 版	指定您需将哪个源云产品的数据流入至微消息队列MQTT版。说明当前仅支持消息队列RocketMQ版。
RocketMQ 实例	MQ_INST_13801563067*****_BbyOD2jQ	指定源云产品的实例ID，即消息队列RocketMQ版的实例ID。说明仅支持选择和微消息队列MQTT版实例为同一地域的云产品实例。
Topic	TopicA	指定源云产品的资源键值，即消息队列RocketMQ版的Topic。TopicA的消息将流转至目标微消息队列MQTT版的Topic。

在配置规则目标配置向导页面，配置数据的流转目标，然后单击创建。


参数	取值示例	说明
Topic	TopicB	指定您需要将其他源云产品的数据导入至微消息队列MQTT版的哪个目标Topic。

您可以在规则管理的规则列表查看到刚创建的数据流入规则。

步骤五：调用Java SDK收发消息

下载第三方的开源Java SDK。下载地址为Eclipse Paho Java Client。
下载阿里云微消息队列MQTT版的Java SDK的Demo示例作为您代码开发的参考。下载地址为mqtt-java-demo。
解压该Demo工程包至您指定的文件夹。

在Intellij IDEA中，导入解压后的文件以创建相应的工程，并确认pom.xml中已包含以下依赖。

<dependencies>
        <dependency>
            <groupId>commons-codec</groupId>
            <artifactId>commons-codec</artifactId>
            <version>1.10</version>
        </dependency>
        <dependency>
            <groupId>org.eclipse.paho</groupId>
            <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
            <version>1.2.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.httpcomponents</groupId>
            <artifactId>httpclient</artifactId>
            <version>4.5.2</version>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.48</version>
        </dependency>
        <dependency>
            <groupId>com.aliyun.openservices</groupId>
            <artifactId>ons-client</artifactId>
            <version>1.8.5.Final</version>
        </dependency>
        <dependency>
            <groupId>com.aliyun</groupId>
            <artifactId>aliyun-java-sdk-onsmqtt</artifactId>
            <version>1.0.3</version>
        </dependency>
        <dependency>
            <groupId>com.aliyun</groupId>
            <artifactId>aliyun-java-sdk-core</artifactId>
            <version>4.5.0</version>
        </dependency>
</dependencies>

说明 ons-client的版本信息，请参见版本说明。

在RocketMQSendMessageToMQ4IoT.java类中，按代码注释说明填写相应参数，主要涉及步骤一至步骤三所创建MQTT资源以及您在消息队列RocketMQ版创建的相应资源，然后执行Main函数运行代码实现消息收发。

示例代码如下。

package com.aliyun.openservices.lmq.example.demo;

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;

/**
 * 阿里云生产环境中，消息队列RocketMQ版除了公网地域，其他地域的实例不允许在本地使用，必须在对应地域的ECS机器上部署使用。
 */
public class RocketMQSendMessageToMQ4IoT {
    public static void main(String[] args) throws Exception {
        /**
         * 初始化消息队列RocketMQ版发送客户端，实际业务中一般部署在服务端应用中。
         */
        Properties properties = new Properties();
        /**
         * 设置消息队列RocketMQ版Group ID，在消息队列RocketMQ版控制台创建。
         */
        properties.setProperty(PropertyKeyConst.GROUP_ID, "GID-XXXXX");
        /**
         * AccessKey ID，阿里云身份验证，在阿里云RAM控制台创建。
         */
        properties.put(PropertyKeyConst.AccessKey, "XXXX");
        /**
         * AccessKey Secret，阿里云身份验证，在阿里云RAM控制台创建。仅在签名鉴权模式下需要设置。
         */
        properties.put(PropertyKeyConst.SecretKey, "XXXX");
        /**
         * 设置TCP接入点，该接入点为消息队列RocketMQ版实例的接入点。进入消息队列RocketMQ版控制台实例详情页面获取。
         */
        properties.put(PropertyKeyConst.NAMESRV_ADDR, "XXXX");
        /**
         * 设置消息队列RocketMQ版的Topic，在消息队列RocketMQ版控制台创建。
         * 消息队列RocketMQ版和微消息队列MQTT配合使用时，RocketMQ客户端仅操作一级Topic。
         */
        final String parentTopic = "XXXXX";
        Producer producer = ONSFactory.createProducer(properties);
        producer.start();
        //////////////////////////////////////////////////////////////////////////////////////////////////////////////////
        /**
         * 初始化微消息队列MQTT接收客户端，实际业务中微消息队列MQTT一般部署在移动终端环境。
         */

        /**
         * 您在控制台创建的微消息队列MQTT的实例ID。
         */
        String instanceId = "XXXXX";
        /**
         * 设置接入点，进入微消息队列MQTT版控制台实例详情页面获取。
         */
        String endPoint = "XXXXXX.mqtt.aliyuncs.com";
        /**
         * AccessKey ID，阿里云身份验证，在阿里云RAM控制台创建。
         */
        String accessKey = "XXXX";
        /**
         * AccessKey Secret，阿里云身份验证，在阿里云RAM控制台创建。仅在签名鉴权模式下需要设置。
         */
        String secretKey = "XXXX";
        /**
         * MQTT客户端ID，由业务系统分配，需要保证每个TCP连接都不一样，保证全局唯一，如果不同的客户端对象（TCP连接）使用了相同的clientId会导致连接异常断开。
         * clientId由两部分组成，格式为GroupID@@@DeviceID，其中GroupID在微消息队列MQTT版控制台创建，DeviceID由业务方自己设置，clientId总长度不得超过64个字符。
         */
        String clientId = "GID_XXXX@@@XXXXX";
        /**
         * 微消息队列MQTT版支持子级Topic，用来做自定义的过滤，此处为示例，可以填写任何字符串。
         * 需要注意的是，完整的Topic长度不得超过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();
        /**
         * 客户端协议和端口。客户端使用的协议和端口必须匹配，如果是SSL加密则设置ssl://endpoint:8883。
         */
        final MqttClient mqttClient = new MqttClient("tcp://" + endPoint + ":1883", clientId, memoryPersistence);
        /**
         * 设置客户端发送超时时间，防止无限阻塞。
         */
        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) {
                /**
                 * 客户端连接成功后就需要尽快订阅需要的Topic。
                 */
                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 {
                /**
                 * 消费消息的回调接口，需要确保该接口不抛异常，该接口运行返回即代表消息消费成功。
                 * 消费消息需要保证在规定时间内完成，如果消费耗时超过服务端约定的超时时间，对于可靠传输的模式，服务端可能会重试推送，业务需要做好幂等去重处理。
                 */
                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++) {
            /**
             * 使用RocketMQ客户端发消息给MQTT客户端时，Topic指定为一级父Topic，Tag指定为MQ2MQTT。
             */
            Message msg = new Message(parentTopic, "MQ2MQTT", "hello mq send mqtt msg".getBytes());
            /**
             * 使用RocketMQ客户端发消息给MQTT客户端时，可以通过MqttSecondTopic属性设置MQTT的子级Topic属性。
             */
            msg.putUserProperties(PropertyKeyConst.MqttSecondTopic, subTopic);
            SendResult result = producer.send(msg);
            System.out.println(result);
            /**
             * 发送P2P消息，设置子级Topic。 
             */
            msg.putUserProperties(PropertyKeyConst.MqttSecondTopic, "/p2p/" + clientId);
            result = producer.send(msg);
            System.out.println(result);
        }
        Thread.sleep(Long.MAX_VALUE);

    }

}

结果验证

完成消息收发后，您可在微消息队列MQTT版控制台查询轨迹以验证消息是否发送并接收成功。详细信息，请参见消息轨迹查询。

前提条件

背景信息

网络访问

使用流程

步骤一：创建MQTT实例并获取接入点

步骤二：创建父级Topic

步骤三：创建Group ID

步骤四：创建数据流入规则

步骤五：调用Java SDK收发消息

结果验证

更多信息