标签: python-asyncio

为了防止上下文切换,我想创建一个大循环来同时服务于网络连接和一些例程.

这是正常功能的实现:

import asyncio
import time


def hello_world(loop):
    print('Hello World')
    loop.call_later(1, hello_world, loop)

def good_evening(loop):
    print('Good Evening')
    loop.call_later(1, good_evening, loop)

print('step: asyncio.get_event_loop()')
loop = asyncio.get_event_loop()

print('step: loop.call_soon(hello_world, loop)')
loop.call_soon(hello_world, loop)
print('step: loop.call_soon(good_evening, loop)')
loop.call_soon(good_evening, loop)

try:
    # Blocking call interrupted by loop.stop()
    print('step: loop.run_forever()')
    loop.run_forever()
except KeyboardInterrupt:
    pass
finally:
    print('step: loop.close()')
    loop.close()

这是协同程序的实现:

import asyncio


@asyncio.coroutine
def hello_world():
    while True:
        yield from asyncio.sleep(1)
        print('Hello World')

@asyncio.coroutine
def good_evening():
    while True:
        yield from asyncio.sleep(1)
        print('Good Evening')

print('step: asyncio.get_event_loop()')
loop = asyncio.get_event_loop() …

我正在使用asyncioPython 3.5中的使用流程,但我还没有看到我应该做什么await以及我不应该做的事情或者可以忽略不计的事情的描述.我是否必须在"这是一个IO操作,因此应该await编辑" 方面使用我的最佳判断？

我怎样才能异步插入任务来运行 asyncio在另一个线程中事件循环中运行？

我的动机是在解释器中支持交互式异步工作负载.我无法阻止主REPL线程.

例

我目前有缺陷的理解说,以下应该有效.为什么不呢？什么是实现上述目标的更好方法？

import asyncio
from threading import Thread

loop = asyncio.new_event_loop()

def f(loop):
    asyncio.set_event_loop(loop)
    loop.run_forever()

t = Thread(target=f, args=(loop,))
t.start()    

@asyncio.coroutine
def g():
    yield from asyncio.sleep(1)
    print('Hello, world!')

asyncio.async(g(), loop=loop)

我安装了以emperor模式运行的uWSGI.Vassals使用不同的python版本,因此我不能在uWSGI二进制文件中嵌入python插件.

有了这个,我想在其中一个封锁中使用asyncio循环引擎,但我无法弄清楚如何运行非嵌入的asyncio插件和greenlet插件.

到目前为止我尝试过的:

• 将asyncio和greenlet嵌入到uWSGI中,使用:

  CFLAGS="-I/usr/local/include/python3.4" make PYTHON=python3.4 asyncio
  

  Run Code Online (Sandbox Code Playgroud)

  建立uWSGI.但这也将嵌入python插件,我不想要它.

• 使用以下方法构建asyncio和uwsgi作为外部插件:

  PYTHON=python3.4 ./uwsgi --build-plugin "plugins/greenlet greenlet"
  PYTHON=python3.4 ./uwsgi --build-plugin "plugins/greenlet greenlet"
  

  Run Code Online (Sandbox Code Playgroud)

  构建插件,但插件无法加载:

  /usr/local/lib/uwsgi/asyncio_plugin.so:unfined defined symbol:up
  /usr/local/lib/uwsgi/greenlet_plugin.so:undefined symbol:up

  在日志中.

  //编辑

  我已经发现必须在设置中的asyncio和greenlet之前启用python插件,因此错误不再发生,但greenlet不起作用,当前greenlet中没有父级.

我还能尝试做些什么？我敢打赌,在python3.4插件中嵌入asyncio和greenlet会起作用,但我不知道该怎么做或者甚至可能.

目前我正在使用内置所需插件的第二个皇帝,但由于平台的限制,我不能再使用该解决方案了.

如何使用aiohttp在客户端设置每秒的最大请求数(限制它们)？

我目前在关闭应用程序的CTRL-C期间关闭asyncio协同程序时遇到问题.以下代码是我现在所拥有的精简版:

#!/usr/bin/env python
# -*- coding: UTF-8 -*-
import asyncio
import time
import functools
import signal


class DummyProtocol(asyncio.Protocol):

    def __init__(self, *args, **kwargs):
        self._shutdown = asyncio.Event()
        self._response = asyncio.Queue(maxsize=1)
        super().__init__(*args, **kwargs)

    def connection_made(self, transport):
        self.transport = transport

    def close(self):
        print("Closing protocol")
        self._shutdown.set()

    def data_received(self, data):

        #data = b'OK MPD '

        # Start listening for commands after a successful handshake
        if data.startswith(b'OK MPD '):
            print("Ready for sending commands")
            self._proxy_task = asyncio.ensure_future(self._send_commands())
            return

        # saving response for later consumption in self._send_commands
        self._response.put_nowait(data)

    async …

假设我们有一堆下载链接,每个链接可能需要不同的下载时间.我只允许使用最多3个连接下载.现在,我想确保使用asyncio有效地执行此操作.

这就是我想要实现的目标:在任何时候,尽量确保我至少运行3次下载.

Connection 1: 1---------7---9---
Connection 2: 2---4----6-----
Connection 3: 3-----5---8-----

