龙卷风(或异步)中是否有行为/模式等待a 中的任何而不是所有期货list
?
yield any_of([future1, future2, future3])
说future2
准备好了,那么结果应该是:
[None, <result>, None]
龙卷风(或异步)中是否有行为/模式等待a 中的任何而不是所有期货list
?
yield any_of([future1, future2, future3])
说future2
准备好了,那么结果应该是:
[None, <result>, None]
更新: Tornado 现在有tornado.gen.WaitIterator,根据其文档中的示例使用它,而不是我下面的想法。
您可以创建一个继承自 Future 的 Any 类,并包装一个期货列表。Any 类等到它的一个期货解决,然后给你结果列表:
import time
from tornado import gen
from tornado.ioloop import IOLoop
from tornado.concurrent import Future
@gen.coroutine
def delayed_msg(seconds, msg):
yield gen.Task(IOLoop.current().add_timeout,
time.time() + seconds)
raise gen.Return(msg)
class Any(Future):
def __init__(self, futures):
super(Any, self).__init__()
self.futures = futures
for future in futures:
future.add_done_callback(self.done_callback)
def done_callback(self, future):
try:
self.set_result(self.make_result())
except Exception as e:
self.set_exception(e)
def make_result(self):
"""A list of results: None for each pending future, a result for
each resolved future. Raises an exception for the first future
that has an exception.
"""
return [f.result() if f.done() else None
for f in self.futures]
def clear(self):
"""Break reference cycle with any pending futures."""
self.futures = None
@gen.coroutine
def f():
start = time.time()
future1 = delayed_msg(2, '2')
future2 = delayed_msg(3, '3')
future3 = delayed_msg(1, '1')
results = yield Any([future1, future2, future3])
end = time.time()
print "finished in %.1f sec: %r" % (end - start, results)
results = yield Any([future1, future2])
end = time.time()
print "finished in %.1f sec: %r" % (end - start, results)
IOLoop.current().run_sync(f)
正如预期的那样,这将打印:
finished in 1.0 sec: [None, None, '1']
finished in 2.0 sec: ['2', None]
但是您可以看到这种方法存在一些复杂性。一方面,如果您想在第一个期货解决后等待其余期货,那么构建仍然未决期货的列表很复杂。我想你可以这样做:
results = yield Any(f for f in [future1, future2, future3] if not f.done())
不漂亮,甚至不正确!有一个比赛条件。如果未来在的连续执行之间yield Any(...)
得到解决,那么您将永远不会收到它的结果。第一个yield
没有得到未来的结果,因为它仍然处于未决状态,但第二个yield
也没有得到它的结果,因为到那时未来已经“完成”并且它不包含在传递给的列表中Any
。
另一个复杂之处是 Any 指的是每个未来,它指的是一个回调,该回调指回 Any。为了及时进行垃圾收集,您应该调用 Any.clear()。
此外,您无法区分未决的未来和解决为无的未来。您需要一个不同于 None 的特殊哨兵值来表示未决的未来。
最后的并发症是最严重的。如果解决了多个期货并且其中一些有例外,那么 Any 没有明显的方法可以将所有这些信息传达给您。将异常和结果混合在一个列表中是不恰当的。
我认为有一个更简单的方法。我们可以让 Any 返回第一个解析的未来,而不是结果列表:
class Any(Future):
def __init__(self, futures):
super(Any, self).__init__()
for future in futures:
future.add_done_callback(self.done_callback)
def done_callback(self, future):
self.set_result(future)
引用循环消失了,异常处理问题得到了解答:Any 类将整个未来返回给您,而不是其结果或异常。您可以随意检查它。在一些解决之后等待剩余的期货也很容易:
@gen.coroutine
def f():
start = time.time()
future1 = delayed_msg(2, '2')
future2 = delayed_msg(3, '3')
future3 = delayed_msg(1, '1')
futures = set([future1, future2, future3])
while futures:
resolved = yield Any(futures)
end = time.time()
print "finished in %.1f sec: %r" % (end - start, resolved.result())
futures.remove(resolved)
根据需要,这将打印:
finished in 1.0 sec: '1'
finished in 2.0 sec: '2'
finished in 3.0 sec: '3'
我们可以通过添加一个新的全局函数来测试异常处理行为:
@gen.coroutine
def delayed_exc(seconds, msg):
yield gen.Task(IOLoop.current().add_timeout,
time.time() + seconds)
raise Exception(msg)
并产生它而不是delayed_msg:
@gen.coroutine
def f():
start = time.time()
future1 = delayed_msg(2, '2')
future2 = delayed_exc(3, '3') # Exception!
future3 = delayed_msg(1, '1')
futures = set([future1, future2, future3])
while futures:
resolved = yield Any(futures)
end = time.time()
try:
outcome = resolved.result()
except Exception as e:
outcome = e
print "finished in %.1f sec: %r" % (end - start, outcome)
futures.remove(resolved)
现在,脚本将打印“1”,然后是“2”,然后是“Exception('3',)”。