73

几乎每个人在第一次看到 Python 中的线程时都知道,对于那些真正想要并行处理的人来说,GIL 让他们的生活变得悲惨——或者至少给它一个机会。

我目前正在考虑实现类似 Reactor 模式的东西。实际上,我想在一个类线程上监听传入的套接字连接,当有人尝试连接时,接受该连接并将其传递给另一个类线程进行处理。

我(还)不确定我可能面临什么样的负载。我知道目前对传入消息设置了 2MB 的上限。从理论上讲,我们每秒可以得到数千(尽管我不知道实际上我们是否见过类似的东西)。处理消息所花费的时间并不是重要,但显然越快越好。

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

有什么可以通过结合asyncio和来咬我的东西multiprocessing吗?

4

3 回答 3

89

您应该能够安全地组合asyncio并且multiprocessing没有太多麻烦,尽管您不应该multiprocessing直接使用。asyncio(以及任何其他基于事件循环的异步框架)的主要罪过是阻塞事件循环。如果你尝试multiprocessing直接使用,任何时候你阻塞等待子进程,你都会阻塞事件循环。显然,这很糟糕。

避免这种情况的最简单方法是使用BaseEventLoop.run_in_executorconcurrent.futures.ProcessPoolExecutor. ProcessPoolExecutor是一个使用 实现的进程池multiprocessing.Process,但asyncio内置支持在其中执行函数而不阻塞事件循环。这是一个简单的例子:

import time
import asyncio
from concurrent.futures import ProcessPoolExecutor

def blocking_func(x):
   time.sleep(x) # Pretend this is expensive calculations
   return x * 5

@asyncio.coroutine
def main():
    #pool = multiprocessing.Pool()
    #out = pool.apply(blocking_func, args=(10,)) # This blocks the event loop.
    executor = ProcessPoolExecutor()
    out = yield from loop.run_in_executor(executor, blocking_func, 10)  # This does not
    print(out)

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

在大多数情况下,仅此功能就足够了。如果您发现自己需要来自multiprocessingQueue、等的其他构造Event,则有一个名为(完全公开:我编写的)Manager的第三方库,它提供所有数据结构的兼容版本。这是一个演示示例:aioprocessingasynciomultiprocessing

import time
import asyncio
import aioprocessing
import multiprocessing

def func(queue, event, lock, items):
    with lock:
        event.set()
        for item in items:
            time.sleep(3)
            queue.put(item+5)
    queue.close()

@asyncio.coroutine
def example(queue, event, lock):
    l = [1,2,3,4,5]
    p = aioprocessing.AioProcess(target=func, args=(queue, event, lock, l)) 
    p.start()
    while True:
        result = yield from queue.coro_get()
        if result is None:
            break
        print("Got result {}".format(result))
    yield from p.coro_join()

@asyncio.coroutine
def example2(queue, event, lock):
    yield from event.coro_wait()
    with (yield from lock):
        yield from queue.coro_put(78)
        yield from queue.coro_put(None) # Shut down the worker

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    queue = aioprocessing.AioQueue()
    lock = aioprocessing.AioLock()
    event = aioprocessing.AioEvent()
    tasks = [ 
        asyncio.async(example(queue, event, lock)),
        asyncio.async(example2(queue, event, lock)),
    ]   
    loop.run_until_complete(asyncio.wait(tasks))
    loop.close()
于 2015-03-19T14:50:04.227 回答
8

是的,有很多可能(或可能不会)咬你。

  • 当您运行类似于asyncio它期望在一个线程或进程上运行的东西时。这(本身)不适用于并行处理。您必须以某种方式分配工作,同时将 IO 操作(特别是套接字上的操作)留在单个线程/进程中。
  • 虽然您将单个连接移交给不同的处理程序进程的想法很好,但很难实现。asyncio第一个障碍是您需要一种在不关闭连接的情况下将其拉出的方法。下一个障碍是您不能简单地将文件描述符发送到不同的进程,除非您使用来自 C 扩展的特定于平台(可能是 Linux)的代码。
  • 请注意,multiprocessing已知该模块会创建许多线程用于通信。大多数情况下,当您使用通信结构(例如Queues)时,会产生一个线程。不幸的是,这些线程并非完全不可见。例如,它们可能无法干净地拆除(当您打算终止程序时),但根据它们的数量,资源使用可能本身就很明显。

如果您真的打算在单个进程中处理单个连接,我建议您检查不同的方法。例如,您可以将套接字置于侦听模式,然后同时并行接受来自多个工作进程的连接。一旦工作人员完成处理请求,它就可以接受下一个连接,因此与为每个连接分叉一个进程相比,您仍然使用更少的资源。Spamassassin 和 Apache (mpm prefork) 可以使用这种工作模型。根据您的用例,它最终可能会更容易和更健壮。具体来说,您可以让您的工作人员在服务于配置数量的请求后死亡,并由主进程重新生成,从而消除内存泄漏的大部分负面影响。

于 2014-01-17T06:03:42.050 回答
1

请参阅 PEP 3156,特别是有关线程交互的部分:

http://www.python.org/dev/peps/pep-3156/#thread-interaction

这清楚地记录了您可能使用的新 asyncio 方法,包括 run_in_executor()。请注意,Executor 是在 concurrent.futures 中定义的,我建议你也看看那里。

于 2014-02-14T21:38:01.197 回答