标签: python-asyncio

考虑以下代码:

#!/usr/bin/env python
# coding=utf-8
from string import letters


def filter_upper(letters):
    for letter in letters:
        if letter.isupper():
            yield letter

def filter_selected(letters, selected):
    selected = set(map(str.lower, selected))
    for letter in letters:
        if letter.lower() in selected:
            yield letter


def main():
    stuff = filter_selected(filter_upper(letters), ['a', 'b', 'c'])
    print(list(stuff))

main()

这是从发电机构建的管道的图示.我经常在实践中使用这种模式来构建数据处理流程.这就像UNIX管道.

将生成器重构为每个暂停执行的协同程序的最优雅方法是什么yield？

UPDATE

我的第一次尝试是这样的:

#!/usr/bin/env python
# coding=utf-8

import asyncio

@asyncio.coroutine
def coro():
    for e in ['a', 'b', 'c']:
        future = asyncio.Future()
        future.set_result(e)
        yield from future


@asyncio.coroutine
def coro2():
    a = yield …

我用的很asyncio漂亮aiohttp.主要的想法是我向服务器发出请求(它返回链接),然后我想从所有链接并行下载文件(类似于一个示例).

码:

import aiohttp
import asyncio

@asyncio.coroutine
def downloader(file):
    print('Download', file['title'])
    yield from asyncio.sleep(1.0) # some actions to download
    print('OK', file['title'])


def run():
    r = yield from aiohttp.request('get', 'my_url.com', True))
    raw = yield from r.json()
    tasks = []
    for file in raw['files']:
        tasks.append(asyncio.async(downloader(file)))
        asyncio.wait(tasks)

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(run())

但是,当我尝试运行它时,我有许多"下载..."输出和

Task was destroyed but it is pending!

而且没有'OK + filename'.

我该如何解决这个问题？

在线程中,我们有一个名为"线程上下文"的东西,我们可以在其中保存一些数据(状态)以便在特殊线程中进行访问.在asyncio中,我需要在当前执行路径中保存一些状态,以便所有后续协同程序都可以访问它.解决办法是什么？注意:我知道每个协同程序函数都是为asyncio中的执行路径实例化的,但由于某种原因,我无法将状态保存在函数属性中.(虽然这种方法反正不太好)

这是简化的代码,它使用python3协同程序并为SIGING和SIGTERM信号设置处理程序以正确停止作业:

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import argparse
import asyncio
import signal
import sys


def my_handler(signum, frame):
    print('Stopping')
    asyncio.get_event_loop().stop()
    # Do some staff
    sys.exit()


@asyncio.coroutine
def prob_ip(ip_addr):

    print('Ping ip:%s' % ip_addr)
    proc = yield from asyncio.create_subprocess_exec('ping', '-c', '3', ip_addr)
    ret_code = yield from proc.wait()
    if ret_code != 0:
        print("ip:%s doesn't responding" % ip_addr)
        # Do some staff
        yield from asyncio.sleep(2)
        # Do more staff
        yield from asyncio.sleep(16)


@asyncio.coroutine
def run_probing():

    print('Start probing')
    # Do some staff
    yield from …

PyPy是否支持aio和Python 3.5？

我需要性能PyPy和异步代码asyncio.我还需要async/await在我的代码中使用.那可能吗？

如果是这样,有什么细微差别？

我正在使用python3编写应用程序,并且我第一次尝试使用asyncio.我遇到的一个问题是我的一些协程阻止事件循环的时间比我想要的长.我试图在事件循环的顶部找到一些东西,它将显示运行每个协同程序所花费的壁/ CPU时间.如果没有任何已经存在的东西有没有人知道一种方法来添加钩子到事件循环,以便我可以进行测量？

我尝试过使用cProfile,它提供了一些有用的输出,但我对阻塞事件循环的时间更感兴趣,而不是总执行时间.

我在Python 3.6中编写了一个工具,它向几个API(带有各种端点)发送请求,并收集它们的解析来解析并将它们保存在数据库中.

我使用的API客户端具有请求URL 的同步版本,例如他们使用的

urllib.request.Request('...

或者他们使用Kenneth Reitz的Requests图书馆.

由于我的API调用依赖于请求URL的同步版本,因此整个过程需要几分钟才能完成.

现在我想将我的API调用包装在async/await(asyncio)中.我正在使用python 3.6.

我发现的所有示例/教程都要我将同步URL调用/ requests更改为异步版本(例如aiohttp).由于我的代码依赖于我未编写的API客户端(我无法更改),因此我需要保持该代码不受影响.

那么有没有办法将我的同步请求(阻塞代码)包装在async/await中以使它们在事件循环中运行？

我是Python新手中的asyncio.这在NodeJS中是不费脑子的.但是我不能用Python来解决这个问题.

我正在做一些例子来了解它是如何异步地运行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 …

我正在测试可能会陷入僵局的异步功能。我试图添加一个固定装置来限制该功能在引发故障之前只能运行5秒钟，但到目前为止还没有奏效。

设定：

pipenv --python==3.6
pipenv install pytest==4.4.1
pipenv install pytest-asyncio==0.10.0

码：

pipenv --python==3.6
pipenv install pytest==4.4.1
pipenv install pytest-asyncio==0.10.0

-

编辑：Mikhail的解决方案工作正常。但是，我找不到将其合并到灯具中的方法。

我只是注意到一些令人惊讶的事情。考虑以下示例：

import asyncio

async def wait_n(n):
    asyncio.sleep(n)

async def main(fn):
    print("meh")
    await fn(1)
    print("foo")

loop = asyncio.get_event_loop()
loop.run_until_complete(main(wait_n))

运行此命令时，我们正确地收到以下警告：

awaitable_lambda.py:5: RuntimeWarning: coroutine 'sleep' was never awaited

asyncio.sleep（n）

这是因为在wait_n我们中，我们asyncio.sleep(n)没有调用await。

但现在考虑第二个示例：

import asyncio

async def main(fn):
    print("meh")
    await fn(1)
    print("foo")

loop = asyncio.get_event_loop()
loop.run_until_complete(main(lambda n: asyncio.sleep(n)))

这次我们使用的是lambda，即使没有，代码也可以正常工作await。

我明白，我们不能用await一个Python里面lambda表情让这似乎是一个功能，可以提高工效，但它导致了我一些问题：

1. 这是如何工作的？这个简单的“注入”是否await在任何协程功能之前？
2. 这在某处有记录吗（PEP）？
3. 这还有其他含义吗？我们可以安全地从lambda表达式中调用协程函数，并依靠Python来为我们等待吗？