Use o SDK para Python 2.7 para conectar um cliente AMQP ao Alibaba Cloud IoT Platform e receber mensagens por assinatura do lado do servidor.
Pré-requisitos
Obtenha um ID de grupo de consumidores e assine as mensagens de tópico necessárias.
Gerenciar grupos de consumidores AMQP: Use o grupo de consumidores padrão (DEFAULT_GROUP) no IoT Platform ou crie um novo grupo.
Configurar uma assinatura do lado do servidor AMQP: Assine as mensagens de tópico desejadas com um grupo de consumidores.
Ambiente de desenvolvimento
Este exemplo usa Python 2.7.
Baixe o SDK
Recomendamos a biblioteca Apache Qpid Proton 0.29.0, que encapsula a API Python. Para baixar a biblioteca e ver as instruções, acesse Qpid Proton 0.29.0.
Instale o Qpid Proton. Para mais informações, consulte Installing Qpid Proton.
Após instalar o Qpid Proton, execute o comando Python a seguir para verificar se a biblioteca SSL está disponível:
import proton;print('%s' % 'SSL present' if proton.SSL.present() else 'SSL NOT AVAILABLE')
Código de exemplo
# encoding=utf-8
import sys
import logging
import time
from proton.handlers import MessagingHandler
from proton.reactor import Container
import hashlib
import hmac
import base64
import os
reload(sys)
sys.setdefaultencoding('utf-8')
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
console_handler = logging.StreamHandler(sys.stdout)
def current_time_millis():
return str(int(round(time.time() * 1000)))
def do_sign(secret, sign_content):
m = hmac.new(secret, sign_content, digestmod=hashlib.sha1)
return base64.b64encode(m.digest())
class AmqpClient(MessagingHandler):
def __init__(self):
super(AmqpClient, self).__init__()
def on_start(self, event):
# The endpoint used to connect to IoT Platform. For details, see the AMQP connection guide.
url = "amqps://${YourHost}:5671"
# Hard-coding an AccessKey pair in your project code poses a security risk. If the code is leaked, your AccessKey pair will be exposed, compromising all resources in your account. The following code retrieves the AccessKey pair from environment variables as a best practice. This is for reference only.
accessKey = os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID']
accessSecret = os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
consumerGroupId = "${YourConsumerGroupId}"
clientId = "${YourClientId}"
# iotInstanceId: The ID of the IoT Platform instance.
iotInstanceId = "${YourIotInstanceId}"
# The signature algorithm. Valid values: hmacmd5, hmacsha1, and hmacsha256.
signMethod = "hmacsha1"
timestamp = current_time_millis()
# For instructions on how to construct the userName parameter, see the AMQP connection guide.
userName = clientId + "|authMode=aksign" + ",signMethod=" + signMethod \
+ ",timestamp=" + timestamp + ",authId=" + accessKey \
+ ",iotInstanceId=" + iotInstanceId + ",consumerGroupId=" + consumerGroupId + "|"
signContent = "authId=" + accessKey + "×tamp=" + timestamp
# Calculate the signature. For instructions on how to construct the password, see the AMQP connection guide.
passWord = do_sign(accessSecret.encode("utf-8"), signContent.encode("utf-8"))
conn = event.container.connect(url, user=userName, password=passWord, heartbeat=60)
self.receiver = event.container.create_receiver(conn)
# Called when the connection is successfully established.
def on_connection_opened(self, event):
logger.info("Connection established, remoteUrl: %s", event.connection.hostname)
# Called when the connection is closed.
def on_connection_closed(self, event):
logger.info("Connection closed: %s", self)
# Called when the remote peer closes the connection due to an error.
def on_connection_error(self, event):
logger.info("Connection error")
# Called when an AMQP connection error occurs, such as an authentication or socket error.
def on_transport_error(self, event):
if event.transport.condition:
if event.transport.condition.info:
logger.error("%s: %s: %s" % (
event.transport.condition.name, event.transport.condition.description,
event.transport.condition.info))
else:
logger.error("%s: %s" % (event.transport.condition.name, event.transport.condition.description))
else:
logging.error("Unspecified transport error")
# Called when a message is received.
def on_message(self, event):
message = event.message
content = message.body.decode('utf-8')
topic = message.properties.get("topic")
message_id = message.properties.get("messageId")
print("receive message: message_id=%s, topic=%s, content=%s" % (message_id, topic, content))
event.receiver.flow(1)
Container(AmqpClient()).run()
Configure os parâmetros no código anterior conforme a tabela a seguir. Para mais informações, consulte Conectar um cliente AMQP ao IoT Platform.
Especifique valores de parâmetro válidos. Caso contrário, o cliente AMQP não conseguirá se conectar ao IoT Platform.
|
Parâmetro |
Descrição |
|
url |
Endpoint que o cliente AMQP usa para se conectar ao IoT Platform. Formato: Para obter mais detalhes sobre o endpoint aplicável à variável |
|
accessKey |
Faça login no console do IoT Platform, passe o mouse sobre a foto de perfil no canto superior direito e clique em AccessKey Management para obter o AccessKey ID e o AccessKey secret. Nota
Se você usar um usuário do Resource Access Management (RAM), anexe a política AliyunIOTFullAccess a esse usuário para conceder permissões de gerenciamento de recursos do IoT Platform. Sem essa configuração, a conexão falhará. Para mais informações, consulte Acessar o IoT Platform como usuário RAM. |
|
accessSecret |
|
|
consumerGroupId |
ID do grupo de consumidores na instância do IoT Platform. Faça login no console do IoT Platform. Na instância correspondente, navegue até para visualizar o ID do seu grupo de consumidores. |
|
iotInstanceId |
ID da instância do IoT Platform. Você pode visualizar esse ID na aba Overview no console do IoT Platform.
|
|
clientId |
Identificador do cliente. Defina este ID, que pode ter até 64 caracteres. Recomendamos usar um identificador exclusivo, como UUID, endereço MAC ou endereço IP do servidor onde o cliente AMQP está hospedado. Após conectar e iniciar o cliente AMQP, faça login no console do IoT Platform. Na página da instância, clique em View ao lado do grupo de consumidores na aba Consumer Groups. A página Consumer Group Details exibirá este parâmetro para facilitar a distinção entre diferentes clientes. |
Resultados de exemplo
-
Sucesso: A mensagem de log a seguir indica que o cliente AMQP se conectou ao IoT Platform e está recebendo mensagens.
202x-xx-xx xx:xx:xx,490 - __main__ - INFO - Connection established, remoteUrl: xxx.amqp.iothub.aliyuncs.com receive message: message_id=13xxx, topic=/xxx/thing/event/property/post, content={"deviceType":"CustomCategory","iotId":"xxx","requestId":"1xxx"} receive message: message_id=13xxx, topic=/xxx/thing/event/property/post, content={"deviceType":"CustomCategory","iotId":"xxx","requestId":"1xxx"}Parâmetro
Exemplo
Descrição
message_id
2****7
ID da mensagem.
topic
//*/thing/event/property/post
Tópico usado para enviar propriedades do dispositivo.
content
{"deviceType":"CustomCategory","iotId":"qPi","requestId":"161","checkFailedData":{},"productKey":"g4","gmtCreate":1613635594038,"deviceName":"de","items":{"Temperature":{"value":24,"time":1613635594036},"Humidity":{"value":26,"time":1613635594036}}}
Conteúdo da mensagem.
-
Se uma saída semelhante à apresentada a seguir for exibida, o cliente AMQP não conseguiu se conectar ao IoT Platform.
Verifique o log de erros para revisar seu código e as configurações de rede. Corrija o problema e execute o código novamente.
2021-02-18 16:19:40,993 - proton - ERROR - Couldn't connect: 10053 2021-02-18 16:19:40,996 - __main__ - ERROR - proton.pythonio: Connection error: 10053 2021-02-18 16:19:40,996 - proton - INFO - Disconnected, reconnecting... Traceback (most recent call last): File "xxx", line 87, in <module> Container(AmqpClient()).run() File "xxx", line 181, in run while self.process(): pass File "xxx", line 240, in process event.dispatch(self._global_handler) File "xxx", line 135, in dispatch _dispatch(handler, type.method, self) File "xxx", line 117, in _dispatch handler.on_unhandled(method, *args) File "xxx", line 665, in on_unhandled event.dispatch(self.base) File "xxx", line 135, in dispatch _dispatch(handler, type.method, self) File "xxx", line 115, in _dispatch m(*args) File "xxx", line 892, in on_connection_bound addrs = socket.getaddrinfo(host, port, socket.AF_UNSPEC, socket.SOCK_STREAM) socket.gaierror: [Errno 11001] getaddrinfo failed
Referências
Para obter mais informações sobre códigos de erro relacionados a mensagens de assinatura do lado do servidor, consulte Códigos de erro relacionados a mensagens.