Der beste Weg, um Nachrichten aus DLQ in Amazon SQS zu verschieben?

Was ist die beste Vorgehensweise, um Nachrichten aus einer Warteschlange für nicht zustellbare Nachrichten wieder in die ursprüngliche Warteschlange in Amazon SQS zu verschieben?

Wäre es

1. Nachricht von DLQ abrufen
2. Nachricht in die Warteschlange schreiben
3. Nachricht aus DLQ löschen

Oder gibt es einen einfacheren Weg?

Wird AWS irgendwann ein Tool in der Konsole haben, um Nachrichten aus dem DLQ zu verschieben?

Hier ist ein kurzer Hack. Dies ist definitiv nicht die beste oder empfohlene Option.

1. Legen Sie die Haupt-SQS-Warteschlange als DLQ für die tatsächliche DLQ mit Maximum Receives als 1 fest.
2. Anzeigen des Inhalts in DLQ (Dadurch werden die Nachrichten in die Hauptwarteschlange verschoben, da dies der DLQ für den tatsächlichen DLQ ist.)
3. Entfernen Sie die Einstellung, damit die Hauptwarteschlange nicht mehr die DLQ der tatsächlichen DLQ ist

Es gibt einige Skripte, die dies für Sie tun:

• npm / nodejs basiert: http://github.com/garryyao/replay-aws-dlq

# install
npm install replay-aws-dlq;

# use
npx replay-aws-dlq [source_queue_url] [dest_queue_url]

• Go based: https://github.com/mercury2269/sqsmover

# compile: https://github.com/mercury2269/sqsmover#compiling-from-source

# use
sqsmover -s [source_queue_url] -d [dest_queue_url] 

Sie müssen die Nachricht nicht verschieben, da sie mit so vielen anderen Herausforderungen verbunden ist, wie doppelten Nachrichten, Wiederherstellungsszenarien, verlorenen Nachrichten, Deduplizierungsprüfung usw.

Hier ist die Lösung, die wir implementiert haben -

Normalerweise verwenden wir den DLQ für vorübergehende Fehler, nicht für dauerhafte Fehler. Also unten Ansatz genommen -

1. Lesen Sie die Nachricht von DLQ wie eine normale Warteschlange

   Leistungen
   • Um doppelte Nachrichtenverarbeitung zu vermeiden
   • Bessere Kontrolle über DLQ - Wie ich es überprüft habe, nur zu verarbeiten, wenn die reguläre Warteschlange vollständig verarbeitet ist.
   • Skalieren Sie den Prozess basierend auf der Nachricht in DLQ

2. Folgen Sie dann demselben Code, dem die reguläre Warteschlange folgt.

3. Zuverlässiger im Falle eines Abbruchs des Jobs oder der Beendigung des Prozesses während der Verarbeitung (z. B. Instanz beendet oder Prozess beendet)

   Leistungen
   • Wiederverwendbarkeit des Codes
   • Fehlerbehandlung
   • Wiederherstellung und Nachrichtenwiedergabe

4. Erweitern Sie die Nachrichtensichtbarkeit, damit kein anderer Thread sie verarbeitet.

   Vorteil
   • Vermeiden Sie es, denselben Datensatz von mehreren Threads zu verarbeiten.

5. Löschen Sie die Nachricht nur, wenn entweder ein dauerhafter Fehler vorliegt oder erfolgreich ist.

   Vorteil
   • Verarbeiten Sie so lange, bis ein vorübergehender Fehler auftritt.

Das scheint Ihre beste Option zu sein. Es besteht die Möglichkeit, dass Ihr Prozess nach Schritt 2 fehlschlägt. In diesem Fall wird die Nachricht am Ende zweimal kopiert, aber Ihre Anwendung sollte die erneute Zustellung von Nachrichten trotzdem behandeln (oder sich nicht darum kümmern).

Hier:

import boto3
import sys
import Queue
import threading

work_queue = Queue.Queue()

sqs = boto3.resource('sqs')

