This topic describes how Message Queue for MQTT clients send messages to and receive messages from each other by using Message Queue for MQTT SDK for Java.

Prerequisites

Background information

The scenario where Message Queue for MQTT clients send messages to and receive messages from each other is the simplest scenario of Message Queue for MQTT. In this scenario, both the producer and consumer are distributed in the mobile environment. Each Message Queue for MQTT client uses the client SDK of Message Queue for MQTT to connect to the Message Queue for MQTT broker to send and receive messages.

Message interaction process

This topic describes how to use Message Queue for MQTT SDK for Java to implement messaging between Message Queue for MQTT clients over the Internet.

Endpoints

To allow a Message Queue for MQTT client to connect to Message Queue for MQTT, you must specify an endpoint of your Message Queue for MQTT instance in the code when you call the client SDK. Then, the Message Queue for MQTT client can use the endpoint to connect to Message Queue for MQTT. For more information about how to obtain endpoints, see Create an instance and obtain an endpoint of the instance.

Message Queue for MQTT provides public endpoints and virtual private cloud (VPC) endpoints.
  • A public endpoint is an IP address for access from the Internet, which is applicable to IoT and mobile Internet scenarios.

    Format of public endpoints: ID of the Message Queue for MQTT instance.mqtt.aliyuncs.com

  • A VPC endpoint is an IP address for access from a private network in the cloud, which is usually used by backend applications to connect to Message Queue for MQTT.

    Format of VPC endpoints: ID of the Message Queue for MQTT instance-internal-vpc.mqtt.aliyuncs.com

Notice To connect a Message Queue for MQTT client to Message Queue for MQTT by using an endpoint, use the domain name instead of the IP address because the IP address is not fixed. The service team of Message Queue for MQTT is not liable for direct or indirect faults or losses that arise in the following scenarios:
  • Your Message Queue for MQTT client uses an IP address instead of a domain name to access the service. The original IP address becomes invalid after the service team of Message Queue for MQTT updates domain name resolution.
  • A firewall policy on IP addresses is set in the network on which your Message Queue for MQTT client is running. New IP addresses are blocked due to the firewall policy after the service team of Message Queue for MQTT updates domain name resolution.

