Asynchrone Django Q2-Aufgaben werden häufiger als erwartet ausgeführt ⇐ Python

[Anonymous]

Ich habe ein Problem mit asynchronen Aufgaben, bei denen meine Funktion anscheinend mehrmals für eine Aufgabe ausgeführt wird.
Hier sind meine Cluster-Einstellungen:

Code: Select all

Q_CLUSTER = {
'name': 'qcluster',
'workers': 4,
'threads': 1,
'recycle': 500,
'timeout': 30,
'retry': 300,
'max_attempts': 3,
'queue_limit': 8,
'bulk': 1,
'save_limit': 10000,
'orm': 'default',
'sync': bool(IS_TEST),
}

Ich habe 3 Modelle. Eines speichert eingehende Berichte, eine Artikeltabelle, in der jeder eingehende Bericht in eine Instanz der Artikeltabelle kategorisiert wird, und eine Treffertabelle, die die Anzahl der Berichte für diesen Artikel für jeden Tag aufzeichnet.
Berichte – Artikel: viele-zu-viele
Artikel – Treffer: eins-zu-viele

Code: Select all

def map_report(report_id: int) -> str:
"""Map report to item."""
logger.warning('Starting map report: (%s)', report_id)
report = Report.objects.get(pk=report_id)
item = matcher(report)
Hit.increment_item_hits(item=item)
logger.warning('Finished map report: (%s)', report_id)
return f'Item id: {item.pk}'

matcher gibt entweder ein neues oder ein vorhandenes Element zurück, das dem Bericht zugeordnet ist.
Ich rufe dies dann mit auf

Code: Select all

for report in Report.objects.all():
async_task(map_report, report_id=report.pk)

Wenn ich dies mit Testdaten von 1000 Berichten ausführe, funktioniert es korrekt. Wenn ich es jedoch mit größeren Datenmengen wie 5000 Aufgaben ausführe, scheint es, als würde es mehrmals ausgeführt. Innerhalb erfolgreicher Aufgaben wird die richtige Anzahl an Aufgaben ausgeführt, aber wenn ich die Protokolle überprüfe, zeigt sich, dass „map_reports“ anscheinend mehrmals für dieselbe Aufgabe ausgeführt wird, was dazu führt, dass meine Treffer am Ende mehr als erwartet ausfallen.
Auch wenn ich es synchron ausführe, erhalte ich die korrekte Anzahl an Treffern, aber asynchron führt dazu, dass die Treffer größer sind als erwartet, da „map_reports“ zu oft aufgerufen wird.
Ich habe ähnliche Probleme gesehen, wo die Lösung war Ich möchte die Wiederholungseinstellung des Clusters erhöhen, aber ich glaube, ich habe sie so eingerichtet, dass die Aufgabe auf jeden Fall eine Zeitüberschreitung erfährt, bevor sie erneut versucht wird.
Bearbeiten:
Das Ausführen des Programms mit einer deutlich längeren Wiederholungszeit hat dieses Problem gelöst. Wie in meinem Kommentar erwähnt, würde sich die Ergebniswarteschlange auf eine beträchtlich große Anzahl an Aufgaben aufbauen, sodass Aufgaben ausgeführt würden, aber in der Ergebniswarteschlange verbleiben würden, sodass sie erfolgreich wären. Aufgrund der enormen Größe der Ergebniswarteschlange würde jedoch die Sperre für die Aufgabe überschritten, was dazu führen würde, dass ein anderer Mitarbeiter die Aufgabe aufnimmt, obwohl sie bereits erfolgreich war. (Das habe ich durch Experimente abgeleitet und wäre dankbar, wenn jemand mit mehr Sachkenntnis dies bestätigen könnte.)