标签: python-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 …

我正在使用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来解决这个问题.

使用下面的例子,如何future2使用future1一次future1完成的结果(没有阻止future3提交)？

from concurrent.futures import ProcessPoolExecutor
import time

def wait(seconds):
    time.sleep(seconds)
    return seconds

pool = ProcessPoolExecutor()

s = time.time()
future1 = pool.submit(wait, 5)
future2 = pool.submit(wait, future1.result())
future3 = pool.submit(wait, 10)

time_taken = time.time() - s
print(time_taken)

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

我使用以下函数强制协程同步运行:

import asyncio
import inspect
import types
from asyncio import BaseEventLoop
from concurrent import futures


def await_sync(coro: types.CoroutineType, timeout_s: int=None):
    """

    :param coro: a coroutine or lambda loop: coroutine(loop)
    :param timeout_s:
    :return:
    """
    loop = asyncio.new_event_loop()  # type: BaseEventLoop
    if not is_awaitable(coro):
        coro = coro(loop)
    if timeout_s is None:
        fut = asyncio.ensure_future(coro, loop=loop)
    else:
        fut = asyncio.ensure_future(asyncio.wait_for(coro, timeout=timeout_s, loop=loop), loop=loop)
    loop.run_until_complete(fut)
    return fut.result()

def is_awaitable(coro_or_future):
    if isinstance(coro_or_future, futures.Future):
        return coro_or_future
    elif asyncio.coroutines.iscoroutine(coro_or_future):
        return True
    elif asyncio.compat.PY35 and inspect.isawaitable(coro_or_future):
        return True …

我正在尝试使用asyncio来处理并发网络I/O. 在一个点上安排了大量的功能,这些功能在每个功能完成所需的时间上变化很大.然后,对于每个输出,在单独的过程中处理接收的数据.

处理数据的顺序是不相关的,因此,考虑到输出的潜在等待时间非常长,我希望await无论将来如何完成而不是预定义的顺序.

def fetch(x):
    sleep()

async def main():
    futures = [loop.run_in_executor(None, fetch, x) for x in range(50)]
    for f in futures:
       await f

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

通常,等待期货排队的顺序很好:

蓝色表示每个任务在执行程序队列中的时间,run_in_executor即已被调用,但该函数尚未执行,因为执行程序仅同时运行5个任务; 绿色是执行功能本身所花费的时间; 红色是等待所有先前期货所花费的时间await.

在我的情况下,函数的时间变化很大,等待队列中的先前期货等待时会有很多时间丢失,而我可以在本地处理GET输出.这使得我的系统空闲一段时间只是在几个输出同时完成时被淹没,然后跳回空闲等待更多请求完成.

有没有办法await在执行者中首先完成任何未来？

我是Python的新手.我用asyncio编写了一个简单的报废程序.这是我的代码片段

 loop = asyncio.get_event_loop()
 task = loop.create_task(conSpi.parse(arguments.url))
 value = loop.run_until_complete(asyncio.wait([task]))
 loop.close()

我想打印返回值的结果.收集打印变量的值,打印出类似这样的内容

 {<Task finished coro=<ConcurrentSpider.parse() done, 
 defined at /home/afraz/PycharmProjects/the-lab/concurrentspider.py:28> result=3>}

`

我怎样才能得到结果而不打印？

API通常具有用户必须遵循的速率限制。例如，让我们接受50个请求/秒。顺序请求需要0.5-1秒，因此太慢了，无法接近该限制。但是，使用aiohttp的并行请求超出了速率限制。

为了尽可能快地轮询API，需要对并行调用进行速率限制。

到目前为止，我发现的示例装饰了session.get，大致像这样：

session.get = rate_limited(max_calls_per_second)(session.get)

这对于顺序调用非常有效。尝试在并行调用中实现此功能无法按预期进行。

这是一些示例代码：

async with aiohttp.ClientSession() as session:
    session.get = rate_limited(max_calls_per_second)(session.get)
    tasks = (asyncio.ensure_future(download_coroutine(  
          timeout, session, url)) for url in urls)
    process_responses_function(await asyncio.gather(*tasks))

问题在于它将限制任务的排队速度。与的执行gather将或多或少地同时发生。两全其美；-)。

是的，我在aiohttp处发现了一个类似的问题：设置每秒的最大请求数，但没有答复回答限制请求速率的实际问题。同样，来自Quentin Pradet的博客文章仅在限制队列速率上起作用。

总结一下：如何限制并行请求的每秒aiohttp请求数？

在Python调试器中是否可以await任意调用async函数？

说我在某些main.py文件中有以下代码：

import asyncio

async def bar(x):
    return x + 1

async def foo():
    import ipdb; ipdb.set_trace()

asyncio.run(foo())

现在，我想bar()在调试器中测试带有某些参数的调用以测试结果。发生以下情况：

$ python3 main.py
> /Users/user/test/main.py(8)foo()
      7     import ipdb; ipdb.set_trace()
----> 8     return None
      9

ipdb> bar(1)
<coroutine object bar at 0x10404ae60>
main.py:1: RuntimeWarning: coroutine 'bar' was never awaited
  import asyncio
RuntimeWarning: Enable tracemalloc to get the object allocation traceback
ipdb> await bar(1)
*** SyntaxError: 'await' outside function

当然，我可以通过在自己的x = await bar(1)之上 …