0

我在使用 Tornado 做一些工作时使用线程池。这是代码：

常见/thread_pool.py

import tornado.ioloop

class Worker(threading.Thread):
    def __init__(self, queue):
        threading.Thread.__init__(self)
        self._queue = queue

    def run(self):
        logging.info('Worker start')
        while True:
            content = self._queue.get()
            if isinstance(content, str) and content == 'quit':
                break
            #content: (func, args, on_complete)
            func = content[0]
            args = content[1]
            on_complete = content[2]
            resp = func(args)
            tornado.ioloop.IOLoop.instance().add_callback(lambda: on_complete(resp))
            #i dont know is correct to call this
            #self._queue.task_done()
        logging.info('Worker stop')

class WorkerPool(object):
    _workers = []
    def __init__(self, num):
        self._queue = Queue.Queue()
        self._size = num

    def start(self):
        logging.info('WorkerPool start %d' % self._size)
        for _ in range(self._size):
            worker = Worker(self._queue)
            worker.start()
            self._workers.append(worker)

    def stop(self):
        for worker in self._workers:
            self._queue.put('quit') 
        for worker in self._workers:
            worker.join()
        logging.info('WorkerPool stopd')

    def append(self, content):
        self._queue.put(content)

网关.py

import tornado.ioloop
import tornado.web

from common import thread_pool

workers = None

class MainServerHandler(tornado.web.RequestHandler):
    @tornado.web.asynchronous
    def get(self):
        start_time = time.time()
        method = 'get'
        content = (self.handle, (method, self.request, start_time), self.on_complete)
        workers.append(content)

    @tornado.web.asynchronous
    def post(self):
        start_time = time.time()
        method = 'post'
        content = (self.handle, (method, self.request, start_time), self.on_complete)
        workers.append(content)

    def handle(self, args):
        method, request, start_time = args
        #for test, just return
        return 'test test'

    def on_complete(self, res):
        logging.debug('on_complete')
        self.write(res)
        self.finish()
        return        

def main(argv):  
    global workers
    workers = thread_pool.WorkerPool(conf_mgr.thread_num)
    workers.start()

    application = tornado.web.Application([(r"/", MainServerHandler)])
    application.listen(8888)
    tornado.ioloop.IOLoop.instance().start()

if __name__ == "__main__":
    main(sys.argv[1:])

当我发出许多并发请求时，我收到此错误：

ERROR: 2014-09-15 18:04:03: ioloop.py:435 * 140500107065056 Exception in callback <tornado.stack_context._StackContextWrapper object at 0x7fc8b4d6b9f0>

  Traceback (most recent call last):
     File "/home/work/nlp_arch/project/ps/se/nlp-arch/gateway/gateway/../third-party/tornado-2.4.1/tornado/ioloop.py", line 421, in _run_callback
       callback()
     File "/home/work/nlp_arch/project/ps/se/nlp-arch/gateway/gateway/../common/thread_pool.py", line 39, in <lambda>
       tornado.ioloop.IOLoop.instance().add_callback(lambda: on_complete(resp))
     File "/home/work/nlp_arch/project/ps/se/nlp-arch/gateway/gateway/gateway.py", line 92, in on_complete
       self.write(res)
     File "/home/work/nlp_arch/project/ps/se/nlp-arch/gateway/gateway/../third-party/tornado-2.4.1/tornado/web.py", line 489, in write
      raise RuntimeError("Cannot write() after finish().  May be caused "
  RuntimeError: Cannot write() after finish().  May be caused by using async operations without the @asynchronous decorator.

write但后来我没有打电话finish。我也在使用@asynchronous装饰器。同时，在日志中我看到write/finish被同一个线程调用。

4

1 回答 1

1

问题在于您将回调添加到 I/O 循环的方式。像这样添加它：

tornado.ioloop.IOLoop.instance().add_callback(on_complete, resp)

并且错误会消失。

您会看到这种奇怪的行为，因为当您使用 lambda 函数时，您正在函数的本地范围内创建一个闭包，并且该闭包中使用的变量在lambda 执行时被绑定，而不是在创建时绑定. 考虑这个例子：

funcs = []
def func(a):
    print a

for i in range(5):
   funcs.append(lambda: func(i))

for f in funcs:
    f()

输出：

4
4
4
4
4

因为您的工作方法在 while 循环中运行，on_complete最终会被重新定义多次，这也会改变on_completelambda 内部的值。这意味着如果一个工作线程on_complete为处理程序 A 设置，但随后在为处理程序 A 运行的回调设置之前获取另一个任务并on_complete为处理程序 B 设置，则两个回调最终都会运行处理程序on_completeB。

如果你真的想使用 lambda，你也可以通过on_complete在 lambda 的本地范围内绑定来避免这种情况：

tornado.ioloop.IOLoop.instance().add_callback(lambda on_complete=on_complete: on_complete(resp))

但是直接添加函数及其参数会更好。

于 2014-09-15T16:19:19.867 回答