python之celery 从 task_success 处理程序关闭工作人员不工作

我试图让一个工作人员一次只运行一个任务，然后关闭。我已经让关闭部分正常工作(这里有一些背景:celery trying shutdown worker by raising SystemExit in task_postrun signal but always hangs and the main process never exits)，但是当它关​​闭时，我收到一个错误:

[2013-02-13 12:19:05,689: CRITICAL/MainProcess] Couldn't ack 1, reason:AttributeError("'NoneType' object has no attribute 'method_writer'",) 
Traceback (most recent call last): 
  File "/usr/local/lib/python2.7/site-packages/kombu/transport/base.py", line 104, in ack_log_error 
    self.ack() 
  File "/usr/local/lib/python2.7/site-packages/kombu/transport/base.py", line 99, in ack 
    self.channel.basic_ack(self.delivery_tag) 
  File "/usr/local/lib/python2.7/site-packages/amqplib/client_0_8/channel.py", line 1742, in basic_ack 
    self._send_method((60, 80), args) 
  File "/usr/local/lib/python2.7/site-packages/amqplib/client_0_8/abstract_channel.py", line 75, in _send_method 
    self.connection.method_writer.write_method(self.channel_id, 
AttributeError: 'NoneType' object has no attribute 'method_writer' 

为什么会这样？它不仅不确认，而且还会清除队列中剩余的所有其他任务(大问题)。

我该如何解决这个问题？

更新

下面是所有更新的堆栈跟踪(pip install -U kombu amqp amqplib celery):

[2013-02-13 11:58:05,357: CRITICAL/MainProcess] Internal error: AttributeError("'NoneType' object has no attribute 'method_writer'",) 
Traceback (most recent call last): 
  File "/usr/local/lib/python2.7/dist-packages/celery/worker/__init__.py", line 372, in process_task 
    req.execute_using_pool(self.pool) 
  File "/usr/local/lib/python2.7/dist-packages/celery/worker/job.py", line 219, in execute_using_pool 
    timeout=task.time_limit) 
  File "/usr/local/lib/python2.7/dist-packages/celery/concurrency/base.py", line 137, in apply_async 
    **options) 
  File "/usr/local/lib/python2.7/dist-packages/celery/concurrency/base.py", line 27, in apply_target 
    callback(target(*args, **kwargs)) 
  File "/usr/local/lib/python2.7/dist-packages/celery/worker/job.py", line 333, in on_success 
    self.acknowledge() 
  File "/usr/local/lib/python2.7/dist-packages/celery/worker/job.py", line 439, in acknowledge 
    self.on_ack(logger, self.connection_errors) 
  File "/usr/local/lib/python2.7/dist-packages/kombu/transport/base.py", line 98, in ack_log_error 
    self.ack() 
  File "/usr/local/lib/python2.7/dist-packages/kombu/transport/base.py", line 93, in ack 
    self.channel.basic_ack(self.delivery_tag) 
  File "/usr/local/lib/python2.7/dist-packages/amqp/channel.py", line 1562, in basic_ack 
    self._send_method((60, 80), args) 
  File "/usr/local/lib/python2.7/dist-packages/amqp/abstract_channel.py", line 57, in _send_method 
    self.connection.method_writer.write_method( 
AttributeError: 'NoneType' object has no attribute 'method_writer' 

请您参考如下方法:

不建议在 task_postrun 中退出，因为 task_postrun 是在“任务主体”错误处理之外执行的。

当任务调用 sys.exit 时究竟发生了什么还没有明确定义， 实际上这取决于所使用的池。

使用多处理，子进程将被一个新进程简单地替换。 在其他池中，工作人员将关闭，但这可能会改变 以便它与多处理行为一致。

在任务体外部调用 exit 被视为内部错误(崩溃)。

“任务主体”是在 task.__call__()

我认为也许更好的解决方案是使用自定义执行 策略:

from celery.worker import strategy 
from functools import wraps 
 
@staticmethod 
def shutdown_after_strategy(task, app, consumer): 
 
    default_handler = strategy.default(task, app, consumer) 
 
    def _shutdown_to_exit_after(fun): 
        @wraps(fun) 
        def _inner(*args, **kwargs): 
            try: 
                return fun(*args, **kwargs) 
            finally: 
                raise SystemExit() 
       return _inner 
    return _decorate_to_exit_after(default_handler) 
 
@celery.task(Strategy=shutdown_after_strategy) 
def shutdown_after(): 
    print('will shutdown after this') 

这不是很漂亮，但是执行策略可以优化 任务执行并且不容易扩展(工作人员“预编译”执行 通过缓存 Task.Strategy 获取每种任务类型的路径)

在 Celery 3.1 中，您可以使用“bootsteps”扩展 worker 和 consumer，所以很可能 届时会有一个漂亮的解决方案。