旋風IOLoop回調異常在芹菜工人無

Question

我使用tornado.ioloopcelery工人，因爲我需要使用mongodb。

class WorkerBase(): 
    @gen.engine 
    def foo(self,args,callback) 
     bar = ['Python','Celery','Javascript','HTML'] 

     # ... process something .... 

     callback(bar) 

    @gen.engine 
    def RunMyTask(self,args): 

     result = yield gen.Task(self.foo,args=args) 
     # Stop IOLoop instance 
     IOLoop.instance().stop() 


@task(name="MyWorker",base=WorkerBase) 
def CeleryWorker(args): 
    # This works because i'm adding base as WorkerBase 
    CeleryWorker.RunMyTask(args) 
    IOLoop.instance().start() 
    return True 

當我通過調用任務它提供了一個錯誤說：

[2014-10-02 12:12:11,561: ERROR/Worker-4] Exception in callback None 
Traceback (most recent call last): 
    File "/var/www/myapp/env/local/lib/python2.7/site-packages/tornado/ioloop.py", line 832, in start 
fd_obj, handler_func = self._handlers[fd] 
KeyError: 16 

或

[2014-10-02 12:12:11,561: ERROR/Worker-4] Exception in callback None 
Traceback (most recent call last): 
    File "/var/www/myapp/env/local/lib/python2.7/site-packages/tornado/ioloop.py", line 832, in start 
fd_obj, handler_func = self._handlers[fd] 
KeyError: 14 

這些錯誤並不一致。有沒有任何提升條件？

Answer 1

這看起來像一個線程問題。我對芹菜的線程模型並不熟悉，但它看起來像啓動了CeleryWorker的多個副本，每個副本都試圖運行同一個單例IOLoop.instance（）。每個工作線程都需要自己的IOLoop，如果你要這樣運行 - 看看同步的tornado.httpclient.HTTPClient如何創建並運行臨時IOLoop

Answer 2

看起來你的工作任務只是返回並被視爲在ioloop之前完成停止，所以gen.engine的回調無法找到原來的stack_context我猜。

@task(name="MyWorker",base=WorkerBase) 
def CeleryWorker(args): 
    # This works because i'm adding base as WorkerBase 
    CeleryWorker.RunMyTask(args) 
    IOLoop.instance().start() 
    return True 

我有一些建議給你

1）除去回

@task(name="MyWorker",base=WorkerBase) 
def CeleryWorker(args): 
    # This works because i'm adding base as WorkerBase 
    CeleryWorker.RunMyTask(args) 
    IOLoop.instance().start() 

2）使用run_sync

import functools 

@task(name="MyWorker",base=WorkerBase) 
def CeleryWorker(args): 
    # This works because i'm adding base as WorkerBase 
    func = functools.partial(CeleryWorker.RunMyTask, args) 
    IOLoop.instance().run_sync(func)