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

Prérequis

Un appareil doit être connecté à IoT Platform et transmettre des données. Pour plus d'informations, consultez la rubrique Connecter un appareil à IoT Platform et envoyer des données.

Préparer l'environnement de développement

Le client AMQP d'exemple présenté dans cette rubrique est développé en Java. Nous vous recommandons d'utiliser le client Apache Qpid JMS. Pour obtenir des instructions d'utilisation, consultez la documentation Qpid JMS 0.57.0.

Cet exemple utilise l'environnement de développement suivant :

Système d'exploitation : Windows 10 (64 bits)
JDK : JDK 8
Environnement de développement intégré (IDE) : IntelliJ IDEA Community Edition

Configurer un abonnement côté serveur AMQP

Dans la console IoT Platform, sous l'onglet Overview, cliquez sur l'instance IoT Test Environment.
Créez un groupe de consommateurs que votre serveur utilisera pour récupérer les messages transférés.
1. Dans le volet de navigation de gauche, accédez à Message Forwarding > Server-Side Subscription, puis cliquez sur l'onglet Consumer Groups.
2. Cliquez sur Create Consumer Group.
3. Dans la boîte de dialogue Create Consumer Group, définissez le nom du groupe de consommateurs sur Thermostat Data Receiver et cliquez sur OK.
  
  Une fois le groupe de consommateurs créé, vous êtes redirigé vers la page Consumer Group Details. Repérez l'ID du groupe de consommateurs consumer group ID et cliquez sur Copy pour enregistrer l'ID du groupe de consommateurs consumer group ID. Vous aurez besoin de cet ID pour connecter votre client AMQP.

Configurez un abonnement côté serveur pour le produit Home Thermostat. Cela permet à votre serveur de s'abonner à différents types de messages provenant des appareils associés à ce produit.

Dans le volet de navigation de gauche, accédez à Message Forwarding > Server-Side Subscription.
Sur la page Server-Side Subscription, sous l'onglet Subscriptions, cliquez sur Create Subscription.

Dans la boîte de dialogue Create Subscription, configurez les paramètres et cliquez sur OK.

Parameter	Description
Product	Sélectionnez Home Thermostat.
Subscription Type	Sélectionnez AMQP.
Consumer Group	Sélectionnez le groupe de consommateurs Thermostat Data Receiver que vous avez créé.
Message Type	Sélectionnez Device Upstream Notification et Submit historical TSL Data. Pour plus d'informations sur les types de messages, consultez la rubrique Configurer un abonnement côté serveur AMQP.

Vous pouvez également utiliser la fonctionnalité de transfert de données pour acheminer les données vers un groupe de consommateurs d'un abonnement AMQP et utiliser un client AMQP pour recevoir les messages. Par exemple, vous pouvez transférer le contenu traité TSL Data Submission de la règle de transfert de données Thermostat Data Forwarding vers le groupe de consommateurs Thermostat Data Reception de l'abonnement AMQP. Pour plus d'informations, consultez la rubrique Transférer des données vers un groupe de consommateurs d'un abonnement AMQP côté serveur.

Exemple de connexion d'un client AMQP

Téléchargez et décompressez le package de démonstration.
Ouvrez IntelliJ IDEA et importez le projet d'exemple amqp-demo depuis le package.

Le fichier pom.xml contient déjà la dépendance Maven requise pour télécharger le client Qpid JMS.

Dans le fichier AmqpClient.java, situé dans le répertoire src/main/java/com.aliyun.iotx.demo, modifiez les paramètres pour connecter le client JMS à IoT Platform. Consultez le tableau ci-dessous pour plus de détails.

    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;

