Django Sellerie erhalten Aufgaben zählen

Question

Ich verwende derzeit Django mit Sellerie und alles funktioniert gut.

Allerdings möchte ich in der Lage sein, den Benutzern eine Möglichkeit zu geben, eine Aufgabe abzubrechen, wenn der Server überlastet ist, indem Sie überprüfen, wie viele Aufgaben derzeit geplant sind.

Wie kann ich das erreichen?

Ich verwende redis als Broker.

ich dies gerade gefunden: Retrieve list of tasks in a queue in Celery

Es ist irgendwie auf meine Frage beziehen, aber ich brauche nicht die Aufgaben aufzulisten, so dass sie nur zählen :)

Answer 1

Wenn Ihr Broker als redis://localhost:6379/1 konfiguriert ist, und die Aufgaben in die allgemeinen celery Warteschlange gestellt werden, dann können Sie die Länge durch die folgenden Mittel erhalten:

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

Oder von einem Shell-Skript (gut für Monitore und so):

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

Answer 2

Wenn Sie bereits redis in Ihrer Anwendung konfiguriert haben, können Sie dies versuchen:

from celery import Celery 

QUEUE_NAME = 'celery' 

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

length = client.llen(QUEUE_NAME) 

Answer 3

Hier ist, wie Sie die Anzahl der Nachrichten in einer Warteschlange mit Sellerie erhalten können, die Broker ist agnostisch.

Mithilfe von connection_or_acquire können Sie die Anzahl der offenen Verbindungen zu Ihrem Broker minimieren, indem Sie das interne Verbindungs-Pooling von Sellery verwenden.

celery = Celery(app) 

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

Sie können auch Sellerie erweitern diese Funktionalität zu bieten:

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')