我正在树莓派上构建一些硬件,其中有许多传感器和一堆执行器。我想使用 Asyncio 协同程序来持续监控多个传感器(基本上这样我就不必从主循环中进行轮询),然后用一堆执行器做一些事情。
我打算为每个传感器创建一个类,它将具有类似于下面代码中协程的方法。
我想将传感器方法的结果生成到某个变量,然后我可以对其进行操作。
我的问题是,如果我有多个协程写入一个地方,我该如何安全地进行写入。 asyncio 中的队列似乎是一对一的,而不是多对一的 - 对吗?
最终我不明白如何让多个协程返回一个地方,有一些逻辑,然后向其他协程发送消息
+------------+
| | +------------+
| Sensor 1 +-------+ | |
| | | +---> actuator1 |
+------------+ | | | |
| | +------------+
+------------+ | +-----------+ |
| | | | | |
| Sensor 2 +------------> | logic +-+
| | | | | |
+------------+ | +-----------+ |
| | +------------+
+------------+ | | | |
| | | +---> actuator2 |
| Sensor 3 +-------+ | |
| | +------------+
+------------+
以上代表了我正在努力实现的目标。我知道我可以通过轮询和 while 循环来实现这一点,但我喜欢尝试异步/事件驱动方法的想法。
import asyncio
import random
async def sensor(queue):
while True:
# Get some sensor data
sensor_data = "data"
await queue.put(sensor_data)
async def actuator(queue):
while True:
# wait for an item from the producer
item = await queue.get()
if item is None:
# the producer emits None to indicate that it is done
break
# process the item
print('consuming item {}...'.format(item))
# simulate i/o operation using sleep
await asyncio.sleep(random.random())
loop = asyncio.get_event_loop()
queue = asyncio.Queue(loop=loop)
sensor_coro = sensor(queue)
actuator_coro = actuator(queue)
loop.run_until_complete(asyncio.gather(sensor_coro, actuator_coro))
loop.close()
最佳答案
My question is, if I have multiple coroutines writing to a single place how do I do that safely. The queues in asyncio seem to be one to one, not many to one - is that correct?
这是不正确的;异步队列是多生产者多消费者。要实现图表的逻辑,您需要两个同步原语:
一个队列,由
sensor
协程的多个实例填充,由logic()
协程的单个实例排空每个执行器都有一个额外的同步装置。哪种设备最好取决于需求。例如,是否允许执行器“丢失”传入速度快于响应速度的消息?或者,
逻辑
是否应该等待?根据这些,logic()
和每个执行器之间的同步将是一个简单的Event
(或者甚至只是一个Future
)或另一个队列.
假设您为每个执行器使用一个队列,您的逻辑
协程可能如下所示:
async def logic(sensor_queue, actuator_queues):
while True:
item = await queue.get()
# process the item and signal some actuators
await actuator_queues[0].put(x1)
await actuator_queues[1].put(x2)
关于python asyncio双向通信硬件控制,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/51797657/