2

我试图让一名工作人员一次只运行一项任务，然后关闭。我已经让关闭部分正常工作（这里的一些背景：celery 通过在 task_postrun 信号中提高 SystemExit 来尝试关闭工作者，但总是挂起并且主进程从不退出），但是当它关​​闭时，我收到一个错误：

[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'
4

1 回答 1

0

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

当任务调用 sys.exit 时发生的确切情况没有明确定义，实际上它取决于正在使用的池。

使用多处理，子进程将简单地被一个新进程替换。在其他池中，worker 将关闭，但这可能会发生变化，以便与多处理行为一致。

在任务主体之外调用 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 中，您可以使用“引导步骤”扩展工作人员和消费者，因此很可能会有一个很好的解决方案。

于 2013-02-14T12:57:36.763 回答