Parameter	Description
accessKey	Connectez-vous à la console IoT Platform. Survolez votre photo de profil et cliquez sur AccessKey Management pour obtenir votre AccessKey ID et votre AccessKey secret. Important Pour éviter les risques de sécurité liés à l'incorporation directe de votre paire AccessKey dans le code métier, cet exemple utilise des variables d'environnement. Vous devez créer les variables d'environnement ALIBABA_CLOUD_ACCESS_KEY_ID et ALIBABA_CLOUD_ACCESS_KEY_SECRET dans votre système d'exploitation local et définir leurs valeurs respectivement sur votre AccessKey ID et votre AccessKey secret.
accessSecret
consumerGroupId	L'ID du groupe de consommateurs Thermostat Data Receiver que vous avez créé. Reportez-vous à l'étape 2 de la section Configurer un abonnement côté serveur AMQP de cette rubrique.
iotInstanceId	L'ID de votre instance IoT Test Environment. Vous trouverez l'ID de l'instance sur la page Overview de la console IoT Platform.
clientId	L'ID client. Vous devez définir cet ID. Sa longueur maximale est de 64 caractères. Nous vous recommandons d'utiliser un identifiant unique, tel que l'UUID, l'adresse MAC ou l'adresse IP du serveur où se trouve votre client AMQP. Une fois le client AMQP connecté et démarré, connectez-vous à la console IoT Platform. Sur la page de l'instance, sous l'onglet Consumer Groups de la section Message Forwarding > > > Server-side Subscription, cliquez sur View à côté du groupe de consommateurs. La page Consumer Group Details affiche ce paramètre. Cela vous aide à identifier les différents clients.
host	Le point de terminaison AMQP de votre instance. Pour plus d'informations, consultez la rubrique Créer une instance édition entreprise. Vous pouvez également trouver le point de terminaison dans la console IoT Platform. Sur la page Instance Details de votre instance IoT Test Environment, cliquez sur View Development Configurations, puis sur l'onglet AMQP.
connectionCount	Le nombre de connexions pour le client AMQP. La valeur maximale est de 128. Ce paramètre sert à augmenter la capacité de diffusion des messages en temps réel. Sur la page Consumer Group Details, les clients connectés s'affichent au format `${clientId}+"-"+number`. La valeur minimale pour number est 0.

Exécutez l'exemple de code AmqpClient.java. Une sortie similaire à celle ci-dessous indique que le client AMQP est connecté à IoT Platform et reçoit des messages.

Remarque

Le code de démonstration inclut l'extrait de code Thread.sleep(60 * 1000);, qui met fin au programme une minute après son démarrage. Dans votre application, définissez la durée d'exécution en fonction de vos besoins métier.

[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}}}}}}}}

Paramètres de réponse :

Parameter	Value	Description
topic	/*********/****/thing/event/property/post	Le topic utilisé pour signaler les propriétés de l'appareil.
messageId	1324198300680719360	L'ID du message.
content	{"temperature":23,"humidity":21,"time":1604548451951}	La charge utile du message.

Une fois le code exécuté avec succès, vous pouvez afficher l'état de base du groupe de consommateurs.

Pour plus d'informations, consultez la rubrique Gérer les groupes de consommateurs AMQP.

L'état du groupe de consommateurs comprend des métriques telles que le débit de consommation des messages en temps réel, le débit de consommation des messages en attente et le nombre de messages en attente. La liste des clients en ligne située sous les métriques affiche le débit de consommation en temps réel par connexion et le débit de consommation des messages en attente par connexion pour chaque client.

Consulter les journaux

Une fois la configuration terminée, connectez-vous à la console IoT Platform. Dans votre instance édition entreprise, accédez à Maintenance > Device Log. Sélectionnez le produit Home Thermostat. Vous pouvez consulter l'intégralité des journaux sous l'onglet Cloud Run Log.

La page des journaux affiche quatre enregistrements pour Device1. Les types d'activité sont Server-Side Subscription (avec une opération AMQP), Cloud-to-Device Message, TSL et Device-to-Cloud Message. Le code d'état de tous les enregistrements est 200, ce qui indique un flux de messages normal.

Étapes suivantes

Envoyer des commandes d'un serveur vers un appareil