Django Celery ottiene il conteggio delle attività

Question

Attualmente sto usando Django con il sedano e tutto funziona correttamente.

Tuttavia, voglio essere in grado di offrire agli utenti l'opportunità di annullare un'attività se il server è sovraccarico controllando il numero di attività attualmente pianificate.

Come posso ottenere questo risultato?

Sto usando redis come broker.

Ho appena trovato questo: Retrieve list of tasks in a queue in Celery

E 'in qualche modo si riferiscono al mio problema, ma non ho bisogno di elencare i compiti, basta :) contare

Answer 1

Se il vostro broker è configurato come redis://localhost:6379/1, e le attività sono sottoposti alla coda generale celery, allora si può ottenere la lunghezza attraverso i seguenti mezzi:

import redis 
queue_name = "celery" 
client = redis.Redis(host="localhost", port=6379, db=1) 
length = client.llen(queue_name) 

Oppure, da uno script di shell (buona per i monitor e così via):

$ redis-cli -n 1 -h localhost -p 6379 llen celery 

Answer 2

Se è già stato configurato Redis nella vostra applicazione, si può provare questo:

from celery import Celery 

QUEUE_NAME = 'celery' 

celery = Celery(app) 
client = celery.connection().channel().client 

length = client.llen(QUEUE_NAME) 

Answer 3

Ecco come è possibile ottenere il numero di messaggi in una coda con sedano che è Broker- agnostico.

Utilizzando connection_or_acquire, è possibile ridurre al minimo il numero di connessioni aperte al proprio broker utilizzando il pool di connessioni interne di sedano.

celery = Celery(app) 

with celery.connection_or_acquire() as conn: 
    conn.default_channel.queue_declare(
     queue='my-queue', passive=True).message_count 

è anche possibile estendere sedano per fornire questa funzionalità:

from celery import Celery as _Celery 


class Celery(_Celery) 

    def get_message_count(self, queue): 
     ''' 
     Raises: amqp.exceptions.NotFound: if queue does not exist 
     ''' 
     with self.connection_or_acquire() as conn: 
      return conn.default_channel.queue_declare(
       queue=queue, passive=True).message_count 


celery = Celery(app) 
num_messages = celery.get_message_count('my-queue')