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 definitiv eine Zeitüberschreitung aufweist, bevor sie erneut versucht wird.