时间戳

首页 > 解决方案 > 在 celery 中，当任务入队时，将上下文元数据从发送方进程传递给工作人员的适当方法是什么？

celery - 在 celery 中，当任务入队时，将上下文元数据从发送方进程传递给工作人员的适当方法是什么？

问题描述

当任何 celery 任务入队时，我想添加工作人员能够使用的上下文元数据。

以下代码示例有效，但我想要一个合适的芹菜式解决方案。

from celery.signals import before_task_publish, task_prerun

@before_task_publish.connect
def receiver_before_task_publish(sender=None, headers=None, body=None, **kwargs):
    task_kwags = body[1]
    metadata = {"foo": "bar"}
    task_kwags['__metadata__'] = metadata

@task_prerun.connect
def receiver_task_pre_run(task_id, task, *args, **kwargs):
    metadata = kwargs['kwargs'].pop('__metadata__', {})
    # metadata == {"foo": "bar"}

标签： celerydjango-celerycelery-task

解决方案

当任务在工作人员中启动时，before_task_publish's的内容header在**kwargsof 中push_request。

芹菜/app/tasks.py:1000

    def push_request(self, *args, **kwargs):
        self.request_stack.push(Context(*args, **kwargs))

的构造函数中做了一些不错的事情Context。 self.__dict__.update()意味着我们可以访问这些值Context(metadata={'foo': 'bar'}).metadata

芹菜/app/tasks.py:99

class Context(object)
# ...
    def __init__(self, *args, **kwargs):
        self.update(*args, **kwargs)

    def update(self, *args, **kwargs):
        return self.__dict__.update(*args, **kwargs)

任务上下文可从Task的request属性访问。

芹菜/app/tasks.py:1019

class Task(object):
# ...
    def _get_request(self):
        """Get current request object."""
        req = self.request_stack.top
        if req is None:
            # task was not called, but some may still expect a request
            # to be there, perhaps that should be deprecated.
            if self._default_request is None:
                self._default_request = Context()
            return self._default_request
        return req
    request = property(_get_request)

这意味着最终的解决方案很简单：

from celery.signals import before_task_publish, task_prerun

@before_task_publish.connect
def receiver_before_task_publish(sender=None, headers=None, body=None, **kwargs):
    metadata = {"foo": "bar"}
    headers['__metadata__'] = metadata

@task_prerun.connect
def receiver_task_pre_run(task_id, task, *args, **kwargs):
    metadata = getattr(task.request, '__metadata__', {}) 
    # metadata == {"foo": "bar"}

注意：task.request.__metadata__也可以，但如果在集成信号之前任务已入队，则会失败。这样更安全。

推荐阅读