标签: python-asyncio

asyncio.gather并且asyncio.wait似乎有类似的用途:我有一堆异步的东西,我想要执行/等待(不一定等待一个在下一个开始之前完成).它们使用不同的语法,并且在一些细节上有所不同,但是对于我来说,拥有2个功能上具有如此巨大重叠的功能似乎非常不灵活.我错过了什么？

假设我们有一个虚函数:

async def foo(arg):
    result = await some_remote_call(arg)
    return result.upper()

有什么区别:

coros = []
for i in range(5):
    coros.append(foo(i))

loop = get_event_loop()
loop.run_until_complete(wait(coros))

和:

from asyncio import ensure_future

futures = []
for i in range(5):
    futures.append(ensure_future(foo(i)))

loop = get_event_loop()
loop.run_until_complete(wait(futures))

注意:该示例返回结果,但这不是问题的焦点.当返回值很重要时,请使用gather()而不是wait().

无论回报价值如何,我都在寻求清晰度ensure_future().wait(coros)并且wait(futures)都运行协同程序,所以何时以及为什么要将协程包装进去ensure_future？

基本上,使用Python 3.5运行一堆非阻塞操作的正确方法(tm)是async什么？

如果我想批量通话,如果需要额外的积分？例如,我需要调用some_remote_call(...)1000次,但我不想粉碎Web服务器/数据库/等1000个同时连接.这对于线程或进程池是可行的,但有没有办法做到这一点asyncio？

有时需要发生一些非关键的异步操作,但我不想等待它完成.在Tornado的协程实现中,您可以通过简单地省略yield关键字来"触发并忘记"异步功能.

我一直试图弄清楚如何使用Python 3.5中发布的新async/ await语法来"解雇" .例如,简化的代码段:

async def async_foo():
    print("Do some stuff asynchronously here...")

def bar():
    async_foo()  # fire and forget "async_foo()"

bar()

但是会发生什么,bar()从不执行,而是我们得到运行时警告:

RuntimeWarning: coroutine 'async_foo' was never awaited
  async_foo()  # fire and forget "async_foo()"

我已经看过几个关于asyncio的基本Python 3.5教程,它们以各种方式执行相同的操作.在这段代码中:

import asyncio  

async def doit(i):
    print("Start %d" % i)
    await asyncio.sleep(3)
    print("End %d" % i)
    return i

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    #futures = [asyncio.ensure_future(doit(i), loop=loop) for i in range(10)]
    #futures = [loop.create_task(doit(i)) for i in range(10)]
    futures = [doit(i) for i in range(10)]
    result = loop.run_until_complete(asyncio.gather(*futures))
    print(result)

上面定义futures变量的所有三个变体都实现了相同的结果; 我能看到的唯一区别是,对于第三个变体,执行是乱序的(在大多数情况下这应该无关紧要).还有其他区别吗？有些情况下我不能只使用最简单的变体(协程的简单列表)吗？

我发现在Python 3.4中,很少有用于多处理/线程的不同库:多处理与线程和asyncio.

但我不知道使用哪一个或是"推荐的".他们做同样的事情,还是不同？如果是这样,哪一个用于什么？我想编写一个在我的计算机中使用多核的程序.但我不知道我应该学习哪个图书馆.

使用Python 3.4 asyncio库编写代码单元测试的最佳方法是什么？假设我想测试TCP客户端(SocketConnection):

import asyncio
import unittest

class TestSocketConnection(unittest.TestCase):
    def setUp(self):
        self.mock_server = MockServer("localhost", 1337)
        self.socket_connection = SocketConnection("localhost", 1337)

    @asyncio.coroutine
    def test_sends_handshake_after_connect(self):
        yield from self.socket_connection.connect()
        self.assertTrue(self.mock_server.received_handshake())

当使用默认测试运行器运行此测试用例时,测试将始终成功,因为该方法仅执行到第一yield from条指令,之后它在执行任何断言之前返回.这导致测试总是成功.

是否有预先构建的测试运行器能够处理这样的异步代码？

我正在尝试Task使用Python 3的相对较新的asyncio模块正确理解和实现两个并发运行的对象.

简而言之,asyncio似乎旨在Task通过事件循环处理异步进程和并发执行.它促进了await(在异步函数中应用)作为无回调方式的使用,等待和使用结果,而不会阻塞事件循环.(期货和回调仍然是一个可行的选择.)

