【发布时间】：2020-12-30 12:41:34
【问题描述】：
我正在编写一个 Python 程序，它同时运行从队列中获取的任务，以学习 asyncio。
通过与主线程（在 REPL 中）交互，将项目放入队列中。 每当一个任务被放入队列时，它应该被立即消费并执行。 我的方法是启动一个单独的线程并将队列传递给该线程内的事件循环。
任务正在运行，但只是按顺序运行，我不清楚如何同时运行这些任务。我的尝试如下：
import asyncio
import time
import queue
import threading
def do_it(task_queue):
'''Process tasks in the queue until the sentinel value is received'''
_sentinel = 'STOP'
def clock():
return time.strftime("%X")
async def process(name, total_time):
status = f'{clock()} {name}_{total_time}:'
print(status, 'START')
current_time = time.time()
end_time = current_time + total_time
while current_time < end_time:
print(status, 'processing...')
await asyncio.sleep(1)
current_time = time.time()
print(status, 'DONE.')
async def main():
while True:
item = task_queue.get()
if item == _sentinel:
break
await asyncio.create_task(process(*item))
print('event loop start')
asyncio.run(main())
print('event loop end')
if __name__ == '__main__':
tasks = queue.Queue()
th = threading.Thread(target=do_it, args=(tasks,))
th.start()
tasks.put(('abc', 5))
tasks.put(('def', 3))
任何能指出我同时运行这些任务方向的建议都将不胜感激！
谢谢
更新
谢谢 Frank Yellin 和 cynthi8！我根据您的建议对 main() 进行了改革：
- 在
asyncio.create_task之前删除了await- 修复了并发性 - 添加了 wait while 循环，以便 main 不会过早返回
- 使用了Queue.get()的非阻塞模式
程序现在按预期运行 ????
更新 2
user4815162342 提供了进一步的改进，我在下面注释了他的建议。
'''
Starts auxiliary thread which establishes a queue and consumes tasks within a
queue.
Allow enqueueing of tasks from within __main__ and termination of aux thread
'''
import asyncio
import time
import threading
import functools
def do_it(started):
'''Process tasks in the queue until the sentinel value is received'''
_sentinel = 'STOP'
def clock():
return time.strftime("%X")
async def process(name, total_time):
print(f'{clock()} {name}_{total_time}:', 'Started.')
current_time = time.time()
end_time = current_time + total_time
while current_time < end_time:
print(f'{clock()} {name}_{total_time}:', 'Processing...')
await asyncio.sleep(1)
current_time = time.time()
print(f'{clock()} {name}_{total_time}:', 'Done.')
async def main():
# get_running_loop() get the running event loop in the current OS thread
# out to __main__ thread
started.loop = asyncio.get_running_loop()
started.queue = task_queue = asyncio.Queue()
started.set()
while True:
item = await task_queue.get()
if item == _sentinel:
# task_done is used to tell join when the work in the queue is
# actually finished. A queue length of zero does not mean work
# is complete.
task_queue.task_done()
break
task = asyncio.create_task(process(*item))
# Add a callback to be run when the Task is done.
# Indicate that a formerly enqueued task is complete. Used by queue
# consumer threads. For each get() used to fetch a task, a
# subsequent call to task_done() tells the queue that the processing
# on the task is complete.
task.add_done_callback(lambda _: task_queue.task_done())
# keep loop going until all the work has completed
# When the count of unfinished tasks drops to zero, join() unblocks.
await task_queue.join()
print('event loop start')
asyncio.run(main())
print('event loop end')
if __name__ == '__main__':
# started Event is used for communication with thread th
started = threading.Event()
th = threading.Thread(target=do_it, args=(started,))
th.start()
# started.wait() blocks until started.set(), ensuring that the tasks and
# loop variables are available from the event loop thread
started.wait()
tasks, loop = started.queue, started.loop
# call_soon schedules the callback callback to be called with args arguments
# at the next iteration of the event loop.
# call_soon_threadsafe is required to schedule callbacks from another thread
# put_nowait enqueues items in non-blocking fashion, == put(block=False)
loop.call_soon_threadsafe(tasks.put_nowait, ('abc', 5))
loop.call_soon_threadsafe(tasks.put_nowait, ('def', 3))
loop.call_soon_threadsafe(tasks.put_nowait, 'STOP')
【问题讨论】：
标签： python concurrency queue task python-asyncio