Reputation: 9933
I want to new a thread to do a io task with consumer/producer pattern in a asyncio task:
import asyncio
import threading
import time
import queue
q = queue.Queue()
def consumer():
while True:
v = q.get()
time.sleep(v)
print(f"log {v}")
async def work(v):
await asyncio.sleep(0.1)
print(f"work {v}")
async def main():
t = threading.Thread(target=consumer, daemon=True)
t.start()
for i in range(1, 5):
print(f"start {i}")
q.put(i)
loop.create_task(work(i))
print(f"done {i}")
t.join()
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.run_forever()
But the asyncio task cannot be execute because of consumer thread.
Upvotes: 0
Views: 418
Reputation: 155436
Instead of explicitly managing the worker thread and a queue, you can use the thread pool built into asyncio (or create your own if you wish) and call run_in_executor
to submit tasks to it. For example:
import asyncio, time
def blocking_work(v):
# this function is not async, it will be run in a separate thread
time.sleep(v)
print(f"log {v}")
async def work(v):
# native asyncio work
await asyncio.sleep(0.1)
# followed by blocking IO wrapped in thread
loop = asyncio.get_event_loop()
await loop.run_in_executor(None, blocking_work, v)
print(f"work {v}")
async def main():
await asyncio.gather(*[work(i) for i in range(1, 5)])
asyncio.run(main())
Upvotes: 2