ASP.NET Web Api 的事件发布者

Question

我已经开始使用微服务，我需要创建一个事件发布机制。

我计划使用 Amazon SQS。

这个想法很简单。我将事件存储在与聚合相同的事务中的数据库中。 如果用户更改他的电子邮件，事件UserChangedEmail 将存储在数据库中。

我还有事件处理程序，例如UserChangedEmailHandler，它将（在这种情况下）负责将此事件发布到 SQS 队列，以便其他服务可以知道用户更改了电子邮件。

我的问题是，实现这一目标的做法是什么？我是否应该有某种后台定时进程来扫描事件表并将事件发布到 SQS？ 这可以是 WebApi 应用程序中的进程（首选），还是应该是一个单独的进程？

其中一个想法是使用 Hangfire，但它不支持不到一分钟的 cron 作业。

有什么建议吗？

编辑：

正如其中一个答案所建议的那样，我已经查看了 NServicebus。 NServiceBus page 上的示例之一显示了我关注的核心。

在他们的示例中，他们创建了一个已下订单的日志。如果日志或数据库条目成功提交，但发布中断和事件从未发布怎么办？

这是事件处理程序的代码：

public class PlaceOrderHandler :
    IHandleMessages<PlaceOrder>
{
    static ILog log = LogManager.GetLogger<PlaceOrderHandler>();
    IBus bus;

    public PlaceOrderHandler(IBus bus)
    {
        this.bus = bus;
    }

    public void Handle(PlaceOrder message)
    {
        log.Info($"Order for Product:{message.Product} placed with id: {message.Id}");
        log.Info($"Publishing: OrderPlaced for Order Id: {message.Id}");

        var orderPlaced = new OrderPlaced
        {
            OrderId = message.Id
        };
        bus.Publish(orderPlaced); <!-- my concern
    }
}

Answer 1

现成的建议

我建议不要自己动手，而是研究现成的产品，因为这里有很多复杂性，一开始不会很明显，例如

• 管理事件订阅者列表 - SQS 队列更适合与事件使用者配对，而不是与事件生产者配对，因为当消息被消耗时，它在队列中不再可用 - 因此，如果您想支持多个订阅者给定事件（这是事件驱动架构的巨大优势），您如何知道在事件消息首次引发时将其推送到哪些 SQS 队列？
• 重试语义、错误转发队列 - 处理因临时基础设施问题导致的临时错误与因业务逻辑语义问题导致的永久错误
• 审计跟踪哪些消息在何时提出以及在何处发送
• 通过 SQS 发送的消息的安全性（您的业务案例是否要求对其进行加密？SQS 是 Amazon 提供的一项应用服务，不提供存储级加密
• 消息大小 - SQS 有消息大小限制，因此您最终可能需要处理大型消息的带外传输

这只是我的想法......

一些可以提供帮助的现成系统：

• NServiceBus 提供了一个管理命令和事件消息传递的框架，并且它有一个允许灵活传输类型的插件框架 - NServiceBus.SQS 提供 SQS 作为传输。
  • 提供全面而灵活的重试、审核和错误处理
  • 命令与事件的主观使用（命令消息说“执行此操作”并发送到单个服务进行处理，事件消息说“发生了某事”并发送给任意数量的灵活订阅者）
  • 发件箱模式提供事务一致的消息传递，即使使用非事务一致的传输（例如 SQS）也是如此
  • 目前 SQS 插件使用默认的 NServiceBus 订阅者持久性，这需要 SQL Server 来存储事件订阅者列表（有关利用 SNS 的选项，请参见下文）
  • 内置对 sagas 的支持，提供一个框架以通过补偿操作确保多事务最终与回滚保持一致
  • 支持计划消息处理的超时
  • 商业产品，所以不是免费的，但许多插件/扩展是开源的
• Mass Transit
  • 不支持现成的 SQS，但支持 Azure 服务总线和 RabbitMq，因此如果可以的话，可以作为您的替代方案
  • 与 NServiceBus 类似，但并非 100% 相同 - NServiceBus vs MassTransit 提供全面比较
  • 完全开源/免费
• Just Saying
  • 专为基于 SQS/SNS 设计的轻量级开源消息传递框架
  • 每个事件的SNS主题，每个微服务的SQS队列，使用原生SNS SQS队列订阅实现扇出
  • 开源免费

可能还有其他人，我对 NServiceBus 的个人经验最为丰富，但我强烈建议您研究现成的解决方案 - 它们会让您腾出时间来开始根据业务事件设计您的系统，而不必担心事件传输的机制。

