Exécuter le rappel link_error de Celery sur une file d'attente/ouvrier séparé

Question

J'ai 2 applications sur 2 serveurs distincts, appelons-les A et B. Les deux applications ont un ouvrier Céleri actif, écoutant des files d'attente séparées (QueueA et QueueB). Le serveur B envoie une tâche à QueueB en utilisant apply_async.

Voici les tâches du serveur B:

@app.task(bind=True, queue="QueueB", name="name_on_server_A") 
def taskForServerB(): 
    # nothing is executed here 

@app.task(bind=True) 
def success(result): 
    print('Task succeeded') 

@app.task(bind=True): 
def failure(...): 
    print('task failed') 


taskForServerB.s().apply_async(link=success.s(), link_error=failure.s()) 

Sur le serveur A, la tâche name_on_server_A reçoit les tâches et l'exécute. Si elle se termine avec succès, la tâche success est exécutée correctement sur ServerB, mais name_on_server_A échoue, la tâche failure n'est pas exécutée. Au lieu de cela, le serveur A lance un NotRegisteredError pour une tâche portant le nom failure.

Y a-t-il quelque chose qui me manque? Comment puis-je exécuter la tâche d'échec sur ServerB, d'où la première tâche est appelée?

Answer 1

Il y a deux questions:

1. La route de tâche à la bonne file d'attente que vous avez définie pour name_on_server_A (avec l'affectation queue) - qui est par la façon dont quelque chose qui est nouveau pour moi (J'utilise ROUTER dans la config de céleri et la route chaque tâche par son nom à la bonne file d'attente

2. lorsque vous définissez votre application de céleri vous pourriez oublié d'inclure la tâche failure il désenregistrer.

   app = Céleri (courtier = 'AMQP: //', back-end = '... ' include = [' file1.py', 'file2.py', ..])