部署SQS使用者

Question

我希望运行一个服务，该服务将使用放入SQS队列中的消息。构建消费者应用程序的最佳方法是什么？

一种想法是创建一堆线程或进程来运行这个：

def run(q, delete_on_error=False):
    while True:
        try:
            m = q.read(VISIBILITY_TIMEOUT, wait_time_seconds=MAX_WAIT_TIME_SECONDS)
            if m is not None:
                try:
                    process(m.id, m.get_body())
                except TransientError:
                    continue
                except Exception as ex:
                    log_exception(ex)
                    if not delete_on_error:
                        continue
                q.delete_message(m)
        except StopIteration:
            break
        except socket.gaierror:
            continue

我倾向于单独的进程而不是线程来避免GIL。是否有一些容器进程可以用来启动和监视这些单独运行的进程？

Answer 1

有几件事：

• SQS API允许您通过单个API调用接收多个消息（最多10个消息，或最多256K值的消息，以先达到的限制为准）。利用这一特性可以降低成本，因为您是按API调用计费的。看起来您正在使用boto库--请查看get_messages。
• 在当前的代码中，如果处理消息时由于瞬时错误而失败，则在可见性超时到期之前无法再次处理该消息。您可能需要考虑立即将消息返回到队列中。您可以通过在该消息上使用0调用change_visibility来完成此操作。然后，该消息将可直接用于处理。（看起来，如果您这样做，那么该消息的可见性超时将永久更改--实际上并非如此。AWS文档声明“下次接收消息时的可见性超时将恢复到原始超时值”。有关详细信息，请参阅文档。）

如果您正在寻找一个健壮的SQS消息使用者的示例，那么您可能希望查看NServiceBus.AmazonSQS（我是其作者）。(C#-对不起，我找不到任何python示例。）