即使您确实想构建自己的学习练习，回顾上述工作如何为您提供一些关于可靠事件驱动消息传递所需的提示。

事务一致性和发件箱模式

已编辑问题以询问如果部分操作成功但发布操作失败会发生什么。我已经看到这被称为消息传递的事务一致性，它通常意味着在一个事务中，所有业务副作用都被提交，或者没有。业务副作用可能意味着：

• 数据库记录已更新
• 删除了另一条数据库记录
• 消息发布到消息队列
• 已发送电子邮件

如果数据库操作失败，您通常不希望发送电子邮件或发布消息，同样，如果消息发布失败，您也不希望提交数据库操作。

那么如何保证消息的一致性呢？

NServiceBus 以两种方式之一处理此问题：

1. 使用事务一致的消息传输，例如 MSMQ。
   1. MSMQ 能够使用Microsoft's DTC (Distributed Transaction Coordinator)，并且 DTC 可以在分布式事务中注册消息发布，并带有 SQL 服务器更新 - 这意味着如果您的业务事务失败，您的发布操作将被回滚，反之亦然
2. Outbox Pattern
   1. 使用发件箱模式，消息不会立即发送 - 它们会添加到数据库中的发件箱表中，最好是与您的业务数据相同的数据库，作为同一事务的一部分
   2. 事务提交后，它会尝试发送每条消息，并且仅在发送成功后将其从发件箱中删除
   3. 如果系统在发送之后但在删除之前发生故障，则消息将被第二次传输。为了弥补这一点，当启用 Outbox 时，NServiceBus 还将通过维护所有入站消息的记录并丢弃重复项来对入站消息进行重复数据删除。
   4. 重复数据删除对于 Amazon SQS 尤其有用，因为它本身最终是一致的，并且可能会收到两次相同的消息。
   5. 这与您问题中的原始概念相距不远，但存在差异：
      1. 您正在构思一个后台定时进程来扫描事件表（又名发件箱表）并将事件发布到 SQS
      2. NServiceBus 在pipeline 中执行处理程序 - 使用发件箱，将消息分派到传输（也就是将消息推送到 SQS 队列）只是管道中的最后一步。因此 - 每当处理消息时，处理期间生成的任何出站消息都将在业务事务提交后立即分派 - 无需对事件表进行定时扫描。
   6. 注意：仅当存在环境 NServiceBus 处理程序事务时，发件箱才会成功 - 即当您在 NServiceBus 管道中处理消息时。在某些情况下不会出现这种情况，例如WebAPI 请求管道。出于这个原因，NServiceBus recommends using your API request to send a single Command message only，然后将业务数据操作与后端端点服务中事务一致的命令处理程序中的进一步消息传递相结合。尽管他们文档中的第 3 点与 MSMQ 比 SQS 传输更相关。

处理程序语义

关于您的提案的更多评论 - 按照惯例，UserChangedEmailHandler 通常与响应电子邮件更改的服务相关联，而不是简单地参与传播电子邮件已更改的信息.当您的系统发布了 50 个事件时，您是否需要 50 个不同的处理程序来将这些消息推送到不同的队列中？

上述系统使用通用框架通过传输传播消息，因此您可以为订阅系统保留UserChangedEmailHandler，并在其中包含用户更改电子邮件时应发生的业务逻辑。

Answer 2

无论如何，我都会选择有状态的服务。如果您想稍微放松一下，请查看Azure Service Fabric。

就我而言，我有自己的一组微服务，在这样的场景中，我首先在 db 上进行了基本的创建操作（更改电子邮件）。我有一个事件实体并推回了该集合中的一个事件（在本例中为 mongodb）。有状态服务正在轮询数据库并批量处理事件。

现在，在您的情况下，如果您的 Web 应用程序进程是持久的，您可以选择立即将消息排入队列并保留一个字段，以防它后来是否由任何服务实际处理。我使用 mongodb 作为数据库，使用 Azure 服务总线作为消息代理。我认为 Amazon SQS 会是类似的。

现在，如果您的 Web 应用程序是一个普通的 asp.net Web api 或 mvc 进程，您只需在数据库中登记事件并离开，因为您不必在每次收到请求时都创建消息代理侦听器.一个服务可以轮询数据库，使用消息代理让其他服务知道。

如果您想要一个完整的事件驱动范例，您可能需要查看Event Hubs

我强烈建议密切关注消息总线是否已处理任何资源，以确保其可靠。

希望对您有所帮助。 :)