Python - 如何使这段代码异步？

Question

这是一些说明我的问题的代码:

def blocking1():
    while True:
        yield 'first blocking function example'

def blocking2():
    while True:
        yield 'second blocking function example'

for i in blocking1():
    print 'this will be shown'

for i in blocking2():
    print 'this will not be shown'

我有两个包含while True循环的函数.这些将生成数据,然后我会在某处(最有可能是sqlite数据库)登录.

我一直在玩线程并让它运行起来.但是,我真的不喜欢它...我想做的是使我的阻塞函数异步.就像是:

def blocking1(callback):
    while True:
        callback('first blocking function example')

def blocking2(callback):
    while True:
        callback('second blocking function example')

def log(data):
    print data

blocking1(log)
blocking2(log)

我怎样才能在Python中实现这一目标？我已经看到标准库带有asyncore,这个游戏中的大牌是Twisted,但这两个似乎都用于套接字IO.

如何异步我的非套接字相关的阻塞函数？

Answer 1

阻塞功能是一种不返回的功能,但仍会使您的进程处于空闲状态 - 无法完成更多工作.

您要求我们使您的阻止功能无阻塞.但是-除非你写一个操作系统- 你没有任何阻挡功能.您可能具有阻塞函数,因为它们调用阻塞系统调用,或者您可能具有"阻塞"的函数,因为它们执行大量计算.

如果不使基础系统调用非阻塞,则使前一类函数无阻塞是不可能的.根据系统调用的内容,如果不向程序添加事件循环,可能很难使其无阻塞; 您不仅需要拨打电话并且不要阻止,您还必须再次拨打电话以确定该呼叫的结果将在您可以关联的地方发送.

这个问题的答案是一个非常长的python程序和许多不同操作系统界面的解释以及它们如何工作,但幸运的是我已经在不同的网站上写了这个答案; 我叫它扭曲.如果Twisted反应堆已经支持您的特定任务,那么您很幸运.否则,只要您的任务映射到某些现有操作系统概念,您就可以扩展反应器以支持它.实际上,这些机制中只有两种:每个合理操作系统上的文件描述符,以及Windows上的I/O完成端口.

在另一种情况下,如果你的功能消耗了大量的CPU,因此没有返回,它们并没有真正阻塞; 你的过程仍在继续,并完成工作.有三种方法可以解决这个问题:

• 单独的线程
• 单独的过程
• 如果你有一个事件循环,编写一个定期产生的任务,通过以一种工作方式编写任务,然后要求事件循环在不久的将来恢复它,以便允许其他任务运行.

在Twisted中,最后一种技术可以通过各种方式实现,但这里有一种语法上方便的技巧,可以轻松实现:

from twisted.internet import reactor
from twisted.internet.task import deferLater
from twisted.internet.defer import inlineCallbacks, returnValue

@inlineCallbacks
def slowButSteady():
    result = SomeResult()
    for something in somethingElse:
        result.workHardForAMoment(something)
        yield deferLater(reactor, 0, lambda : None)
    returnValue(result)

Answer 2

您可以使用生成器进行协作式多任务处理,但您必须编写自己的主循环,以便在它们之间传递控制权.

以下是使用上面示例的(非常简单)示例:

def blocking1():
    while True:
        yield 'first blocking function example'

def blocking2():
    while True:
        yield 'second blocking function example'


tasks = [blocking1(), blocking2()]

# Repeat until all tasks have stopped
while tasks:
    # Iterate through all current tasks. Use
    # tasks[:] to copy the list because we
    # might mutate it.
    for t in tasks[:]:
        try:
            print t.next()
        except StopIteration:
            # If the generator stops, remove it from the task list
            tasks.remove(t)

您可以通过允许生成器生成新的生成器来进一步改进它,然后可以将其添加到任务中,但希望这个简化的示例将给出一般的想法.

Answer 3

如果您不想使用完整的操作系统线程，您可以尝试Stackless，它是 Python 的一个变体，添加了许多有趣的功能，包括“微线程”。有许多很好的例子会对您有所帮助。