Skip to main content
 首页 » 编程设计

python之AWS IoT Python SDK 和 asyncio

2025年01月19日30dflying

我需要使用 AWS IoT MQTT 服务。我正在用 https://github.com/aws/aws-iot-device-sdk-python 做一些试验目前。

我的应用程序将使用 websockets 与其他服务通信,然后发布/订阅 MQTT 主题以转发/接收消息。

这个库是否可能会阻止代码执行?我仍然试图了解 asyncio,但不确定我应该注意什么。我怎么知道它是否会引起问题?

我相信我只需要使用 AWSIoTMQTTClient来自上面的库。

这是我的工作代码的摘录:

class AWSIoTClient: 
 
    def __init__(): 
        ... 
        self.client = AWSIoTMQTTClient(...) 
 
    def subscribe(self, callback): 
        self.client.subscribe(f'{self.TOPIC}/subscribe/', 0, callback) 
 
    def publish(self, message): 
        self.client.publish(self.TOPIC, message, 0) 
 
 
class MyWSProtocol(WebSocketClientProtocol): 
 
    def set_aws_client(self, client: AWSIoTClient): 
        client.subscribe(self.customCallback) 
        self.client = client 
 
    def customCallback(self, client, userdata, message): 
        # This will be called when we send message from AWS 
        if message.payload: 
            message = json.loads(message.payload.decode('utf-8').replace("'", '"')) 
            message['id'] = self.next_id() 
            self.sendMessage(json.dumps(message).encode('utf-8')) 
 
    def onMessage(self, payload, isBinary): 
        message = json.loads(payload) 
 
        # This will forward message to AWS 
        self.client.publish(str(payload)) 

请您参考如下方法:

Is it likely that this library will be blocking code execution?

How do I know if it will cause problems?

您不应允许在任何协程中包含长时间运行的阻塞(同步)代码。它会导致阻塞你的全局事件循环,并进一步阻塞你所有的协程。

async def main(): 
    await asyncio.sleep(3)  # async sleeping, it's ok 
 
    time.sleep(3)           # synchronous sleeping, this freezes event loop  
                            # and all coroutines for 3 seconds,  
                            # you should avoid it! 
     
    await asyncio.sleep(3)  # async sleeping, it's ok 

如果您需要在协程中运行阻塞代码,您应该在执行器中执行(read here 关于它)。

在编写协程时应该牢记这一点,但如果启用 debug mode,asyncio 通常会警告您此错误。 :

import asyncio 
import time 
 
 
async def main(): 
    await asyncio.sleep(3) 
    time.sleep(3) 
    await asyncio.sleep(3) 
 
 
loop = asyncio.get_event_loop() 
loop.set_debug(True)  # debug mode 
try: 
    loop.run_until_complete(main()) 
finally: 
    loop.run_until_complete(loop.shutdown_asyncgens()) 
    loop.close() 

你会看到警告:

Executing <Handle <TaskWakeupMethWrapper object at 0x000002063C2521F8>(<Future finis...events.py:275>) created at C:\Users\gmn\AppData\Local\Programs\Python\Python36\Lib\asyncio\futures.py:348> took 3.000 seconds