Python3标准库：asyncio异步I/O、事件循环和并发工具(7)

当前位置:

首页 > Python基础教程 >

Python3标准库：asyncio异步I/O、事件循环和并发工具(7)

			 
			async def main(loop):

			# Create a shared event

			event = asyncio.Event()

			print('event start state: {}'.format(event.is_set()))

			loop.call_later(

			0.1, functools.partial(set_event, event)

			)

			await asyncio.wait([coro1(event), coro2(event)])

			print('event end state: {}'.format(event.is_set()))

			event_loop = asyncio.get_event_loop()

			try:

			event_loop.run_until_complete(main(event_loop))

			finally:

			event_loop.close()

与Lock一样，coro1()和coro2()会等待设置事件。区别是一旦事件状态改变，它们便可以立即启动，并且它们不需要得到事件对象上的唯一的锁。

1.7.3 条件

Condition的做法与Event类似，只不过不是通知所有等待的协程，被唤醒的等待协程的数目由notify()的一个参数控制。

import asyncio
async def consumer(condition, n):
async with condition:
print('consumer {} is waiting'.format(n))
await condition.wait()
print('consumer {} triggered'.format(n))
print('ending consumer {}'.format(n))
async def manipulate_condition(condition):
print('starting manipulate_condition')
# pause to let consumers start
await asyncio.sleep(0.1)
for i in range(1, 3):
async with condition:
print('notifying {} consumers'.format(i))
condition.notify(n=i)
await asyncio.sleep(0.1)
async with condition:
print('notifying remaining consumers')
condition.notify_all()
print('ending manipulate_condition')
async def main(loop):
# Create a condition
condition = asyncio.Condition()
# Set up tasks watching the condition
consumers = [
consumer(condition, i)
for i in range(5)
]
# Schedule a task to manipulate the condition variable
loop.create_task(manipulate_condition(condition))
# Wait for the consumers to be done
await asyncio.wait(consumers)
event_loop = asyncio.get_event_loop()
try:
result = event_loop.run_until_complete(main(event_loop))
finally:
event_loop.close()

这个例子启动Condition的5个消费者。它们分别使用wait()方法来等待通知让它继续。manipulate_condition()通知一个消费者，再通知两个消费者，然后通知所有其余的消费者。

1.7.4 队列

asyncio.Queue为协程提供了一个先进先出的数据结构，这与线程的queue.Queue或进程的multiprocessing.Queue很类似。

import asyncio
async def consumer(n, q):
print('consumer {}: starting'.format(n))
while True:
print('consumer {}: waiting for item'.format(n))
item = await q.get()
print('consumer {}: has item {}'.format(n, item))
if item is None:
# None is the signal to stop.
q.task_done()
break
else:
await asyncio.sleep(0.01 * item)
q.task_done()
print('consumer {}: ending'.format(n))
async def producer(q, num_workers):
print('producer: starting')
# Add some numbers to the queue to simulate jobs
for i in range(num_workers * 3):
await q.put(i)
print('producer: added task {} to the queue'.format

栏目列表