它还提供了asyncio.Task()类,一个专门Future 用于包装协同程序的子类.优选地通过使用该asyncio.ensure_future()方法来调用.asyncio任务的预期用途是允许独立运行的任务与同一事件循环中的其他任务"同时"运行.我的理解是Tasks连接到事件循环,然后自动保持在await语句之间驱动协程.

我喜欢能够使用并发任务而不需要使用其中一个Executor类的想法,但我没有找到很多关于实现的详细说明.

这就是我目前正在做的事情:

import asyncio

print('running async test')

async def say_boo():
    i = 0
    while True:
        await asyncio.sleep(0)
        print('...boo {0}'.format(i))
        i += 1

async def say_baa():
    i = 0
    while True:
        await asyncio.sleep(0)
        print('...baa {0}'.format(i))
        i += 1

# wrap in Task object
# -> automatically attaches to event loop and executes
boo = asyncio.ensure_future(say_boo())
baa = …

这个问题是由我的另一个问题推动的:如何在cdef中等待？

网上有大量的文章和博客文章asyncio,但它们都非常肤浅.我找不到任何有关如何asyncio实际实现的信息,以及I/O异步的原因.我试图阅读源代码,但它是成千上万行不是最高级别的C代码,其中很多处理辅助对象,但最重要的是,很难在Python语法和它将翻译的C代码之间建立连接成.

Asycnio自己的文档甚至没那么有用.没有关于它是如何工作的信息,只有关于如何使用它的一些指导,这些指导有时也会误导/写得很差.

我熟悉Go的coroutines实现,并且希望Python做同样的事情.如果是这种情况,我在上面链接的帖子中出现的代码就可以了.既然没有,我现在正试图找出原因.到目前为止我最好的猜测如下,请纠正我错在哪里:

1. 表单的过程定义async def foo(): ...实际上被解释为类继承的方法coroutine.
2. 也许,async def实际上是通过await语句拆分成多个方法,其中调用这些方法的对象能够跟踪到目前为止通过执行所做的进度.
3. 如果上述情况属实,那么,从本质上讲,协程的执行归结为一些全局管理器调用协程对象的方法(循环？).
4. 全局管理器以某种方式(如何？)了解I/O操作何时由Python(仅？)代码执行,并且能够在当前执行方法放弃控制之后选择一个待执行的协程方法执行(命中await语句) ).

换句话说,这是我尝试将某些asyncio语法"贬低"为更容易理解的东西:

async def coro(name):
    print('before', name)
    await asyncio.sleep()
    print('after', name)

asyncio.gather(coro('first'), coro('second'))

# translated from async def coro(name)
class Coro(coroutine):
    def before(self, name):
        print('before', name)

    def after(self, name):
        print('after', name)

    def __init__(self, name):
        self.name = name
        self.parts = self.before, self.after
        self.pos = 0

    def __call__():
        self.parts[self.pos](self.name)
        self.pos += 1

    def …

如何await在构造函数或类体中定义类？

例如我想要的:

import asyncio

# some code


class Foo(object):

    async def __init__(self, settings):
        self.settings = settings
        self.pool = await create_pool(dsn)

foo = Foo(settings)
# it raises:
# TypeError: __init__() should return None, not 'coroutine'

或者类body属性的示例:

class Foo(object):

    self.pool = await create_pool(dsn)  # Sure it raises syntax Error

    def __init__(self, settings):
        self.settings = settings

foo = Foo(settings)

我的解决方案(但我希望看到更优雅的方式)

class Foo(object):

    def __init__(self, settings):
        self.settings = settings

    async def init(self):
        self.pool = await create_pool(dsn)

foo = Foo(settings)
await foo.init()

几乎每个人都知道他们第一次看到Python中的线程时,GIL会让那些真正希望并行处理的人生活变得悲惨 - 或者至少给它一个机会.

我目前正在研究像Reactor模式这样的实现.实际上我想在一个线程上监听传入的套接字连接,当有人试图连接时,接受该连接并将其传递给另一个线程进行处理.

我(还)不确定我可能会遇到什么样的负担.我知道目前设置的传入消息上限为2MB.从理论上讲,我们每秒可以获得数千(虽然我不知道我们是否已经看到过类似的东西).处理消息所花费的时间并不是非常重要,但显然更快会更好.

我正在研究Reactor模式,并开发了一个使用该multiprocessing库的小例子(至少在测试中)似乎工作得很好.但是,现在/很快我们将有asyncio库可用,它将为我处理事件循环.

是否有任何东西可以通过组合asyncio和咬我multiprocessing？