标签: 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中的执行路径实例化的,但由于某种原因,我无法将状态保存在函数属性中.(虽然这种方法反正不太好)

PyPy是否支持aio和Python 3.5？

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

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

我一直在尝试创建一个服务器进程，该进程从客户端进程异步接收输入文件路径和输出路径。服务器执行一些依赖于数据库的转换，但为了简单起见，我们假设它只是将所有内容都设为大写。这是服务器的一个玩具示例：

import asyncio
import aiofiles as aiof
import logging
import sys


ADDRESS = ("localhost", 10000)

logging.basicConfig(level=logging.DEBUG,
                    format="%(name)s: %(message)s",
                    stream=sys.stderr)

log = logging.getLogger("main")
loop = asyncio.get_event_loop()


async def server(reader, writer):
    log = logging.getLogger("process at {}:{}".format(*ADDRESS))
    paths = await reader.read()
    in_fp, out_fp = paths.splitlines()
    log.debug("connection accepted")
    log.debug("processing file {!r}, writing output to {!r}".format(in_fp, out_fp))
    async with aiof.open(in_fp, loop=loop) as inp, aiof.open(out_fp, "w", loop=loop) as out:
        async for line in inp:
            out.write(line.upper())
        out.flush()
    writer.write(b"done")
    await writer.drain()
    log.debug("closing")
    writer.close()
    return


factory = asyncio.start_server(server, *ADDRESS) …

为什么我们需要asyncio.coroutine装饰器？它提供什么功能？

例如：

# import asyncio
# @asyncio.coroutine
def gen():
    value = yield("Started")
    print(value)

a = gen()
a.send(None)
a.send("Done")

现在，如果我取消对前两行的注释并使用asyncio.coroutine装饰器，我仍然会得到相同的输出。

我的意思是这已经是一个coroutine- 一个可以暂停并带参数传入的函数。为什么我需要用另一个coroutineie装饰它asyncio.coroutine？

来自 python 3.4 的新手开发人员在这里。

我的天真理解只是async with在看到协程是上下文管理器时才使用关键字？

我想同时从网站上抓取数据，但我发现以下程序不是同时执行的。

async def return_soup(url):
    r = requests.get(url)
    r.encoding = "utf-8"
    soup = BeautifulSoup(r.text, "html.parser")

    future = asyncio.Future()
    future.set_result(soup)
    return future

async def parseURL_async(url):    
    print("Started to download {0}".format(url))
    soup = await return_soup(url)
    print("Finished downloading {0}".format(url))

    return soup

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
t = [parseURL_async(url_1), parseURL_async(url_2)]
loop.run_until_complete(asyncio.gather(*t))

但是，该程序仅在第一个内容完成后才开始下载第二个内容。如果我的理解是正确的，awaitawait 上的关键字await return_soup(url)等待函数完成，并在等待完成时将控制权返回给事件循环，从而使循环开始第二次下载。

一旦函数最终完成执行，其中的未来实例将获得结果值。

但是为什么这不能同时工作？我在这里缺少什么？

我试图通过 Flask 方法调用阻塞函数，但需要几秒钟，所以我想我可以做一些异步调用来加快速度，但它没有按预期工作。显然使用 asyncio 我不能只是在后台启动一个协程而不等待执行结束，也许我需要使用线程？或者使用 grequest 因为我的阻塞功能正在使用请求...

到目前为止，这是我的代码：

@app.route("/ressource", methods=["GET"])
def get_ressource():
    do_stuff()
    return make_response("OK",200)  

def do_stuff():
  # Some stuff
  fetch_ressource()


async def fetch_ressource():
    return await blocking_function()


def blocking_function():
  # Take 2-3 seconds
  result = request.get('path/to/remote/ressource')
  put_in_database(result)

我听说过 Celeri，但对于只有一个功能来说似乎有点矫枉过正。

我正在测试可能会陷入僵局的异步功能。我试图添加一个固定装置来限制该功能在引发故障之前只能运行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的解决方案工作正常。但是，我找不到将其合并到灯具中的方法。