Лучший способ переместить сообщения из DLQ в Amazon SQS?

Как лучше всего переместить сообщения из очереди недоставленных сообщений обратно в исходную очередь в Amazon SQS?

Будет ли это

1. Получить сообщение от DLQ
2. Написать сообщение в очередь
3. Удалить сообщение из DLQ

Или есть способ попроще?

Кроме того, будет ли у AWS в конечном итоге инструмент в консоли для перемещения сообщений из DLQ?

Вот быстрый совет. Это определенно не лучший и не рекомендуемый вариант.

1. Установите основную очередь SQS как DLQ для фактического DLQ с максимальным количеством приемов равным 1.
2. Просмотрите содержимое в DLQ (это переместит сообщения в основную очередь, поскольку это DLQ для фактического DLQ)
3. Удалите настройку, чтобы основная очередь больше не была DLQ фактического DLQ.

Есть несколько скриптов, которые сделают это за вас:

• на основе npm / nodejs: http://github.com/garryyao/replay-aws-dlq

# install
npm install replay-aws-dlq;

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

• перейти на основе: https://github.com/mercury2269/sqsmover

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

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

Не нужно перемещать сообщение, потому что оно сопряжено с множеством других проблем, таких как дублирование сообщений, сценарии восстановления, потерянное сообщение, проверка исключения дубликатов и т. Д.

Вот решение, которое мы реализовали -

Обычно мы используем DLQ для временных ошибок, а не для постоянных ошибок. Итак, подход ниже -

1. Прочтите сообщение из DLQ как обычную очередь

   Льготы
   • Чтобы избежать дублирования обработки сообщений
   • Лучший контроль на DLQ - вроде я поставил чек, обрабатывать только тогда, когда обычная очередь полностью обработана.
   • Масштабируйте процесс на основе сообщения в DLQ

2. Затем следуйте тому же коду, что и обычная очередь.

3. Более надежен в случае прерывания задания или прекращения процесса во время обработки (например, инстанс убит или процесс завершен)

   Льготы
   • Возможность повторного использования кода
   • Обработка ошибок
   • Восстановление и воспроизведение сообщений

4. Расширьте видимость сообщений, чтобы никакой другой поток их не обрабатывал.

   Выгода
   • Избегайте обработки одной и той же записи несколькими потоками.

5. Удаляйте сообщение только в том случае, если есть постоянная ошибка или успешно.

   Выгода
   • Продолжайте обработку, пока мы не получим временную ошибку.

Похоже, это ваш лучший вариант. Существует вероятность того, что ваш процесс завершится ошибкой после шага 2. В этом случае вы дважды скопируете сообщение, но ваше приложение все равно должно обрабатывать повторную доставку сообщений (или не заботиться).

Вот:

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

Есть другой способ добиться этого без написания единственной строчки кода. Учтите, что ваше фактическое имя очереди - SQS_Queue, а DLQ для нее - SQS_DLQ. Теперь выполните следующие действия:

1. Установите SQS_Queue как dlq для SQS_DLQ. Поскольку SQS_DLQ уже является dlq SQS_Queue. Теперь оба действуют как dlq друг друга.
2. Установите максимальное количество приемов вашего SQS_DLQ равным 1.
3. Теперь прочтите сообщения из консоли SQS_DLQ. Поскольку счетчик получения сообщений равен 1, он отправит все сообщения в свой собственный dlq, который является вашей реальной очередью SQS_Queue.

Для этого я написал небольшой скрипт на Python, используя boto3 lib:

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

вы можете получить этот скрипт по этой ссылке

этот скрипт в основном может перемещать сообщения между любыми произвольными очередями. и он поддерживает очереди fifo, а также вы можете указать message_group_idполе.

Мы используем следующий скрипт для перенаправления сообщения из очереди src в очередь tgt:

имя файла: redrive.py

Применение: 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 вступает в игру только тогда, когда исходный потребитель не может успешно принять сообщение после различных попыток. Мы не хотим удалять сообщение, так как считаем, что мы все еще можем что-то с ним сделать (возможно, попытаться снова обработать или зарегистрировать его или собрать некоторую статистику), и мы не хотим снова и снова сталкиваться с этим сообщением и останавливать возможность обрабатывать другие сообщения за этим.

DLQ - это не что иное, как очередная очередь. Это означает, что нам нужно будет написать потребителя для DLQ, который в идеале работал бы реже (по сравнению с исходной очередью), который потреблял бы из DLQ и возвращал сообщение в исходную очередь и удалял его из DLQ - если это предполагаемое поведение, и мы думаем исходный потребитель теперь будет готов обработать его снова. Если этот цикл продолжится какое-то время, все должно быть в порядке, поскольку теперь у нас также есть возможность вручную проверить и внести необходимые изменения и развернуть другую версию исходного потребителя без потери сообщения (конечно, в течение периода хранения сообщения, который составляет 4 дня к по умолчанию).

Было бы неплохо, если бы AWS предоставил эту возможность из коробки, но я ее еще не вижу - они оставляют это на усмотрение конечного пользователя, чтобы использовать его так, как они считают нужным.