标签: python-asyncio

我试图了解asyncio库,特别是使用套接字.我写了一些代码以试图获得理解,

我想异步运行发送器和接收器套接字.我到了最后一个发送所有数据的地步,但是我必须再运行一个循环.看看如何做到这一点,我从stackoverflow找到了这个链接,我在下面实现了 - 但是这里发生了什么？是否有更好/更理智的方式来做到这一点,而不是打电话stop跟随run_forever？

stop()事件循环中的文档是:

停止运行事件循环.

调用stop()之前安排的每个回调都将运行.调用stop()后调度的回调将不会运行.但是,如果稍后再次调用run_forever(),那些回调将会运行.

而且run_forever()的文件是:

运行直到调用stop().

问题:

• 为什么世界上run_forever唯一的出路run_once？这甚至没有意义
• 有一个更好的方法吗？
• 我的代码看起来像是使用asyncio库编程的合理方式吗？
• 是否有更好的方法将事件添加到事件循环中asyncio.async()？loop.create_task在我的Linux系统上出错.

https://gist.github.com/cloudformdesign/b30e0860497f19bd6596

我写了一个测试程序来尝试使用create_task(),需要等到创建的任务完成.

我尝试使用loop.run_until_complete()等待任务完成,但它导致跟踪错误.

/Users/jason/.virtualenvs/xxx/bin/python3.5 /Users/jason/asyncio/examples/hello_coroutine.py
    Traceback (most recent call last):
    Test
      File "/Users/jason/asyncio/examples/hello_coroutine.py", line 42, in <module>
    Hello World, is a task
        loop.run_until_complete(test.greet_every_two_seconds())
      File "/Users/jason/asyncio/asyncio/base_events.py", line 373, in run_until_complete
        return future.result()
      File "/Users/jason/asyncio/asyncio/futures.py", line 274, in result
        raise self._exception
      File "/Users/jason/asyncio/asyncio/tasks.py", line 240, in _step
        result = coro.send(None)
      File "/Users/jason/asyncio/examples/hello_coroutine.py", line 33, in greet_every_two_seconds
        self.a()
      File "/Users/jason/asyncio/examples/hello_coroutine.py", line 26, in a
        t = loop.run_until_complete(self.greet_every_one_seconds(self.db_presell))
      File "/Users/jason/asyncio/asyncio/base_events.py", line 361, in run_until_complete
        self.run_forever()
      File "/Users/jason/asyncio/asyncio/base_events.py", line 326, in run_forever …

我正在尝试使用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>}

`

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

我正在尝试为python实现异步,分布式计算引擎,它与jupyter notebook兼容.该系统应该基于"推送通知"方法,这使得(几乎,我希望)不可能允许用户等待特定的计算结果(即,给定笔记本单元的块执行直到具有预期结果的消息被传递).确切地说,我试图:

1. 将新任务添加到jupyter notebook事件循环(该任务定期检查特定msg是否已在循环中到达,当msg到达时中断)
2. 阻止当前单元格等待任务完成.
3. 仍然能够处理传入的消息(使用RabbitMQ,Pika,来自http://pika.readthedocs.io/en/0.10.0/examples/asynchronous_consumer_example.html的略微修改的代码)

我准备好了解决我的问题的笔记本:https://github.com/SLEEP-MAN/RabbitMQ_jupyterNotebook_asyncio

有任何想法吗？是否有可能(可能是一些IPython/IpyKernel魔法;>？),或者我必须将我的方法改为180度？

我想在我的Python库中使用select.epoll().

不幸的是epoll并不是随处可见的.

我需要一种方法来回退到select.select().

我试图在pypi找到一些东西,但未能找到匹配的包:https://pypi.python.org/pypi？%3Aaction = search&term = epoll&subsmit = search

我怎么能解决这个"从epoll回退选择是否epoll不可用"？

我已经阅读了所有可以找到的有关如何优雅地处理带有asyncio事件循环并以Ctrl-C终止的脚本的文章，而且在我不打印一个或多个回溯的情况下，我无法使其中的任何一个都能正常工作这样做。答案几乎无处不在，我还无法在这个小脚本中实现其中的任何一个：

import asyncio
import datetime
import functools
import signal


async def display_date(loop):
    end_time = loop.time() + 5.0
    while True:
        print(datetime.datetime.now())
        if (loop.time() + 1.0) >= end_time:
            break
        await asyncio.sleep(1)


def stopper(signame, loop):
    print("Got %s, stopping..." % signame)
    loop.stop()


loop = asyncio.get_event_loop()
for signame in ('SIGINT', 'SIGTERM'):
    loop.add_signal_handler(getattr(signal, signame), functools.partial(stopper, signame, loop))

loop.run_until_complete(display_date(loop))
loop.close()

我想让脚本退出，而不在Ctrl-C（或通过发送的SIGTERM / SIGINT kill）之后打印任何回溯。这段代码RuntimeError: Event loop stopped before Future completed.以我根据先前的答案尝试过的许多其他形式打印，我得到了很多其他类型的异常类和错误消息，却不知道如何解决它们。上面的代码现在很小，但是我之前所做的一些尝试不过是什么，而且都不是正确的。

如果您能够修改脚本以使其优雅地终止，那么不胜感激地向您解释为什么这样做是正确的方法。

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请求数？

我在模拟aiohttp.client.ClientSession.get上下文管理器时遇到了一些麻烦.我找到了一些文章,这里有一个似乎正在起作用的例子:第1条

所以我要测试的代码:

async_app.py

import random
from aiohttp.client import ClientSession

async def get_random_photo_url():
    while True:
        async with ClientSession() as session:
            async with session.get('random.photos') as resp:
                json = await resp.json()
        photos = json['photos']
        if not photos:
            continue
        return random.choice(photos)['img_src']

并测试:

test_async_app.py

from asynctest import CoroutineMock, MagicMock, patch

from asynctest import TestCase as TestCaseAsync

from async_app import get_random_photo_url


class AsyncContextManagerMock(MagicMock):
    async def __aenter__(self):
        return self.aenter

    async def __aexit__(self, *args):
        pass

class TestAsyncExample(TestCaseAsync):
    @patch('aiohttp.client.ClientSession.get', new_callable=AsyncContextManagerMock)
    async def test_call_api_again_if_photos_not_found(self, mock_get):
        mock_get.return_value.aenter.json = CoroutineMock(side_effect=[{'photos': []}, …

在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)之上 …