Forward data to an AMQP consumer group

Step 2: Run the AMQP client

1. Log on to the ECS instance. For more information about logon methods, see Select an ECS remote connection method.

2. Run the following command to download the demo file.

   wget https://linkkit-export.oss-cn-shanghai.aliyuncs.com/amqp/amqp-demo.zip

3. Run the following command to decompress the demo file.

   unzip amqp-demo.zip

4. In the AmqpClient.java file in the src/main/java/com.aliyun.iotx.demo directory, modify the AMQP access information as described in the following table.

   Important

   The demo code in this example includes code to terminate the program (Thread.sleep(60 * 1000);). This means the program terminates one minute after it starts successfully. In a real-world scenario, you can set the runtime as needed.

   Parameter	Description
   accessKey	The AccessKey ID and AccessKey secret of an Alibaba Cloud account or a Resource Access Management (RAM) user. Log on to the IoT Platform console, move the pointer over your profile picture, and then click AccessKey Management to get the AccessKey ID and AccessKey secret. Important To avoid the security risks of hard coding an AccessKey into your business code, use environment variables to manage your AccessKey. Set the ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET environment variables in your local operating system to your AccessKey ID and AccessKey secret, respectively. In the sample code, you can retrieve them using the following methods: System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID") System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
   accessSecret
   consumerGroupId	The ID of the consumer group in the corresponding instance of IoT Platform. Log on to the IoT Platform console. In the corresponding instance, choose Message Forwarding > Server-side Subscription > Consumer Group List to view your consumer group ID.
   iotInstanceId	The instance ID. You can view the ID of the current instance on the Instance Overview page of the IoT Platform console. If an ID value exists, you must pass this ID value. If there is no Instance Overview page or ID value, pass an empty value, which is iotInstanceId = "".
   clientId	The client ID. It is user-defined and cannot exceed 64 characters. Use a unique identifier such as the UUID, MAC address, or IP address of the server where your AMQP client is located. After the AMQP client is connected and started, log on to the IoT Platform console. In the corresponding instance, choose Message Forwarding > Server-side Subscription > Consumer Group List. Click View for the consumer group. The Consumer Group Details page will display this parameter, making it easy for you to identify and distinguish different clients.
   connectionCount	The number of connections for starting the AMQP client. The maximum is 128. This is used for scaling out real-time message pushes. The Consumer Group Details page displays connected clients in the format ${clientId}+"-"+number. The minimum value for the number is 0.
   host	The AMQP access domain name. For information about the AMQP access domain name corresponding to ${YourHost}, see View and configure instance endpoint information (Endpoint).

5. The relevant Maven dependencies have been added to the pom.xml file. In the amqp-demo root directory, run the following command to reload Maven changes and build the project.

   mvn clean package

6. In the amqp-demo/target directory, run the following command to run the generated JAR package.

   java -jar demo-0.0.1-SNAPSHOT-jar-with-dependencies.jar

7. After you run the sample code, the following information is returned, which indicates that the AMQP client has connected to IoT Platform and is successfully receiving messages.

   Important

   Device messages can be received on the server only when the AMQP client is online.

   10:42:43.254 [main] INFO com.aliyun.iotx.demo.AmqpClient - amqp demo is started successfully, and will exit after 60s 
   10:59:46.405 [AmqpProvider :(3):[amqps://iot-cn-6******.amqp.iothub.aliyuncs.com:5671]] DEBUG org.apache.qpid.jms.provider.amqp.AmqpConsumer - Dispatching received message: JmsInboundMessageDispatch { sequence = 1, messageId = 1731508564705******, consumerId = ID:4e6d6f2b-1423-4c44-91ca-37202a******:1:1:1 }
   10:59:46.409 [JmsSession [ID:4e6d6f2b-1423-4c44-91ca-37202a******:1:1] delivery dispatcher] DEBUG org.apache.qpid.jms.provider.failover.FailoverProvider - Executing Failover Task: message acknowledge -> JmsInboundMessageDispatch { sequence = 1, messageId = 1731508564705******, consumerId = ID:4e6d6f2b-1423-4c44-91ca-37202******:1:1:1 } ackType: DELIVERED (5)
   10:59:46.432 [AmqpProvider :(3):[amqps://iot-cn-6******.amqp.iothub.aliyuncs.com:5671]] DEBUG org.apache.qpid.jms.provider.amqp.AmqpConsumer - Delivered Ack of message: JmsInboundMessageDispatch { sequence = 1, messageId = 1731508564705******, consumerId = ID:4e6d6f2b-1423-4c44-91ca-37202a******:1:1:1 }
   10:59:46.441 [JmsSession [ID:4e6d6f2b-1423-4c44-91ca-37202******:1:1] delivery dispatcher] DEBUG org.apache.qpid.jms.provider.failover.FailoverProvider - Executing Failover Task: message acknowledge -> JmsInboundMessageDispatch { sequence = 1, messageId = 1731508564705******, consumerId = ID:4e6d6f2b-1423-4c44-91ca-37202a******:1:1:1 } ackType: ACCEPTED (6)
   10:59:46.442 [AmqpProvider :(3):[amqps://iot-cn-6******.amqp.iothub.aliyuncs.com:5671]] DEBUG org.apache.qpid.jms.provider.amqp.AmqpConsumer - Accepted Ack of message: JmsInboundMessageDispatch { sequence = 1, messageId = 1731508564705******, consumerId = ID:4e6d6f2b-1423-4c44-91ca-37202a******:1:1:1 }
   10:59:46.452 [pool-1-thread-1] INFO com.aliyun.iotx.demo.AmqpClient - receive message,
    topic = /g18******/device01/thing/event/property/post,
    messageId = 1731508564705******,
    content = {"temperature":10,"humidity":56}

   The online AMQP clients are displayed in the corresponding consumer group. In amqp-demo, connectionCount = 4 indicates four clients.

   

What to do next

1. If the AMQP client is offline, messages for the AMQP server-side subscription accumulate. After the AMQP client comes back online, IoT Platform pushes the messages again. If you do not need to consume the stacked messages, you can clear them before the AMQP client goes online.

2. After all configurations are complete and the device reports subscription data that is received by the AMQP client, you can log on to the IoT Platform console and go to the corresponding instance to view message operational logs.

   • On the Monitoring and O&M > Simple Log Service > IoT Platform Logs tab, you can view the log records of device data reporting, data forwarding from IoT Platform to the AMQP client, and acknowledgments (ACKs) returned by the AMQP client. For more information, see Query IoT Platform logs.

   • On the Message Forwarding > Server-side Subscription > Consumer Group List tab, click View in the Actions column of the target consumer group. On the consumer group details page, you can view the message consumption rate, message accumulation, consumption logs, and more. For more information, see View and monitor consumer groups.

FAQ

• How do I implement device connection and status synchronization on IoT Platform?

• Can hexadecimal data reported by a device be subscribed to through an AMQP server-side subscription?

• Does one consumer group correspond to one AMQP message queue?

• Can one device correspond to only one AMQP message queue?

• After subscribing to device-reported messages via an AMQP server-side subscription, why are messages from the revert-RPC communication topic not received?

• Can hexadecimal data reported by a device be subscribed to through an AMQP server-side subscription?

• Why can't the AMQP client receive real-time messages?

• Why is there a delay when IoT Platform forwards messages through a consumer group?

• Why does the AMQP client go offline automatically?

• Why doesn't the device receive messages published by the AMQP client?