c# - 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作为传输。


提供全面，灵活的重试，审核和错误处理
公然使用命令vs事件（命令消息说“执行此操作”，并发送到单个服务进行处理，事件消息说“发生了什么事”，并发送到任意数量的灵活订户）
发件箱模式提供了事务一致的消息传递，即使使用非事务一致的传输，例如SQS
当前，SQS插件使用默认的NServiceBus订阅者持久性，这需要使用SQL Server来存储事件订阅者列表（有关利用SNS的选项，请参见下文）
内置对sagas的支持，提供了一个框架，以通过补偿动作来确保多事务最终与回滚的一致性
支持预定消息处理的超时
商业产品，因此不是免费的，但是许多插件/扩展是开源的

Mass Transit


不支持现成的SQS，但支持Azure Service Bus和RabbitMq，因此如果可以的话，可以替代您
与NServiceBus类似的产品，但并非100％相同-NServiceBus vs MassTransit提供了全面的比较
完全开源/免费

Just Saying


轻量级开源消息传递框架，专为基于SQS / SNS的设计
每个事件的SNS主题，每个微服务的SQS队列，使用本机SNS SQS Queue订阅来实现扇出
免费开源



可能还有其他人，并且我在NServiceBus方面拥有最丰富的个人经验，但是我强烈建议您研究现成的解决方案-它们将使您腾出时间来根据业务事件开始设计系统，而不用担心系统的工作原理事件传输。

即使您确实想构建自己的学习练习，但回顾一下上述工作的方式可能会为您提供一些可靠的事件驱动消息传递所需的技巧。

交易一致性和发件箱模式

已对问题进行了编辑，以询问如果部分操作成功但发布操作失败会发生什么情况。我已经将其称为消息传递的事务一致性，它通常意味着在事务中，所有业务副作用都已落实，或者没有。业务副作用可能意味着：


数据库记录已更新
另一个数据库记录已删除
消息已发布到消息队列
邮件已发送


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

那么如何确保消息传递的一致性呢？

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


使用事务一致的消息传输，例如MSMQ。


MSMQ能够使用Microsoft's DTC (Distributed Transaction Coordinator)，而DTC可以使用SQL Server更新在分布式事务中注册消息的发布-这意味着，如果您的业务事务失败，则发布操作将被回滚，反之亦然

Outbox Pattern


使用发件箱模式，不会立即发送消息-作为同一事务的一部分，消息将被添加到数据库（最好是与业务数据相同的数据库）中的发件箱表中
提交事务后，它将尝试分派每条消息，并仅在成功分派后将其从发件箱中删除
如果在发送后但删除之前系统发生故障，则将再次发送该消息。为了弥补这一点，启用发件箱后，NServiceBus还将通过保留所有入站消息的记录并丢弃重复项来对入站消息进行重复数据删除。
重复数据删除对于Amazon SQS尤其有用，因为它本身最终是一致的，并且相同的消息可能会收到两次。
这与您的问题中的原始概念相去不远，但是有一些区别：


您正在构思后台定时过程，以扫描事件表（又称为发件箱表）并将事件发布到SQS
NServiceBus在pipeline中执行处理程序-使用发件箱，将消息分发到传输（也就是将消息推入SQS队列）只是管道中的最后步骤之一。因此，每当处理一条消息时，在处理事务过程中生成的所有出站消息都将在提交业务事务后立即被调度，而无需对事件表进行定时扫描。

注意：仅当有环境NServiceBus Handler事务时，即在NServiceBus管道中处理消息时，发件箱才成功。在某些情况下（例如， WebAPI请求管道。因此，NServiceBus recommends using your API request to send a single Command message only，然后将业务数据操作与后端端点服务中事务一致的命令处理程序内的其他消息传递结合在一起。尽管文档中的第3点与MSMQ的关系比与SQS传输的关系更大。



处理程序语义

关于您的建议的另一条评论-按照惯例，UserChangedEmailHandler通常会与响应电子邮件更改而执行某些操作的服务相关联，而不是简单地参与电子邮件更改信息的传播。当系统发布了50个事件时，是否要50个不同的处理程序仅将这些消息推送到不同的队列中？

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