数字代表下载链接,而连字符代表等待下载.

这是我正在使用的代码

from random import randint
import asyncio

count = 0


async def download(code, permit_download, no_concurrent, downloading_event):
    global count
    downloading_event.set()
    wait_time = randint(1, 3)
    print('downloading {} will take {} second(s)'.format(code, wait_time))
    await asyncio.sleep(wait_time)  # I/O, context will switch to main function
    print('downloaded {}'.format(code))
    count -= 1
    if count < no_concurrent and not permit_download.is_set():
        permit_download.set()


async def main(loop):
    global count
    permit_download = asyncio.Event()
    permit_download.set()
    downloading_event = asyncio.Event()
    no_concurrent = 3 …

背景

我曾经在一个Python2系统上工作，该系统具有许多同步编写的自定义I / O代码，并使用线程进行了扩展。在某个时候，我们无法进一步扩展它，并意识到我们必须切换到异步编程。

• Twisted是最受欢迎的选择，但我们想避免其回调地狱。
• 它确实具有@inlineCallbacks装饰器，该装饰器与其他一些库一样，使用生成器魔术有效地实现了协程。那是可以忍受的，但感觉有点片状。
• 然后我们找到了gevent。您要做的只是：

from gevent import monkey
monkey.patch_all()

就像那样，您的所有标准I / O（套接字，数据库事务以及用纯Python编写的所有内容）实际上都是异步的，使用greenlet产生并在后台切换。

这不是完美的：

• 那时，它在Windows上运行不佳（并且今天仍然有一些限制）。幸运的是，我们在Linux上运行。
• 它不能猴子补丁C扩展，因此，例如，我们不能使用MySQLdb。幸运的是，有许多纯Python替代品，例如PyMySQL。

题

如今，Python 3更加流行，并且有了它-asyncio。就个人而言，我认为这很棒，但是最近有人问我，它比我们使用gevent做的更好吗？无法给出足够好的答案。

这听起来可能是主观的，但是我实际上是在寻找一个实际的用例，其中一个将大大优于另一个，或者允许另一个不这样做。到目前为止，这是我收集的注意事项：

1. 就像我说的那样，gevent在Windows上非常受限制。再说一次，我所知道的大多数生产代码都在Linux上运行。

   如果需要在Windows上运行，请使用asyncio。

2. Gevent无法猴子C扩展。但是，asyncio无法猴子修补任何东西。

   想象一下，出现了一种新的数据库技术，并且您想使用它，但是它没有一个纯Python库，因此您无法将其与Gevent集成。问题是，当没有可以与asyncio集成的io *库时，您就像被困一样！当然，有工作线程和执行程序，但这不是重点，无论如何在两种情况下都一样好。

3. 有人说这是个人喜好，但我认为可以说同步编程从本质上讲要比异步编程更容易（请考虑一下：您是否遇到过可以使用套接字的新手程序员，但是很难理解如何正确选择/投票，或考虑期货/承诺？您遇到过相反的情况吗？）。

   无论如何，我们不要去那里。我想解决这一点，因为它经常出现（这是关于reddit的讨论），但是我真正要关注的是您有实际理由使用其中一种的情况。

4. Asyncio是标准库的一部分。这是巨大的：这意味着它得到了很好的维护，良好的记录，并且每个人都知道它并默认使用它。

   但是，考虑到您需要使用Gevent的知识很少（并且它也得到很好的维护和记录），它似乎并不重要。因此，即使对于涉及期货的最复杂的方案，StackOverflow上也有多个答案，但根本不使用期货的可能性似乎也是可行的。

那么：asyncio盛行的一些具体用例是什么？当然，Guido和Python社区有充分的理由对此投入大量精力，甚至在语言中引入了新的关键字-我似乎找不到它们。

我正在开发一个应用程序,它使用LevelDB并使用多个长期存在的进程来执行不同的任务.

由于LevelDB只允许单个进程维护数据库连接,因此我们所有的数据库访问都通过特殊的数据库进程进行汇集.

要从另一个进程访问数据库,我们使用a BaseProxy.但是因为我们使用asyncio我们的代理不应该阻塞这些调用db进程然后最终从db读取的API.因此,我们使用执行程序在代理上实现API.

    loop = asyncio.get_event_loop()

    return await loop.run_in_executor(
        thread_pool_executor,
        self._callmethod,
        method_name,
        args,
    )

尽管这工作得很好,我不知道是否有到包裹一个更好的选择_callmethod的通话BaseProxy中ThreadPoolExecutor.

我理解它的方式,BaseProxy调用DB进程是等待IO的教科书示例,因此使用线程这似乎是不必要的浪费.

在一个完美的世界中,我假设async _acallmethod存在,BaseProxy但不幸的是,API不存在.

因此,我的问题基本归结为:在使用时,BaseProxy是否有更有效的替代方法来运行这些跨进程调用ThreadPoolExecutor？

在处理 FastAPI 请求时，我有一个受 CPU 限制的任务要对列表的每个元素执行。我想在多个 CPU 内核上进行此处理。

在 FastAPI 中执行此操作的正确方法是什么？我可以使用标准multiprocessing模块吗？到目前为止，我发现的所有教程/问题仅涵盖 I/O 绑定任务，如 Web 请求。