如何在 Trio 中收集任务结果？

Question

我编写了一个脚本，该脚本使用 nursery 和 asks 模块来循环并根据循环变量调用 API。我收到了回复，但不知道如何像使用 asyncio 那样 return 数据。

我还有一个关于将 API 限制为每秒 5 个的问题。

from datetime import datetime
import asks
import time
import trio

asks.init("trio")
s = asks.Session(connections=4)

async def main():
    start_time = time.time()

    api_key = 'API-KEY'
    org_id = 'ORG-ID'
    networkIds = ['id1','id2','idn']

    url = 'https://api.meraki.com/api/v0/networks/{0}/airMarshal?timespan=3600'
    headers = {'X-Cisco-Meraki-API-Key': api_key, 'Content-Type': 'application/json'}

    async with trio.open_nursery() as nursery:
        for i in networkIds:
            nursery.start_soon(fetch, url.format(i), headers)

    print("Total time:", time.time() - start_time)



async def fetch(url, headers):
    print("Start: ", url)
    response = await s.get(url, headers=headers)
    print("Finished: ", url, len(response.content), response.status_code)




if __name__ == "__main__":
    trio.run(main)

当我运行 nursery.start_soon(fetch...) 时，我在 fetch 中打印数据，但是如何 return 打印数据？我没有看到任何类似于 asyncio.gather(*tasks) 函数的内容。

此外，我可以将会话数限制为 1-4，这有助于降低到每秒 5 API 的限制以下，但我想知道是否有内置的方法来确保不再超过 5 API 秒被调用？

Answer 1

返回数据：将网络 ID 和字典传递给 fetch 任务：

async def main():
    …
    results = {}
    async with trio.open_nursery() as nursery:
        for i in networkIds:
            nursery.start_soon(fetch, url.format(i), headers, results, i)
    ## results are available here

async def fetch(url, headers, results, i):
    print("Start: ", url)
    response = await s.get(url, headers=headers)
    print("Finished: ", url, len(response.content), response.status_code)
    results[i] = response

或者，创建一个 trio.Queue 给你 put 结果；然后您的主要任务可以从队列中读取结果。

API 限制：创建一个 trio.Queue(10) 并按照这些行开始任务：

async def limiter(queue):
    while True:
        await trio.sleep(0.2)
        await queue.put(None)

将该队列作为另一个参数传递给 fetch，并在每次 API 调用之前调用 await limit_queue.get()。

Answer 2

When I run nursery.start_soon(fetch...) , I am printing data within fetch, but how do I return the data? I didn't see anything similar to asyncio.gather(*tasks) function.

你问的是两个不同的问题，所以我只回答这个问题。 Matthias 已经回答了你的其他问题。

当您调用 start_soon() 时，您是在要求 Trio 运行在后台执行任务，然后继续执行。这就是为什么 Trio 能够同时运行 fetch() 多次。但是因为 Trio 继续运行，所以无法像 Python 函数通常那样 "return" 结果。它甚至 return 会去哪里？

您可以使用队列让 fetch() 个任务将结果发送到另一个任务以进行额外处理。

创建队列：

response_queue = trio.Queue()

当您开始获取任务时，将队列作为参数传递，并在完成后将一个哨兵发送到队列：

async with trio.open_nursery() as nursery:
    for i in networkIds:
        nursery.start_soon(fetch, url.format(i), headers)
await response_queue.put(None)

下载 URL 后，将响应放入队列：

async def fetch(url, headers, response_queue):
    print("Start: ", url)
    response = await s.get(url, headers=headers)
    # Add responses to queue
    await response_queue.put(response)
    print("Finished: ", url, len(response.content), response.status_code)

进行上述更改后，您的提取任务会将响应放入队列中。现在您需要从队列中读取响应以便处理它们。您可以添加一个新函数来执行此操作：

async def process(response_queue):
    async for response in response_queue:
        if response is None:
            break
        # Do whatever processing you want here.

您应该在启动任何提取任务之前将此处理函数作为后台任务启动，以便它会在收到响应后立即处理。

在 Trio 文档的 Synchronizing and Communicating Between Tasks 部分阅读更多内容。

Answer 3

从技术上讲，trio.Queue 已在 trio 0.9 中弃用。它已被替换为 trio.open_memory_channel。

简短示例：

sender, receiver = trio.open_memory_channel(len(networkIds)
async with trio.open_nursery() as nursery:
    for i in networkIds:
        nursery.start_soon(fetch, sender, url.format(i), headers)

async for value in receiver:
    # Do your job here
    pass

并且在您的 fetch 函数中，您应该在某处调用 async sender.send(value)。

Answer 4

基于，可以定义如下函数：

async def gather(*tasks):

    async def collect(index, task, results):
        task_func, *task_args = task
        results[index] = await task_func(*task_args)

    results = {}
    async with trio.open_nursery() as nursery:
        for index, task in enumerate(tasks):
            nursery.start_soon(collect, index, task, results)
    return [results[i] for i in range(len(tasks))]

然后您可以通过简单地修补 trio（添加 gather 函数）以与 asyncio 完全相同的方式使用 trio：

import trio
trio.gather = gather

这是一个实际的例子：

async def child(x):
    print(f"Child sleeping {x}")
    await trio.sleep(x)
    return 2*x

async def parent():
    tasks = [(child, t) for t in range(3)]
    return await trio.gather(*tasks)

print("results:", trio.run(parent))

如何在 Trio 中收集任务结果？

How to gather task results in Trio?

python-3.x

python-trio