DocsICP FilingConsole
Documentation
Home Page IoT Platform (Deprecated) User Guide Communications Data forwarding (new version) Data forwarding examples Forward data to an AMQP server-side subscription consumer group for consumption

Forward data to an AMQP server-side subscription consumer group for consumption

Updated at:
Copy as MD
Product
My Favorites

AMQP server-side subscriptions retrieve real-time data from all devices under a product. To forward messages from specific devices only, use the data forwarding feature. Data forwarding processes and filters messages through a Resolver script before forwarding them to an AMQP server-side subscription consumer group for consumption by an AMQP client. This topic uses a Thing Specification Language (TSL) model data reporting topic as an example to describe the process of forwarding message data.

How it works

image

Data forwarding sends messages from a specified topic of all or specific devices under the same product to one or more consumer groups in real time. Each consumer group contains multiple consumers (AMQP clients). When a message arrives at a consumer group, a random consumer receives it. Consumer groups are identified by their IDs. The preceding figure illustrates this process:

  1. Consumer: A consumer that is registered with a consumer group using an AMQP software development kit (SDK). It receives messages that IoT Platform forwards to the AMQP consumer group.

  2. IoT Platform: You can configure the data source topic, data destination, and Resolver script for data forwarding. Then, you can start the Resolver to forward device messages to the AMQP server-side subscription consumer group.

  3. AMQP client: After the Resolver is started, IoT Platform automatically forwards device messages to the AMQP client. Do not subscribe to topics in the AMQP client code. The AMQP client receives messages as long as it is online.

Note

Data forwarding does not support sending messages from an AMQP client to a device. To send instructions to a device, you can call the messaging APIs.

Scenarios

Business server receives device messages: Data forwarding provides a flexible way to forward device messages to an AMQP server-side subscription consumer group.

  • Forward messages from specific devices.

  • Forward messages from specific topics.

  • Filter or process messages before forwarding.

To retrieve real-time messages from all devices under a product, configure an AMQP server-side subscription directly. For more information, see Configure an AMQP server-side subscription.

Prerequisites

Limits

  • After an AMQP client establishes a connection, it must immediately send an authentication request. If authentication fails within 15 seconds, the server closes the connection.

  • A single connection of an AMQP client is limited to 1,000 transactions per second (TPS). The TPS limit for message forwarding is determined by the instance's Message Forwarding TPS specification. There is no limit on message size. For more information about the limits on AMQP server-side subscriptions, see Limits on server-side subscriptions.

Step 1: Configure a data destination and a Resolver in IoT Platform

Step 1: Configure the data destination

  1. Log on to the IoT Platform console.

  2. On the Overview page, find the instance that you want to manage and click the instance ID or instance name.

  3. In the left-side navigation pane, choose Message Forwarding > Data Forwarding.

  4. On the Data Forwarding page, click Go to New Version in the upper-right corner.

    Note

    If you have performed this operation before, you are automatically redirected to the new version of the Data Forwarding page.

  5. Click the Data destination tab, and then click Create data destination.

  6. In the Create Data Destination dialog box, enter a name for the data destination, such as DataPurpose. Configure the parameters as described in the following table and then click OK.

    Parameter

    Description

    Select Operation

    Select Publish to AMQP Server-side Subscription Consumer Group.

    Consumer Group

    Select an existing consumer group as the data forwarding target. Click Create Consumer Group to create a new one.

Step 2: Configure and start the Resolver

  1. Create a parser, such as DataParser. For more information, see Step 1: Create a parser.

  2. On the Resolver Details page, associate the data source.

    1. In the configuration wizard, under Data Source, click Associate Data Source.

    2. In the dialog box, select DataSource from the Data Source drop-down list and click OK.

  3. On the Resolver Details page, you can associate data destinations.

    1. In the configuration wizard, click Data Destination. Then, in the upper-right corner of the Data Destination list, click Associate Data Destination.

    2. In the dialog box that appears, from the Data Destination drop-down list, select DataPurpose and click OK.

    3. In the data destination list, find the Data Destination ID and record it. For example, 1000.

      You will need this Data Destination ID in the parsing script.

  4. On the parser details page, click the Parser Script tab.

  5. In the script editor, enter a Resolver script.

    The Resolver script uses JavaScript-like syntax. For more information about editing methods, see Script syntax.

    To forward data to an AMQP server-side subscription consumer group, use the writeAmqp(destinationId, payload, tag) function. For more information about the function parameters, see Function list.

    • To forward data from all devices under a product without specifying devices:

      // Use the payload() function to get the message content reported by the device and convert it to the JSON format.
var data = payload("json");
// Directly forward the TSL model data.
writeAmqp(1000, data, "debug");

    • To forward messages from only a specific device:

      // Use the payload() function to get the message content reported by the device and convert it to the JSON format.
var data = payload("json");
// Get the name of the device that reported the message.
var dn = deviceName();
// Forward the TSL model data of the specified device.
if (dn == 'device01') { 
    writeAmqp(1000, data, "debug");  
}

  6. Click Debug. Follow the on-screen instructions to select a product and device, enter the topic and payload data, and verify that the script is executable.

    In the Debug dialog on the Debug Parameters tab, the Topic path ends with /thing/event/property/post, and the Payload data is in JSON format containing productKey, gmtCreate, deviceName, and items (example: Temperature value is 38, Humidity value is 25).

    The following result indicates that the script ran successfully: action: transmit to amqp[destinationId=1000]

  7. Click Publish.

  8. Return to the Resolver tab on the Data Forwarding page and click Start for the DataParser resolver.

Step 2: Run the AMQP client

Important

  • Use the AMQP SDK sample provided by Alibaba Cloud IoT Platform. Alibaba Cloud does not provide technical support for self-developed AMQP SDKs.

  • This example uses Java. For examples in other languages, see AMQP client access instructions.

  • This example uses an ECS instance with the Alibaba Cloud Linux operating system as the development environment for 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 consumer group ID in the corresponding IoT Platform instance.

    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. View the ID 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

    A user-defined client ID, up to 64 characters. Use a unique identifier such as the UUID, MAC address, or IP address of your AMQP client server.

    After the AMQP client connects, 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, helping you identify and distinguish clients.

    connectionCount

    The number of connections for the AMQP client, up to 128. Used to scale out real-time message delivery.

    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 output indicates that the AMQP client has connected to IoT Platform and is 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. When the AMQP client is offline, subscription messages accumulate. After the client comes back online, IoT Platform pushes the accumulated messages. To discard these messages, you can clear them before the AMQP client goes online.

  2. After configuration is complete and the AMQP client receives device data, 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

Previous：Forward data to other topicsNext：Forward data to the cloud-native database PolarDB for MySQL