Python3 asyncio处理字典内传感器任务并将结果存入字典的实现方法

问题背景 #

我正在学习asyncio，需要拉取多个传感器的数据，每个传感器拉取耗时约1秒，因此选择asyncio实现，且传感器列表会动态变化，asyncio任务也需要同步调整。
我已编写如下代码，但不知道如何将拉取到的数值存入结果字典。我认为消费者向字典存值的操作非常快，完全可以放在生产者中完成，不需要额外的消费者，但不确定是否符合asyncio范式要求。
或许我想复杂了，有没有代码量更少、更简单的编程范式可以实现需求？代码注释中还有更多细节问题和说明，供参考：

初始实现代码 #

#!/usr/bin/python3

import asyncio
import random

sensors={ # 该列表会频繁变更
"sensor1" : "http://abc.example.org",
"sensor2" : "http://outside.example.org",
"temperature" : "http://xe.example.com",
"outdoor" : "http://anywhere.example.org"
}

results = dict() # 传感器查询结果需要存入该字典

async def queryAll(sensors):
    q = asyncio.Queue()
    queries = [asyncio.create_task(querySensor(sensorname,q)) for sensorname in sensors]
    process = [asyncio.create_task(storeValues(sensorname, q)) for sensorname in sensors] # 因为只是往字典存值，一个消费者就够，甚至不需要消费者，生产者直接把值写入字典就行
    await asyncio.gather(*queries)
    await q.join()
    for c in process:
        c.cancel()

async def querySensor(sensorname: str, q: asyncio.Queue):
    res = str(random.randint(0,100))
    resString = "Result for " + sensorname + " is " + res
    await q.put(resString)

async def storeValues(sensorname: str, q: asyncio.Queue):
    while True:
        res = await q.get()
        print("Value: ", res)
        q.task_done()

if __name__ == "__main__":
    asyncio.run(queryAll(sensors))
    for result in results: # 此时结果应该已经存入results字典
        print(result, "measured:", results[result])

解答与最终实现 #

你的思路完全正确，这种简单场景下不需要额外引入队列和消费者逻辑：写入字典是耗时极短的同步操作，不会阻塞asyncio事件循环，直接在传感器查询任务中完成结果写入即可，既符合asyncio使用规范，还能大幅降低代码复杂度。
最终优化后的实现代码如下：

#!/usr/bin/python3

import asyncio
import random

sensors={ # 该列表会频繁变更
"sensor1" : "http://abc.example.org",
"sensor2" : "http://outside.example.org",
"temperature" : "http://xe.example.com",
"outdoor" : "http://anywhere.example.org"
}

results = dict() # 存储传感器查询结果

async def queryAll(sensors):
    queries = [asyncio.create_task(querySensor(sensorname)) for sensorname in sensors]
    await asyncio.gather(*queries)

async def querySensor(sensorname: str):
    # 实际场景下此处替换为真实的传感器接口请求逻辑
    res = str(random.randint(0,100))
    resString = "Result for " + sensorname + " is " + res
    results[sensorname] = resString

if __name__ == "__main__":
    asyncio.run(queryAll(sensors))
    for result in results:
        print(result, "measured:", results[result])

如果后续业务迭代需要对所有查询结果做统一的过滤、格式转换、异常告警等处理，再引入消费者队列做分层处理也不迟，当前简化实现完全满足需求且更易维护。

内容的提问来源于stack exchange，提问作者Hannes