今天,我找到了一个名为trio的库,它表示自己是一个针对人类的异步API.这些词与requests' 有点类似.由于这requests是一个很好的图书馆,我想知道它的优点是什么trio.

关于它的文章不多,我只是找到一篇文章讨论curio和asyncio.令我惊讶的是,trio它本身甚至比curio(下一代古玩)更好.

阅读完一半文章后,我找不到这两个异步框架之间的核心区别.它只是给出了一些实例curio比实现更方便的例子asyncio.但底层结构几乎相同(基于回调,我认为所有异步IO框架都基于回调而没有任何异常.)

那么有人能给我一个理由我必须接受trio或者curio更好asyncio吗？或者解释一下为什么我应该选择trio而不是内置asyncio？

我正在写一个会在其生命周期中产生任务的类.因为我正在使用Trio,所以我不能在没有托儿所的情况下产生任务.我的第一个想法是self._nursery在我的课堂上有一个我可以将任务产生的东西.但似乎托儿所对象只能在上下文管理器中使用,因此它们始终在创建它们的同一范围内关闭.我不想从外面传递托儿所,因为它是一个实现细节,但我确实希望我的对象能够产生与对象一样长的任务(例如心跳任务).

我如何使用Trio编写这样一个具有长期后台任务的类？

我编写了一个脚本，它使用托儿所和 asks 模块循环并根据循环变量调用 API。我收到回复，但不知道如何像使用 asyncio 一样返回数据。

我还有一个关于将 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: ", …

我正在做一些例子来了解它是如何异步地运行python的.我读了三重奏的文档,我还以为只有一个任务可在循环每次执行,并且在每一个checkpoint在scheduler决定哪些任务将被执行.

我做了一个测试它的例子,在三个例子中,我没有在我生成的子节点中使用任何检查点,nursery但是这个例子比同步版本快两倍.

异步示例:

import time
import trio

results = []

async def sum_numbers(first, last):
    result = 0
    for i in range(first, last):
        result += i
    results.append(result)

async def main():
    start_time = time.time()
    async with trio.open_nursery() as nursery:
        nursery.start_soon(sum_numbers, 0, 50000000)
        nursery.start_soon(sum_numbers, 50000000, 100000000)

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

trio.run(main)

结果:

4999999950000000
Total time: 4.150018930435181

同步示例:

import time

start_time = time.time()
result = 0
for i in range(0, 100000000):
    result += i

print(result)
print("Total …

我想使用 asyncio (/curio/trio) 和 pytest 同时运行多个测试，但我找不到任何相关信息。我需要自己安排吗？如果我这样做，有没有办法有一个很好的输出来分离（子）测试用例？

这是我正在尝试的一个小玩具示例：

import pytest
import time
import asyncio

pytestmark = pytest.mark.asyncio
expected_duration = 1
accepted_error = 0.1

async def test_sleep():
    start = time.time()
    time.sleep(expected_duration)
    duration = time.time() - start
    assert abs(duration-expected_duration) < accepted_error

async def test_async_sleep():
    start = time.time()
    await asyncio.sleep(expected_duration)
    duration = time.time() - start
    assert abs(duration-expected_duration) < accepted_error

雷蒙德赫廷杰了对并发性的演讲在蟒蛇,其中的一个例子看上去像是:

import urllib.request

sites = [
    'https://www.yahoo.com/',
    'http://www.cnn.com',
    'http://www.python.org',
    'http://www.jython.org',
    'http://www.pypy.org',
    'http://www.perl.org',
    'http://www.cisco.com',
    'http://www.facebook.com',
    'http://www.twitter.com',
    'http://www.macrumors.com/',
    'http://arstechnica.com/',
    'http://www.reuters.com/',
    'http://abcnews.go.com/',
    'http://www.cnbc.com/',
]

for url in sites:
    with urllib.request.urlopen(url) as u:
        page = u.read()
        print(url, len(page))

基本上我们追求这些链接并打印接收字节的数量,运行大约需要20秒.

今天我发现三人图书馆有相当友好的api.但是,当我试图在这个相当基本的例子中使用它时,我没有做对.

第一次尝试(运行大约相同的20秒):

import urllib.request
import trio, time

sites = [
    'https://www.yahoo.com/',
    'http://www.cnn.com',
    'http://www.python.org',
    'http://www.jython.org',
    'http://www.pypy.org',
    'http://www.perl.org',
    'http://www.cisco.com',
    'http://www.facebook.com',
    'http://www.twitter.com',
    'http://www.macrumors.com/',
    'http://arstechnica.com/',
    'http://www.reuters.com/',
    'http://abcnews.go.com/',
    'http://www.cnbc.com/',
]


async def show_len(sites):
    t1 = time.time()
    for url in sites:
        with urllib.request.urlopen(url) as u:
            page …

asyncio有StreamReader.readline(),允许这样的东西:

while True:
    line = await reader.readline()
    ...

(我认为async for在asyncio中没有,但那将是明显的演变)

如何实现三重奏的等效？

在三人组0.9中,我没有直接看到任何高级别的支持.我所看到的只是ReceiveStream.receive_some()返回任意大小的二进制块; 对我来说,解码并将其转换为线性似乎并非易事.我可以使用标准库函数或代码片段吗？我发现io stdlib模块看起来很有前途,但我认为没有办法提供"feed"方法.

我正在写一些异步库，并决定支持asyncio和trio并发库来运行它。无论选择哪个库，我都有一些代码试图变得聪明并做正确的事情。

如何检测其中哪一个用于运行我的代码？两者都可以吗？

我有一个监听特定端口的异步函数。我想一次在几个端口上运行该函数，当用户想要停止在特定端口上侦听时，停止在该端口上侦听的函数。

之前我使用 asyncio 库来完成此任务，并通过创建以唯一 id 作为名称的任务来解决此问题。

asyncio.create_task(Func(), name=UNIQUE_ID)

由于 trio 使用 Nursery 来生成任务，我可以使用 Nursery.child_tasks 查看正在运行的任务，但这些任务没有办法命名它们，甚至没有办法按需取消任务

长话短说

由于 trio 没有取消特定任务的 cancel() 函数，因此如何手动取消任务。

我被告知以下代码不安全，因为不允许有一个从托儿所内部生成的异步生成器，除非它是异步上下文管理器。

T = TypeVar('T')

async def delay(interval: float, source: AsyncIterable[T]) -> AsyncIterable[T]:
    """Delays each item in source by an interval.

    Received items are temporarily stored in an unbounded queue, along with a timestamp, using
    a background task. The foreground task takes items from the queue, and waits until the
    item is older than the given interval and then yields it."""

    send_channel, receive_channel = trio.open_memory_channel(math.inf)

    async def pull_task():
        async with aclosing(source) as agen:
            async for item in agen:
                send_channel.send_nowait((item, trio.current_time() + …