Enrutamiento y creación de colas dinámicas de apio

Question

Estoy tratando de llamar a una tarea y crear una cola para esa tarea si no existe, luego inserte inmediatamente en esa cola la tarea llamada. Tengo el siguiente código:

@task 
def greet(name): 
    return "Hello %s!" % name 


def run(): 
    result = greet.delay(args=['marc'], queue='greet.1', 
     routing_key='greet.1') 
    print result.ready() 

entonces tengo un router personalizado:

class MyRouter(object): 

    def route_for_task(self, task, args=None, kwargs=None): 
     if task == 'tasks.greet': 
      return {'queue': kwargs['queue'], 
        'exchange': 'greet', 
        'exchange_type': 'direct', 
        'routing_key': kwargs['routing_key']} 
     return None 

esto crea un intercambio llamado greet.1 y una cola llamada greet.1 pero la cola está vacía. El intercambio debe llamarse greet que sabe cómo enrutar una clave de enrutamiento como greet.1 a la cola llamada greet.1.

¿Alguna idea?

Answer 1

Al hacer lo siguiente:

task.apply_async(queue='foo', routing_key='foobar') 

Entonces apio se llevará a valores por defecto de la cola 'foo' en CELERY_QUEUES, o si no existe, entonces creará automáticamente usando (cola = foo, el intercambio = foo, routing_key = foo)

Así que si 'foo' no existe en CELERY_QUEUES el resultado final será con:

queues['foo'] = Queue('foo', exchange=Exchange('foo'), routing_key='foo') 

entonces, el productor declare que cola, pero ya se ha sobrepasado la routing_key, realidad enviar el mensaje utilizando routing_key = 'foobar'

Esto puede parecer extraño, pero el comportamiento es realmente útil para el intercambio de temas, donde se publica en diferentes temas.

Aunque es más difícil hacer lo que quiere, puede crear la cola usted mismo y declararla, pero eso no funcionará bien con los reintentos de publicación automática de mensajes. Sería mejor si el argumento de cola para apply_async pudiera admitir un kombu.Queue personalizado que se declararía y usaría como destino. Quizás podría abrir un problema al http://github.com/celery/celery/issues