如何在 ASP.NET Core 应用程序中持续监听 Pub/Sub 消息？

Question

我想实现一个 ASP.NET Core API，它不响应 HTTP 请求，但在启动时开始侦听 Google Cloud Pub/Sub 消息，并且在整个生命周期中无限期地侦听。

使用官方 Pub/Sub SDK 实现此功能的首选方式是什么？

我可以想到两种方法：

方法1：只需使用SimpleSubscriber，并在Startup.Configure 开始收听消息：

public void Configure(IApplicationBuilder app)
{
    var simpleSubscriber = await SimpleSubscriber.CreateAsync(subscriptionName);
    var receivedMessages = new List<PubsubMessage>();

    simpleSubscriber.StartAsync((msg, cancellationToken) =>
    {
        // Process the message here.

        return Task.FromResult(SimpleSubscriber.Reply.Ack);
    });

    ...
}

方法 2：使用专门创建的库来定期运行作业，例如 Quartz、Hangfire 或 FluentScheduler，每次触发作业时，使用 SubscriberClient 拉取新消息。

哪种方法是首选方法？第一个看起来比较简单，但我不确定它是否真的可靠。

Answer 1

第一种方法肯定是打算如何使用它。

但是，请参阅StartAsync 的文档：

开始接收消息。返回的Task 在以下任一情况下完成 StopAsync(CancellationToken) 被调用或 如果不可恢复 发生故障。此方法不能多次调用 SubscriberClient 实例。

因此，您确实需要处理因不可恢复错误而导致的意外 StartAsync 关机。最简单的做法是使用外部循环，尽管考虑到这些错误被认为是不可恢复的，因此可能需要更改调用的某些内容才能成功。

代码可能如下所示：

while (true)
{
    // Each SubscriberClientinstance must only be used once.
    var subscriberClient = await SubscriberClient.CreateAsync(subscriptionName);
    try
    {
        await subscriberClient.StartAsync((msg, cancellationToken) =>
        {
            // Process the message here.
            return Task.FromResult(SimpleSubscriber.Reply.Ack);
        });
    }
    catch (Exception e)
    {
        // Handle the unrecoverable error somehow...
    }
}

如果这没有按预期工作，请let us know。

编辑：SimpleSubscriber 在库中已重命名为 SubscriberClient，因此已对答案进行了相应编辑。