Я работаю над схемой взаимодействия между несколькими очередями для обработки элементов в конвейере. Я использую датчики для связи между очередями, когда следует прекратить работу, однако в следующем коде я вижу результаты, которые меня смущают.
При чтении из write_q в write_task () Я вижу, что первое значение входит в качестве дозорного None вместо задач в том порядке, в котором они были помещены в response_task(). Если я правильно понимаю, write_task() должна получать элементы по порядку и обрабатывать их по мере создания задач.
Кроме того, при печати qsize() в write_task() после того, как я нашел дозорного, он говорит, что есть 0 элементов, однако при обратной печати в основном кажется, что qsize() из write_q все еще имеет 2 элемента . Я где-то читал, что aiofiles использует run_in_executor(), что означает, что может быть расхождение в том, где находится очередь.
Большая часть приведенного ниже кода является шаблонным чтобы проиллюстрировать реальный сценарий того, почему мой код продолжает бесконечно блокироваться.
import asyncio
import aiohttp
import aiofiles
import aiocsv
import json
async def fetch(req: dict) -> dict:
# Make the request
async with aiohttp.ClientSession() as session:
try:
async with session.request("POST", url=req["url"], data=json.dumps(req["data"]), headers=req["headers"]) as response:
payload = await response.json()
response.raise_for_status()
print(f"INFO: response status was: {response.status}")
# Put response into queue to be written to file
return payload
except Exception as err:
print(f"ERROR: error making request: {err}")
async def task(l: list, request_q: asyncio.Queue) -> None:
# Read tasks from source of data
for i in l:
# Put a request task into the queue
req: dict = {
"headers": {"Accept": "application/json"},
"url": "https://httpbin.org/post",
"data": i
}
await request_q.put(
asyncio.create_task(fetch(req))#, response_q=response_q))
)
# Sentinel value to signal we are done receiving from source
await request_q.put(None)
async def request_task(request_q: asyncio.Queue, response_q: asyncio.Queue) -> None:
while True:
# Retrieve necessary data to make request
req = await request_q.get()
print(f"INFO: request from request_q: {req}")
# If we received sentinel for tasks, pass message to next queue
if not req:
print(f"INFO: request in request_task: {req}")
print("INFO: received sentinel from request_q")
request_q.task_done()
await response_q.put(None)
break
# Make the request which will put data into the response queue
resp = await req
print(f"INFO: response in request_task: {resp}")
await response_q.put(resp)
request_q.task_done()
async def response_task(response_q: asyncio.Queue, write_q: asyncio.Queue) -> None:
while True:
# Retrieve response
resp = await response_q.get()
# If we received sentinel for tasks, pass message to next queue
if not resp:
print("INFO: received sentinel from response_q")
response_q.task_done()
await write_q.put(None)
break
await write_q.put(resp)
response_q.task_done()
async def write_task(write_q: asyncio.Queue) -> None:
headers: bool = True
while True:
print(f"INFO: write_q address in write_task: {id(write_q)}")
async with aiofiles.open("file.csv", mode="a+", newline='') as f:
w = aiocsv.AsyncWriter(f)
# Get data out of the queue to write it
data = await write_q.get()
print(f"INFO: data in write_task: {data}")
if not data:
print(f"INFO: Found sentinel in write_task, queue size was: {write_q.qsize()}")
write_q.task_done()
await f.flush()
break
if headers:
await w.writerow([
"status",
"data",
])
headers = False
# Write the data from the response
await w.writerow([
"200",
json.dumps(data)
])
await f.flush()
print("INFO: finished write task")
write_q.task_done()
async def main() -> None:
# Create fake data to POST
items: list[str] = [["hello", "world"], ["asyncio", "test"]] * 1
# Queues for orchestrating
request_q = asyncio.Queue()
response_q = asyncio.Queue()
write_q = asyncio.Queue()
# one producer
producer = asyncio.create_task(
task(items, request_q)
)
# 5 request consumers
request_consumers = [
asyncio.create_task(
request_task(request_q, response_q)
)
for _ in range(5)
]
# 5 response consumers
response_consumers = [
asyncio.create_task(
response_task(response_q, write_q)
)
for _ in range(5)
]
# 5 write consumers
write_consumer = asyncio.create_task(
write_task(write_q)
)
errors = await asyncio.gather(producer, return_exceptions=True)
print(f"INFO: Producer has completed! exceptions: {errors}")
await request_q.join()
for c in request_consumers:
c.cancel()
print("INFO: request consumer has completed! ")
await response_q.join()
for c in response_consumers:
c.cancel()
print("INFO: response consumer has completed! ")
print(f"INFO: write_q address in main: {id(write_q)}")
print(f"INFO: write_q in main qsize: {write_q.qsize()}")
await write_q.join()
await write_consumer
print("INFO: write consumer has completed! ")
print("INFO: Complete!")
if __name__ == "__main__":
loop = asyncio.new_event_loop()
loop.run_until_complete(main())
Подробнее здесь: https://stackoverflow.com/questions/786 ... s-expected
