Обработка сообщений из AWS SQS с использованием нескольких хостов

Question

У меня есть приложение с участием AWS SNS, SQS и должно выполняться на нескольких хостах. Точное описание проблемы:

Всякий раз, когда есть событие, сообщение, содержащее ID, публикуется в теме SNS, на которой подписана очередь SQS. Теперь я получаю сообщения в очереди. Теперь я хочу, чтобы несколько хостов читали сообщения из очереди (ни один из двух хостов не должен читать одно и то же сообщение) и записывать сообщение в общий файл в Amazon S3. Проблемы, такие как «что делать, если хост, читающий сообщение, не работает» и «не читать одно и то же сообщение дважды на одном хосте или разных хостах», следует рассмотреть.

Может ли кто-нибудь предложить какой-либо метод или некоторые ссылки, через которые я могу пройти, чтобы достичь этой задачи?

Answer 1

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

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

Это документально here

Answer 2

Раствор рассеянную вокруг. Вы можете прочитать SQS Dead Letter queue setup и ссылку на мой пример. вы можете делать то же самое, используя консоль AWS SQS без кодирования.

import boto3 
sqs = boto3.client("sqs") 
# I want to "lock" my queue for 5 minutes to allow my process have time to 
# complete the task and delete the message afterwards. 
response = sqs.create_queue(
    QueueName="foo", 
    Attributes= { 
     "VisibilityTimeout" : "300" 
    } 
) 
# create a queue to store the "dead letter message" 
dlq_response = sqs.create_queue(
    QueueName="dlq-foo", 
    Attributes= { 
     "VisibilityTimeout" : "300" 
    } 
) 
queue_url = response["QueueUrl"] 

# Attach RedrivePolicy to drive message to dead letter queue 
# I want to make sure the message only read 1 time. Assume the program crash 
# if it is not deleted. 
# deadLetterTargetArn : You must specify the queue exact region name, 
# exact Account name(replace 1234567890) and your dead letter queue name dlq-foo 
sqs.set_queue_attributes(
    QueueUrl = queue_url, 
    Attributes = { 
     "RedrivePolicy" : """{ 
       "maxReceiveCount" : "1" , 
       "deadLetterTargetArn" : "arn:aws:sqs:<region-name>:1234567890:dlq-foo" 
      }""" 
     } 
    ) 

Примечание: RedrivePolicy имеет доступ только к литеральным строкам, а не к словарю. Однако, как следует из выдержки, вам нужно поместить значение «словаря» и отформатировать его как строку. Вы можете использовать str (dict()) для преобразования dict в строку, я использую кавычки triple-quotes python для ясности чтения.