from_q_name = sys.argv[1]
to_q_name = sys.argv[2]
print("From: " + from_q_name + " To: " + to_q_name)

from_q = sqs.get_queue_by_name(QueueName=from_q_name)
to_q = sqs.get_queue_by_name(QueueName=to_q_name)

def process_queue():
    while True:
        messages = work_queue.get()

        bodies = list()
        for i in range(0, len(messages)):
            bodies.append({'Id': str(i+1), 'MessageBody': messages[i].body})

        to_q.send_messages(Entries=bodies)

        for message in messages:
            print("Coppied " + str(message.body))
            message.delete()

for i in range(10):
     t = threading.Thread(target=process_queue)
     t.daemon = True
     t.start()

while True:
    messages = list()
    for message in from_q.receive_messages(
            MaxNumberOfMessages=10,
            VisibilityTimeout=123,
            WaitTimeSeconds=20):
        messages.append(message)
    work_queue.put(messages)

work_queue.join()

Es gibt eine andere Möglichkeit, dies zu erreichen, ohne eine einzige Codezeile zu schreiben. Angenommen, Ihr tatsächlicher Warteschlangenname lautet SQS_Queue und der DLQ dafür ist SQS_DLQ. Befolgen Sie nun diese Schritte:

1. Legen Sie SQS_Queue als dlq von SQS_DLQ fest. Da SQS_DLQ bereits ein dlq von SQS_Queue ist. Jetzt fungieren beide als dlq des anderen.
2. Setzen Sie die maximale Empfangsanzahl Ihres SQS_DLQ auf 1.
3. Lesen Sie nun Nachrichten von der SQS_DLQ-Konsole. Da die Anzahl der Nachrichtenempfänge 1 beträgt, wird die gesamte Nachricht an ihre eigene dlq gesendet, die Ihre eigentliche SQS_Queue-Warteschlange ist.

Ich habe dazu ein kleines Python-Skript geschrieben, indem ich boto3 lib verwendet habe:

conf = {
  "sqs-access-key": "",
  "sqs-secret-key": "",
  "reader-sqs-queue": "",
  "writer-sqs-queue": "",
  "message-group-id": ""
}

import boto3
client = boto3.client(
    'sqs',
        aws_access_key_id       = conf.get('sqs-access-key'),
        aws_secret_access_key   = conf.get('sqs-secret-key')
)

while True:
    messages = client.receive_message(QueueUrl=conf['reader-sqs-queue'], MaxNumberOfMessages=10, WaitTimeSeconds=10)

    if 'Messages' in messages:
        for m in messages['Messages']:
            print(m['Body'])
            ret = client.send_message( QueueUrl=conf['writer-sqs-queue'], MessageBody=m['Body'], MessageGroupId=conf['message-group-id'])
            print(ret)
            client.delete_message(QueueUrl=conf['reader-sqs-queue'], ReceiptHandle=m['ReceiptHandle'])
    else:
        print('Queue is currently empty or messages are invisible')
        break

Sie können dieses Skript unter diesem Link erhalten

Dieses Skript kann grundsätzlich Nachrichten zwischen beliebigen Warteschlangen verschieben. und es unterstützt FIFO-Warteschlangen sowie die Bereitstellung des message_group_idFeldes.

Wir verwenden das folgende Skript, um Nachrichten von der src-Warteschlange zur tgt-Warteschlange umzuleiten:

Dateiname: redrive.py

Verwendung: python redrive.py -s {source queue name} -t {target queue name}

'''
This script is used to redrive message in (src) queue to (tgt) queue

The solution is to set the Target Queue as the Source Queue's Dead Letter Queue.
Also set Source Queue's redrive policy, Maximum Receives to 1. 
Also set Source Queue's VisibilityTimeout to 5 seconds (a small period)
Then read data from the Source Queue.

Source Queue's Redrive Policy will copy the message to the Target Queue.
'''
import argparse
import json
import boto3
sqs = boto3.client('sqs')


