使用 RabbitMQ 发布/订阅微服务事件总线时如何去重事件

Question

我一直在阅读第 58 页的This Book，以了解如何在微服务之间进行异步事件集成。

使用 RabbitMQ 和发布/订阅模式有助于将事件推送给订阅者。但是，考虑到微服务架构和 docker 使用情况，我希望有不止一次运行微服务“类型”的实例。据我了解，所有实例都会订阅该事件，因此都会收到它。

这本书没有清楚地解释如何确保只有一个实例处理请求。

我已经查看了重复部分，但它描述了一种模式，该模式解释了如何在服务实例中进行重复数据删除，但不一定针对它们...

每个微服务实例都将使用类似于以下内容的方式进行订阅：

public void Subscribe<T, TH>()
        where T : IntegrationEvent
        where TH : IIntegrationEventHandler<T>
    {
        var eventName = _subsManager.GetEventKey<T>();

        var containsKey = _subsManager.HasSubscriptionsForEvent(eventName);
        if (!containsKey)
        {
            if (!_persistentConnection.IsConnected)
            {
                _persistentConnection.TryConnect();
            }

            using (var channel = _persistentConnection.CreateModel())
            {
                channel.QueueBind(queue: _queueName,
                                    exchange: BROKER_NAME,
                                    routingKey: eventName);
            }
        }

        _subsManager.AddSubscription<T, TH>();
    }

我需要了解如果服务在处理过程中出现故障，相同“类型”微服务的多个微服务实例如何在不丢失消息的情况下进行重复数据删除。

Answer 1

据我了解，所有实例都将订阅该事件并 因此都会收到。

只有一个订阅者实例会处理消息/事件。当您有多个正在运行的服务实例并订阅了同一个订阅时，第一个选择消息的实例会将消息设置为对订阅不可见（称为可见性超时）。如果服务实例能够在给定的时间内处理消息，它将告诉队列删除消息，如果它不能及时处理消息，则消息将重新出现在队列中，以便任何实例再次提取它.

所有标准服务总线（rabbitMQ、SQS、Azure 服务总线等）都提供了开箱即用的此功能。

顺便说一句，我已经阅读了这本书并使用了来自 eShotContainers 的上述代码，它的工作方式与我描述的方式相同。

您还应该研究以下模式 Competing Consumers pattern

希望有帮助！