Python Twisted运行多个回调

Question

我对twisted还是个新手，我真的需要一件事——运行任意数量的函数（从同一个tame开始），收集所有函数的结果并进行一些处理。在

以下是我所拥有的：

from twisted.internet import defer
import time


# slow computing query
def process_data(num, data):
    time.sleep(5)
    array = []
    # mock the results obtained from processed data
    for i in range(0, 5):
        array.append(num)
    return array

def process_results(arrays):
    # this should collect return arrays of all callbacks 
    print arrays

data = []
callbacks_refs = []
for i in range(0, 5):
   d=defer.Deferred()
   d.addCallback(process_data)
   callbacks_refs.append(d)

callbacks = defer.DeferredList(callbacks_refs)
callbacks.addCallback(process_results)

for i, d in enumerate(callbacks_refs):
    d.callback(i, data)

我希望最后一个for循环将异步开始所有回调的执行（就像通常的Promises），并且所有结果都将传递给process_results函数，该函数将在所有来自回调的回调完成后执行，但我觉得我错得很厉害。在

Answer 1

我不知道您的示例与实际代码有多相似，但是示例代码显示了一些关于Twisted的错误理解。Twisted不会神奇地使同步代码异步。您正在阻止time.sleep中的事件循环。如果您正在执行CPU限制的操作（与I/O绑定相反），则可以使用多个线程或进程。在

我假设process_data是一个阻塞调用，并为您提供一个基于多线程的解决方案：

import time
from twisted.internet import defer, task, threads

# slow computing query
def process_data(num):
    time.sleep(5)
    array = []
    # mock the results obtained from processed data
    for i in range(0, 5):
        array.append(num)
    return array

def process_results(arrays):
    # this should collect return arrays of all callbacks
    print arrays

def main(_):
    callbacks_refs = []
    for i in range(0, 5):
        callbacks_refs.append(threads.deferToThread(process_data, i))
    callbacks = defer.DeferredList(callbacks_refs)
    callbacks.addCallback(process_results)
    return callbacks

task.react(main)

我还将给你一个关于Twisted编程的一般性建议——如果你发现自己在输入d = defer.Deferred()，那么你的设计可能有问题。在

Answer 2

我不知道这是否有解决方法，但是使用defer.callback()的方式，您将获得传递到回调的错误参数。在

如果你在回调的同时附加一个errback，你可能会发现你得到的只是一堆失败的结果。。。所以它起作用了，但没有如预期的那样工作。在

我看到了两个修正。在

from functools import partial
for i in range(0, 5):
    d=defer.Deferred()
    d.addCallback(partial(process_data,i,data[i]))
    # This partial is still kinda crooked, but hopefully I have made my point
    callbacks_refs.append(d)

或者更改回调函数中传递数据的方式

^{pr2}$

很抱歉，我对延迟列表不太熟悉，但是我想一旦你解决了延迟，延迟列表可能会自动工作。在