def parse_args():
    parser = argparse.ArgumentParser()
    parser.add_argument('-s', '--src', required=True,
                        help='Name of source SQS')
    parser.add_argument('-t', '--tgt', required=True,
                        help='Name of targeted SQS')

    args = parser.parse_args()
    return args


def verify_queue(queue_name):
    queue_url = sqs.get_queue_url(QueueName=queue_name)
    return True if queue_url.get('QueueUrl') else False


def get_queue_attribute(queue_url):
    queue_attributes = sqs.get_queue_attributes(
        QueueUrl=queue_url,
        AttributeNames=['All'])['Attributes']
    print(queue_attributes)

    return queue_attributes


def main():
    args = parse_args()
    for q in [args.src, args.tgt]:
        if not verify_queue(q):
            print(f"Cannot find {q} in AWS SQS")

    src_queue_url = sqs.get_queue_url(QueueName=args.src)['QueueUrl']

    target_queue_url = sqs.get_queue_url(QueueName=args.tgt)['QueueUrl']
    target_queue_attributes = get_queue_attribute(target_queue_url)

    # Set the Source Queue's Redrive policy
    redrive_policy = {
        'deadLetterTargetArn': target_queue_attributes['QueueArn'],
        'maxReceiveCount': '1'
    }
    sqs.set_queue_attributes(
        QueueUrl=src_queue_url,
        Attributes={
            'VisibilityTimeout': '5',
            'RedrivePolicy': json.dumps(redrive_policy)
        }
    )
    get_queue_attribute(src_queue_url)

    # read all messages
    num_received = 0
    while True:
        try:
            resp = sqs.receive_message(
                QueueUrl=src_queue_url,
                MaxNumberOfMessages=10,
                AttributeNames=['All'],
                WaitTimeSeconds=5)

            num_message = len(resp.get('Messages', []))
            if not num_message:
                break

            num_received += num_message
        except Exception:
            break
    print(f"Redrive {num_received} messages")

    # Reset the Source Queue's Redrive policy
    sqs.set_queue_attributes(
        QueueUrl=src_queue_url,
        Attributes={
            'VisibilityTimeout': '30',
            'RedrivePolicy': ''
        }
    )
    get_queue_attribute(src_queue_url)


if __name__ == "__main__":
    main()

DLQ kommt nur ins Spiel, wenn der ursprüngliche Verbraucher die Nachricht nach verschiedenen Versuchen nicht erfolgreich konsumiert. Wir möchten die Nachricht nicht löschen, da wir glauben, dass wir noch etwas damit anfangen können (möglicherweise versuchen, sie erneut zu verarbeiten oder zu protokollieren oder einige Statistiken zu sammeln), und wir möchten nicht immer wieder auf diese Nachricht stoßen und die Fähigkeit dazu stoppen andere Nachrichten dahinter verarbeiten.

DLQ ist nichts anderes als eine weitere Warteschlange. Das heißt, wir müssten einen Consumer für DLQ schreiben, der idealerweise weniger häufig ausgeführt wird (im Vergleich zur ursprünglichen Warteschlange), der von DLQ verbraucht wird und Nachrichten zurück in die ursprüngliche Warteschlange erzeugt und aus DLQ löscht - wenn dies das beabsichtigte Verhalten ist und wir denken Der ursprüngliche Verbraucher wäre jetzt bereit, es erneut zu verarbeiten. Es sollte in Ordnung sein, wenn dieser Zyklus eine Weile andauert, da wir jetzt auch die Möglichkeit haben, manuell zu prüfen und notwendige Änderungen vorzunehmen und eine andere Version des ursprünglichen Verbrauchers bereitzustellen, ohne die Nachricht zu verlieren (natürlich innerhalb der Aufbewahrungsfrist für Nachrichten - die 4 Tage beträgt) Standard).

Es wäre schön, wenn AWS diese Funktion sofort bereitstellen würde, aber ich sehe sie noch nicht - sie überlassen dies dem Endbenutzer, um sie so zu verwenden, wie sie es für angemessen halten.