Celery/RabbitMQ messaggi non bloccati che bloccano la coda?

Question

Ho richiamato un'attività che recupera alcune informazioni in remoto con urllib2 alcune migliaia di volte. Le attività sono pianificate con un eta casuale (entro una settimana) in modo che non colpiscano il server contemporaneamente. A volte ottengo un 404, a volte no. Sto gestendo l'errore nel caso in cui ciò accada.

Nella console RabbitMQ posso vedere 16 messaggi non riconosciuti:

ho smesso di sedano, epurato la coda e riavviato esso. I 16 messaggi non riconosciuti erano ancora lì.

Ho altri compiti che vanno nella stessa coda e nessuno di essi è stato eseguito. Dopo lo spurgo, ho cercato di inviare un altro compito ed è stato rimane pronto:

Tutte le idee come posso scoprire perché messaggi rimangono non riconosciuti?

Versioni:

celery==3.1.4 
{rabbit,"RabbitMQ","3.5.3"} 

celeryapp.py

CELERYBEAT_SCHEDULE = { 
    'social_grabber': { 
     'task': '<django app>.tasks.task_social_grabber', 
     'schedule': crontab(hour=5, minute=0, day_of_week='sunday'), 
    }, 
} 

tasks.py

@app.task 
def task_social_grabber(): 
    for user in users: 
     eta = randint(0, 60 * 60 * 24 * 7) #week in seconds 
     task_social_grabber_single.apply_async((user), countdown=eta) 

Non esiste alcun instradamento per questa attività definita, quindi entra nella coda predefinita: sedano. C'è un lavoratore che elabora questa coda.

supervisord.conf:

[program:celery] 
autostart = true 
autorestart = true 
command = celery worker -A <django app>.celeryapp:app --concurrency=3 -l INFO -n celery 

Answer 1

RabbitMQ rotto impostazioni di QoS nella versione 3.3. È necessario aggiornare il sedano ad almeno 3.1.11 (changelog) e kombu ad almeno 3.0.15 (changelog). Dovresti usare le ultime versioni.

Ho raggiunto lo stesso identico comportamento quando è stato rilasciato 3.3. RabbitMQ ha capovolto il comportamento predefinito del flag prefetch_count. Prima di questo, se un consumatore ha raggiunto il limite CELERYD_PREFETCH_MULTIPLIER nei messaggi eta, il lavoratore aumenterebbe questo limite per recuperare più messaggi. Il cambiamento ha rotto questo comportamento, in quanto il nuovo comportamento predefinito ha negato questa funzionalità.

Answer 2

Ho avuto sintomi simili. Messaggi in cui raggiungere il MQ (visibile nei grafici) ma dove non è stato rilevato dal lavoratore.

questo mi ha portato alla supposizione che la mia app Django aveva correttamente l'installazione di sedano app, ma mi mancava an import ensuring Celery would be configured during Django startup:

from __future__ import absolute_import 

# This will make sure the app is always imported when 
# Django starts so that shared_task will use this app. 
from .celery import app as celery_app # noqa 

E 'un errore stupido, ma i messaggi che ottengono al broker, dopo aver restituito un AsyncResult, mi ha messo fuori strada, e mi ha fatto guardare nei posti sbagliati. Poi ho notato che l'impostazione CELERY_ALWAYS_EAGER = True non eseguiva lo squat, quindi l'attività non veniva eseguita affatto.

PS: Questa potrebbe non essere una risposta alla domanda @kev, ma poiché sono arrivato qui un paio di volte, mentre cercavo la soluzione al mio problema, lo postò qui per chiunque si trovi in ​​una situazione simile.