使用 Celery 作为 Twisted 应用程序的控制通道

Question

我正在尝试使用 Celery 作为 Twisted 应用程序的控制通道。我的 Twisted 应用程序是一个抽象层，为各种本地运行的进程（通过 ProcessProtocol）提供标准接口。我想使用 Celery 来远程控制 - AMQP 似乎是从中央位置控制许多 Twisted 应用程序的理想方法，并且我想利用 Celery 的基于任务的功能，例如任务重试、子任务等。

这是没有按我计划的那样工作，我希望有人能帮助我指明正确的方向以使其正常工作。

当我运行脚本时我试图实现的行为是：

• 启动一个稍微修改过的 celeryd （请参阅 当
• 等待 Celery 任务
• 收到“启动进程”任务时，生成一个 ProcessProtocol
• 当收到其他任务时，在 Twisted 协议上运行一个函数并使用 Deferreds 返回结果

'稍微修改的 celeryd' 是 celeryd 进行了一些小修改，允许任务访问通过 self.app.twisted 扭曲反应器，通过 self.app.process 生成进程。为了简单起见，我使用 Celery 的“单独”进程池实现，它不会为任务工作人员分叉新进程。

当我尝试使用 Celery 任务来初始化 ProcessProtocol （即启动外部进程）时，就会出现问题。该进程正确启动，但 ProcessProtocol 的 childDataReceived 永远不会被调用。我认为这与文件描述符未正确继承/设置有关。

下面是一些示例代码，基于 ProcessProtocol 文档中的“wc”示例。它包括两个 Celery 任务 - 一个用于启动 wc 进程，另一个用于计算某些文本中的单词数（使用之前启动的 wc 进程）。

这个例子相当人为，但如果我能让这个工作正常，它将作为实现我的 ProcessProtocols 的一个很好的起点，它是长时间运行的进程，它将响应写入 stdin 的命令。

我首先通过运行 Celery 守护进程来测试这一点：

python2.6 mycelery.py -l info -P alone

然后，在另一个窗口中，运行发送两个任务的脚本：

python2.6 command_test.py

command_test.py 的预期行为用于执行两个命令 - 一个启动 wc 进程，另一个将一些文本发送到 CountWordsTask。实际发生的情况是：

• StartProcTask 生成进程，并通过 Defered 接收“进程启动”作为响应
• CountWordsTask 永远不会收到结果，因为 childDataReceived 永远不会被调用

任何人都可以阐明这一点，或者提供一些关于如何最好的建议使用 Celery 作为 Twisted ProcessProtocols 的控制通道？

为 Celery 编写一个 Twisted-backed ProcessPool 实现会更好吗？我通过reactor.callLater 调用WorkerCommand.execute_from_commandline 的方法是否是确保一切发生在Twisted 线程内的正确方法？

我读过有关 AMPoule 的文章，我认为它可以提供其中一些功能，但如果可能的话，我想坚持使用 Celery，因为我在应用程序的其他部分使用它。

任何帮助或协助将不胜感激！

myceleryd.py

from functools import partial
from celery.app import App
from celery.bin.celeryd import WorkerCommand
from twisted.internet import reactor


class MyCeleryApp(App):
    def __init__(self, twisted, *args, **kwargs):
        self.twisted = twisted
        super(MyCeleryApp, self).__init__(*args, **kwargs)

def main():
    get_my_app = partial(MyCeleryApp, reactor)
    worker = WorkerCommand(get_app=get_my_app)
    reactor.callLater(1, worker.execute_from_commandline)
    reactor.run()

if __name__ == '__main__':
    main()

协议.py

from twisted.internet import protocol
from twisted.internet.defer import Deferred

class WCProcessProtocol(protocol.ProcessProtocol):

    def __init__(self, text):
        self.text = text
        self._waiting = {} # Dict to contain deferreds, keyed by command name

    def connectionMade(self):
        if 'startup' in self._waiting:
            self._waiting['startup'].callback('process started')

    def outReceived(self, data):
        fieldLength = len(data) / 3
        lines = int(data[:fieldLength])
        words = int(data[fieldLength:fieldLength*2])
        chars = int(data[fieldLength*2:])
        self.transport.loseConnection()
        self.receiveCounts(lines, words, chars)

        if 'countWords' in self._waiting:
            self._waiting['countWords'].callback(words)

    def processExited(self, status):
        print 'exiting'


    def receiveCounts(self, lines, words, chars):
        print >> sys.stderr, 'Received counts from wc.'
        print >> sys.stderr, 'Lines:', lines
        print >> sys.stderr, 'Words:', words
        print >> sys.stderr, 'Characters:', chars

    def countWords(self, text):
        self._waiting['countWords'] = Deferred()
        self.transport.write(text)
        return self._waiting['countWords']

任务.py

from celery.task import Task
from protocol import WCProcessProtocol
from twisted.internet.defer import Deferred
from twisted.internet import reactor

class StartProcTask(Task):
    def run(self):
        self.app.proc = WCProcessProtocol('testing')
        self.app.proc._waiting['startup'] = Deferred()
        self.app.twisted.spawnProcess(self.app.proc,
                                      'wc',
                                      ['wc'],
                                      usePTY=True)
        return self.app.proc._waiting['startup']

class CountWordsTask(Task):
    def run(self):
        return self.app.proc.countWords('test test')

Answer 1

Celery 在等待来自网络的新消息时可能会阻塞。由于您将其与 Twisted 反应器一起在一个单线程进程中运行，因此它会阻止反应器运行。这将禁用大部分 Twisted，这需要反应器实际运行（您调用了reactor.run，但由于 Celery 阻止了它，所以它实际上没有运行）。

reactor.callLater 仅延迟 Celery 的启动。一旦 Celery 启动，它仍然会阻塞反应器。

您需要避免的问题是阻塞反应堆。

一种解决方案是在一个线程中运行 Celery，在另一个线程中运行 Reactor。使用reactor.callFromThread从Celery线程向Twisted发送消息（“在反应器线程中调用函数”）。如果您需要从 Twisted 线程将消息发送回 Celery，请使用 Celery 等效项。

另一种解决方案是将 Celery 协议（AMQP？ - 请参阅 txAMQP）作为本机 Twisted 库实现，并使用它来不阻塞地处理 Celery 消息。