Vermeidung von doppelten Aufgaben in Sellerie broker

Ich möchte zu erstellen, der folgende Ablauf mit Sellerie-Konfiguration\api:

  • Senden TaskA(argB) Nur, wenn Sellerie Warteschlange hat keine TaskA(argB) bereits anhängig

Ist es möglich? wie?

InformationsquelleAutor arikg | 2014-11-09
  1. 2

    Ich kann nicht denken Sie an einen Weg, sondern

    1. Abrufen von allen ausgeführten und geplanten Aufgaben über Sellerie inspizieren

    2. Durchlaufen, um zu sehen, ob Ihre Aufgabe ist es.

    überprüfen diese Frage ALSO zu sehen, wie der erste Punkt erledigt ist.

    Glück

    InformationsquelleAutor srj
  2. 2

    Können Sie Ihre Arbeit bewusst von anderen Aufgaben durch eine Art von memoization. Wenn Sie einen cache-control-Taste (redis, memcached, /tmp, was auch immer praktisch ist), können Sie die Ausführung hängt von diesem Schlüssel. Ich bin mit redis als Beispiel.

    from redis import Redis

@app.task
def run_only_one_instance(params):
    try:
        sentinel =  Redis().incr("run_only_one_instance_sentinel")
        if sentinel == 1:
            #I am the legitimate running task
            perform_task()
        else:
            #Do you want to do something else on task duplicate?
            pass
        Redis().decr("run_only_one_instance_sentinel")
    except Exception as e:
        Redis().decr("run_only_one_instance_sentinel")
        # potentially log error with Sentry?
        # decrement the counter to insure tasks can run
        # or: raise e
    • Dies wird verhindern, dass die Aufgabe ausgeführt wird, wenn die gleiche Aufgabe, ist immer noch läuft. Und wenn die gleiche Aufgabe vorhanden 3-mal in der Warteschlange ausgeführt werden soll drei mal so lange, wie Sie nicht gleichzeitig ausgeführt
    • incr und decr sind atomar, so sind Sie falsch.
    • Ändert nichts an der Tatsache, dass mein Vorschlag nicht im Einklang mit der Frage (zumindest nicht mehr). Ich denke, die Frage hat sich durch die änderungen seit dem ich antwortete. Die akzeptierte Antwort ist schwer runtimewise, sondern wird tun, was gefragt ist.
    • Ich denke, Ramast könnten Recht haben, aber die Grundidee ist da. Es wäre besser, reset-Zähler auf 0 in der if-Zweig nach perform_task und entfernen Sie alle decr Anrufe. Auf diese Weise werden alle doppelten Anrufe wird nicht erneut ausgeführt, bis die reset-Aufruf. Derzeit könnte es passieren, dass task1 Anrufe incr und läuft Aufgabe task2 Anrufe incr aber nicht ausgeführt, da sentinel !=1 ; task1 Anrufe decr. Also im Grunde alle Aufgaben ausführen können, jetzt da es keine mehr doppelte Aufgaben ausgeführt perform_task, aber das wird nicht passieren, wenn eine neue task3 kommt jetzt, weil task2 noch nicht genannt und decr Zähler ist 1, nicht 0.
    InformationsquelleAutor Árni St. Sigurðsson
  3. 1

    Ich weiß nicht, es ist gonna helfen Sie mehr als die anderen Antworten, aber es geht meinen Ansatz, die gleiche Idee von srj. Ich brauchte einen Weg, um die Sperrung meines Servers zu starten, Aufgaben mit derselben id in die Warteschlange. Also machte ich eine Allgemeine Funktion, um mir zu helfen.

    def is_task_active_or_registered(app, task_id):

    i = app.control.inspect()

    active_dict = i.active()
    scheduled_dict = i.scheduled()
    keys_set = set(active_dict.keys() + scheduled_dict.keys())
    tasks_ids_set = set()

    for _dict in [active_dict, scheduled_dict]:
        for k in keys_set:
            for task in _dict[k]:
                tasks_ids_set.add(task['id'])

    if task_id in tasks_ids_set:
        return True
    else:
        return False

    So, ich benutze es so:

    In dem Kontext, in dem meine Sellerie-app-Objekt verfügbar ist, bestimme ich:

    def check_task_can_not_run(task_id):
    return is_task_active_or_registered(app=celery, task_id=task_id)

    Und so, von meinen Kunden verlangen, das nenne ich check_task_can_not_run(...) - und block-task aus gestartet werden, im Fall von True.

    InformationsquelleAutor Pedro Thompson
  4. 0

    Ich Stand vor dem ähnlichen problem. Der Beat wurde machen Duplikate in meiner Warteschlange. Ich wollte expires aber diese Funktion nicht richtig funktioniert https://github.com/celery/celery/issues/4300.

    So, hier ist der scheduler, der überprüft, ob die Aufgabe bereits in die Warteschlange eingereiht (basierend auf den task-Namen).

    # -*- coding: UTF-8 -*-
from __future__ import unicode_literals

import json
from heapq import heappop, heappush

from celery.beat import event_t
from celery.schedules import schedstate
from django_celery_beat.schedulers import DatabaseScheduler
from typing import List, Optional
from typing import TYPE_CHECKING

from your_project import celery_app

if TYPE_CHECKING:
    from celery.beat import ScheduleEntry


def is_task_in_queue(task, queue_name=None):
    # type: (str, Optional[str]) -> bool
    queues = [queue_name] if queue_name else celery_app.amqp.queues.keys()

    for queue in queues:
        if task in get_celery_queue_tasks(queue):
            return True
    return False


def get_celery_queue_tasks(queue_name):
    # type: (str) -> List[str]
    with celery_app.pool.acquire(block=True) as conn:
        tasks = conn.default_channel.client.lrange(queue_name, 0, -1)
        decoded_tasks = []

    for task in tasks:
        j = json.loads(task)
        task = j['headers']['task']
        if task not in decoded_tasks:
            decoded_tasks.append(task)

    return decoded_tasks


class SmartScheduler(DatabaseScheduler):
    """
    Smart means that prevents duplicating of tasks in queues.
    """
    def is_due(self, entry):
        # type: (ScheduleEntry) -> schedstate
        is_due, next_time_to_run = entry.is_due()

        if (
            not is_due or  # duplicate wouldn't be created
            not is_task_in_queue(entry.task)  # not in queue so let it run
        ):
            return schedstate(is_due, next_time_to_run)

        # Task should be run (is_due) and it is present in queue (is_task_in_queue)

        H = self._heap
        if not H:
            return schedstate(False, self.max_interval)

        event = H[0]
        verify = heappop(H)
        if verify is event:
            next_entry = self.reserve(entry)
            heappush(H, event_t(self._when(next_entry, next_time_to_run), event[1], next_entry))
        else:
            heappush(H, verify)
            next_time_to_run = min(verify[0], next_time_to_run)

        return schedstate(False, min(next_time_to_run, self.max_interval))
    InformationsquelleAutor Petr Přikryl
Schreibe einen Kommentar