Call SDK for Java to send and receive messages

  1. Download a third-party open source SDK for Java. Download address: Eclipse Paho Java Client.
  2. Download the demo of Alibaba Cloud Message Queue for MQTT SDK for Java for reference during code development. Download address: mqtt-java-demo.
  3. Decompress this demo project package to a specific folder.
  4. In IntelliJ IDEA, import the decompressed files to create a project and check whether the following dependencies are included in the pom.xml file:
    <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</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>
  5. In the MQ4IoTSendMessageToMQ4IoTUseSignatureMode.java class, set parameters as instructed in code annotations. Most of these parameters can be set based on the information of the Message Queue for MQTT resources created in advance. For more information, see Create resources. Use the main() function to run the sample code to send and receive messages.
    Sample code:
    package com.aliyun.openservices.lmq.example.demo;
    
    import com.aliyun.openservices.lmq.example.util.ConnectionOptionWrapper;
    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 MQ4IoTSendMessageToMQ4IoTUseSignatureMode {
        public static void main(String[] args) throws Exception {
            /**
             * The ID of the Message Queue for MQTT instance that you created in the Message Queue for MQTT console. 
             */
            String instanceId = "XXXXX";
            /**
             * The endpoint of the Message Queue for MQTT instance. You can obtain the endpoint on the Instance Details page in the Message Queue for MQTT console. 
             */
            String endPoint = "XXXXX.mqtt.aliyuncs.com";
            /**
             * The AccessKey ID that you created in the Alibaba Cloud RAM console for identity authentication. 
             */
            String accessKey = "XXXXX";
            /**
             * The AccessKey secret that you created in the Alibaba Cloud RAM console for identity authentication. The AccessKey secret is required only for signature verification. 
             */
            String secretKey = "XXXXX";
            /**
             * The globally unique ID that the system assigns to the Message Queue for MQTT client. The client ID must be different in different TCP connections. If multiple TCP connections use the same client ID, an exception occurs and the connections are closed. 
             * The value of the clientId parameter is in the format of GroupID@@@DeviceID. GroupID is the group ID that you created in the Message Queue for MQTT console and DeviceID is the custom ID of the device. The value of the clientId parameter can be up to 64 characters in length. 
             */
            String clientId = "GID_XXXXX@@@XXXXX";
            /**
             * The parent topic that you created in the Message Queue for MQTT console. 
             * If you specify a topic that does not exist or a topic that the Message Queue for MQTT client is not authorized to access, the Message Queue for MQTT broker closes the connection. 
             */
            final String parentTopic = "XXXXX";
            /**
             * Message Queue for MQTT allows you to use a subtopic to filter messages. You can specify a string as a subtopic, as shown in the following code. 
             * The value of the mq4IotTopic parameter can be up to 128 characters in length. 
             */
            final String mq4IotTopic = parentTopic + "/" + "testMq4Iot";
            /**
             * The quality of service (QoS) level in message transmission. Valid values: 0, 1, and 2. For more information, see Terms. 
             */
            final int qosLevel = 0;
            ConnectionOptionWrapper connectionOptionWrapper = new ConnectionOptionWrapper(instanceId, accessKey, secretKey, clientId);
            final MemoryPersistence memoryPersistence = new MemoryPersistence();
            /**
             * The protocol and port that are used by the Message Queue for MQTT client. The protocol and port that are used by the Message Queue for MQTT client must match. If SSL encryption is used, the protocol and port must be specified as ssl://endpoint:8883. 
             */
            final MqttClient mqttClient = new MqttClient("tcp://" + endPoint + ":1883", clientId, memoryPersistence);
            /**
             * The timeout period for the Message Queue for MQTT client to wait for a response. The timeout period can prevent the Message Queue for MQTT client from keeping waiting for a response. 
             */
            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) {
                    /**
                     * After a connection is established, the Message Queue for MQTT client must subscribe to the required topic as soon as possible. 
                     */
                    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 {
                    /**
                     * The callback that is invoked to consume the messages that are sent. Make sure that the callback has no exception. If a response is returned for the callback, the messages are consumed. 
                     * The messages must be consumed in a specified period. If the amount of time used to consume messages exceeds the timeout period that is specified by the Message Queue for MQTT broker, the Message Queue for MQTT broker retries sending messages. Make sure that the idempotence of the business data is kept and the business data is de-duplicated. 
                     */
                    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++) {
                MqttMessage message = new MqttMessage("hello mq4Iot pub sub msg".getBytes());
                message.setQos(qosLevel);
                /**
                 * The topic to which messages are sent. This topic must be the same as the topic to which the consumer subscribes or can be matched by using wildcards. 
                 */
                mqttClient.publish(mq4IotTopic, message);
                /**
                 * Message Queue for MQTT supports point-to-point (P2P) messaging. If the producer sends a message to the only consumer and obtains the value of the clientId parameter, a P2P message is sent. 
                 * In P2P messaging, the consumer does not need to subscribe to the topic to which the producer sends messages. The logic at the consumer side is simplified. In P2P messaging, specify a topic in the format of {{parentTopic}}/p2p/{{targetClientId}}. 
                 */
                final String p2pSendTopic = parentTopic + "/p2p/" + clientId;
                message = new MqttMessage("hello mq4Iot p2p msg".getBytes());
                message.setQos(qosLevel);
                mqttClient.publish(p2pSendTopic, message);
            }
            Thread.sleep(Long.MAX_VALUE);
        }
    }

Verify the result

After the producer sends the message and the consumer subscribes to the message, you can query the trace of the message in the Message Queue for MQTT console to verify whether the message is sent and received. For more information, see Query the message traces.

References

Use Message Queue for MQTT SDK for Java to implement messaging between Message Queue for MQTT clients and backend service applications