Могу ли я применить пользовательскую маршрутизацию к TaskSet, используя apply_async?

Question

Согласно документации, найденной в http://ask.github.com/celery/userguide/routing.html#manual-routing, я могу передать параметр queue в apply_async, чтобы направить задачу в определенную очередь.Однако при использовании TaskSet я получаю

TypeError at /some/path/
apply_async() got an unexpected keyword argument 'queue'

, что невозможно понять, учитывая следующий код в классе TaskSet https://github.com/ask/celery/blob/master/celery/task/sets.py#L122

def apply_async(self, connection=None, connect_timeout=None,
        publisher=None, taskset_id=None):
    """Apply taskset."""
    app = self.app

    if app.conf.CELERY_ALWAYS_EAGER:
        return self.apply(taskset_id=taskset_id)

    with app.default_connection(connection, connect_timeout) as conn:
        setid = taskset_id or uuid()
        pub = publisher or self.Publisher(connection=conn)
        try:
            results = self._async_results(setid, pub)
        finally:
            if not publisher:  # created by us.
                pub.close()

        return app.TaskSetResult(setid, results)

У меня есть неопределенное количество задач, для которых мне нужноприменить специальную маршрутизацию в некоторых ситуациях, как мне справиться с этим?не использовать TaskSet?

Answer 1

Вы можете использовать подзадачи с параметрами аргумент

>>> from celery.task.sets import TaskSet
>>> from tasks import add     
>>> 
>>> job = TaskSet(tasks=[add.subtask(args=(i, i),options={'queue':'celery'}) for i in range(10)])
>>> result = job.apply_async()
>>> result.ready()
True
>>> job
[tasks.add(0, 0), tasks.add(1, 1), tasks.add(2, 2), tasks.add(3, 3), tasks.add(4, 4), tasks.add(5, 5), tasks.add(6, 6), tasks.add(7, 7), tasks.add(8, 8), tasks.add(9, 9)]
>>>