Configure an AMQP server-side subscription and connect an AMQP client - IoT Platform

Configure una suscripción del lado del servidor AMQP para que su servidor pueda conectarse como cliente AMQP y recibir datos de un dispositivo de lámpara de calle de ejemplo.

Requisitos previos

Debe haber un dispositivo conectado a IoT Platform que esté reportando datos. Para obtener más información, consulte Connect a device to IoT Platform and submit data.

Preparar el entorno de desarrollo

El cliente AMQP de ejemplo en este tema está desarrollado en Java. Se recomienda utilizar el cliente Apache Qpid JMS. Para consultar las instrucciones de uso, visite Qpid JMS 0.57.0.

Este ejemplo utiliza el siguiente entorno de desarrollo:

Sistema operativo: Windows 10 (64 bits)
JDK: JDK 8
Entorno de desarrollo integrado (IDE): IntelliJ IDEA Community Edition

Configurar una suscripción del lado del servidor AMQP

En la consola de IoT Platform, en la pestaña Overview, haga clic en la instancia IoT Test Environment.
Cree un grupo de consumidores que su servidor utilizará para recuperar los mensajes reenviados.
1. En el panel de navegación izquierdo, vaya a Message Forwarding > Server-Side Subscription y haga clic en la pestaña Consumer Groups.
2. Haga clic en Create Consumer Group.
3. En el cuadro de diálogo Create Consumer Group, establezca el nombre del grupo de consumidores como Thermostat Data Receiver y haga clic en OK.
  
  Una vez creado el grupo de consumidores, se le redirige a la página Consumer Group Details. Busque el consumer group ID y haga clic en Copy para guardar el consumer group ID. Necesitará este ID para conectar su cliente AMQP.

Configure una suscripción del lado del servidor para el producto Home Thermostat. Esto permite que su servidor se suscriba a diversos tipos de mensajes provenientes de dispositivos bajo este producto.

En el panel de navegación izquierdo, vaya a Message Forwarding > Server-Side Subscription.
En la página Server-Side Subscription, dentro de la pestaña Subscriptions, haga clic en Create Subscription.

En el cuadro de diálogo Create Subscription, configure los parámetros y haga clic en OK.

Parámetro	Descripción
Product	Seleccione Home Thermostat.
Subscription Type	Seleccione AMQP.
Consumer Group	Seleccione el grupo de consumidores Thermostat Data Receiver que creó anteriormente.
Message Type	Seleccione Device Upstream Notification y Submit historical TSL Data. Para obtener más información sobre los tipos de mensajes, consulte Configure an AMQP server-side subscription.

También puede utilizar la función de reenvío de datos para enviar información a un grupo de consumidores de una suscripción AMQP y usar un cliente AMQP para recibir mensajes. Por ejemplo, puede reenviar el contenido procesado de TSL Data Submission desde la regla de reenvío de datos Thermostat Data Forwarding hacia el grupo de consumidores Thermostat Data Reception de la suscripción AMQP. Para obtener más información, consulte Forward data to a consumer group of a server-side AMQP subscription.

Ejemplo de conexión de cliente AMQP

Descargue y descomprima el paquete de demostración.
Abra IntelliJ IDEA e importe el proyecto de ejemplo amqp-demo desde el paquete.

El archivo pom.xml ya contiene la dependencia de Maven necesaria para descargar el cliente Qpid JMS.

En el archivo AmqpClient.java, ubicado en el directorio src/main/java/com.aliyun.iotx.demo, modifique los parámetros para conectar el cliente JMS a IoT Platform. Consulte la siguiente tabla para más detalles.

    private static String accessKey = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID");
    private static String accessSecret = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET");
    private static String consumerGroupId = "1yRO*****************";
    // The ID of your instance. If you are using a public instance that was activated before July 30, 2021, leave this parameter empty.
    private static String iotInstanceId = "iot-cn-******";
    // The client ID. The value of this parameter is displayed in the Client ID column on the consumer group status page in the IoT Platform console.
    // We recommend that you use a unique identifier, such as a machine UUID, MAC address, or IP address, as the client ID to easily distinguish between different clients.
    private static String clientId = "test_IoT";
    // The endpoint for your instance. For more information, see the documentation on how to connect an AMQP client.
    private static String host = "iot-cn-******.amqp.iothub.aliyuncs.com";
    // The number of connections to be established by a single process.
    // The consumption rate of a single connection is limited. You can create up to 128 connections. For more information, see Limits.
    // The number of connections is related to the consumption rate and rebalancing. We recommend that you add one connection for every 500 QPS increase in consumption rate.
    private static int connectionCount = 4;

Parámetro	Descripción
accessKey	Inicie sesión en la consola de IoT Platform. Pase el cursor sobre su foto de perfil y haga clic en AccessKey Management para obtener su AccessKey ID y su AccessKey secret. Importante Para evitar el riesgo de seguridad que implica codificar directamente su par de AccessKey en el código de negocio, este ejemplo utiliza variables de entorno. Debe crear las variables de entorno ALIBABA_CLOUD_ACCESS_KEY_ID y ALIBABA_CLOUD_ACCESS_KEY_SECRET en su sistema operativo local y asignarles su AccessKey ID y su AccessKey secret, respectivamente.
accessSecret
consumerGroupId	El ID del grupo de consumidores Thermostat Data Receiver que creó. Consulte el paso 2 en la sección Configure an AMQP server-side subscription de este tema.
iotInstanceId	El ID de su instancia IoT Test Environment. Puede encontrar el ID de la instancia en la página Overview de la consola de IoT Platform.
clientId	El ID del cliente. Debe definir este ID, el cual puede tener hasta 64 caracteres de longitud. Se recomienda utilizar un identificador único, como el UUID, la dirección MAC o la dirección IP del servidor donde se encuentra su cliente AMQP. Una vez que el cliente AMQP esté conectado e iniciado, inicie sesión en la consola de IoT Platform. En la pestaña Consumer Groups de la página Message Forwarding > > > Server-side Subscription correspondiente a la instancia, haga clic en View junto al grupo de consumidores. La página Consumer Group Details muestra este parámetro, lo que facilita la identificación de los distintos clientes.
host	El endpoint AMQP para su instancia. Para obtener más información, consulte Create an enterprise edition instance. También puede encontrar el endpoint en la consola de IoT Platform. En la página Instance Details de su instancia IoT Test Environment, haga clic en View Development Configurations y luego en la pestaña AMQP.
connectionCount	El número de conexiones para el cliente AMQP. El valor máximo es 128. Este parámetro se utiliza para escalar horizontalmente el envío de mensajes en tiempo real. En la página Consumer Group Details, los clientes conectados se muestran con el formato `${clientId}+"-"+number`. El valor mínimo para number es 0.

Ejecute el código de ejemplo AmqpClient.java. Una salida similar a la siguiente indica que el cliente AMQP se ha conectado correctamente a IoT Platform y está recibiendo mensajes.

Nota

El código de demostración incluye el fragmento Thread.sleep(60 * 1000);, que finaliza el programa un minuto después de iniciarse. En su aplicación, configure el tiempo de ejecución según sus requisitos de negocio.

[main] INFO com.aliyun.iotx.demo.AmqpClient - amqp demo is started successfully, and will exit after 60s
[AmqpProvider :(2):[amqps://xxx.aliyuncs.com:5671]] DEBUG org.apache.qpid.jms.provider.amqp.AmqpConsumer - Dispatching received message: JmsInboundMessageDispatch { sequence
[JmsSession [ID:d0cd8ff1-5c3xxx] xxx dispatcher] DEBUG org.apache.qpid.jms.provider.failover.FailoverProvider - Executing Failover Task: message acknowledge -> JmsInboundMessag...
[AmqpProvider :(2):[amqps://xxx.aliyuncs.com:5671]] DEBUG org.apache.qpid.jms.provider.amqp.AmqpConsumer - Delivered Ack of message: JmsInboundMessageDispatch { sequence =
[JmsSession [ID:d0cd8ff1-5c3xxx] xxx dispatcher] DEBUG org.apache.qpid.jms.provider.failover.FailoverProvider - Executing Failover Task: message acknowledge -> JmsInboundMessag...
[AmqpProvider :(2):[amqps://xxx.aliyuncs.com:5671]] DEBUG org.apache.qpid.jms.provider.amqp.AmqpConsumer - Accepted Ack of message: JmsInboundMessageDispatch { sequence = 1
[pool-1-thread-1] INFO com.aliyun.iotx.demo.AmqpClient - receive message,
 topic = /a1xxx/device1/thing/event/property/post,
 messageId = 1380044478987006465,
 content = {"deviceType":"CustomCategory","iotId":"nTT9rxxx","requestId":"1617863381153","checkFailedData":{},"productKey":"a10xxx","gmtCreate":1617863218430,"deviceName":"device1","items":{"LightSwitch":{"value":1,"time":1617863218430}}}}}}}}

Parámetros de respuesta:

Parámetro	Valor	Descripción
topic	//*/thing/event/property/post	El topic utilizado para reportar propiedades del dispositivo.
messageId	1324198300680719360	El ID del mensaje.
content	{"temperature":23,"humidity":21,"time":1604548451951}	La carga útil del mensaje.

Después de ejecutar correctamente el código, puede consultar el estado básico del grupo de consumidores.

Para obtener más información, consulte Manage AMQP consumer groups.

El estado del grupo de consumidores incluye métricas como la tasa de consumo de mensajes en tiempo real, la tasa de consumo de mensajes acumulados y la acumulación de mensajes. La lista de clientes en línea debajo de las métricas muestra la tasa de consumo en tiempo real por conexión y la tasa de consumo acumulado por conexión para cada cliente.

Consultar registros

Una vez completada la configuración, inicie sesión en la consola de IoT Platform. En su instancia de edición empresarial, vaya a Maintenance > Device Log. Seleccione el producto Home Thermostat. Podrá consultar los registros completos en la pestaña Cloud Run Log.

La página de registros muestra cuatro entradas para Device1. Los tipos de operación son Server-Side Subscription (con una operación AMQP), Cloud-to-Device Message, TSL y Device-to-Cloud Message. El código de estado para todos los registros es 200, lo que indica un flujo normal de mensajes.

Siguientes pasos

Send commands from a server to a device