From 1d4295431368fb47087188ad199091b67e280579 Mon Sep 17 00:00:00 2001 From: liangshiwei Date: Sun, 9 Jan 2022 13:08:03 +0800 Subject: [PATCH 01/11] Batch publish events from outbox to the event bus --- .../PostgreSqlDbContextEventOutbox.cs | 2 +- .../DistributedEvents/DbContextEventOutbox.cs | 11 + .../Azure/AzureDistributedEventBus.cs | 9 +- .../Kafka/KafkaDistributedEventBus.cs | 9 +- .../RabbitMq/RabbitMqDistributedEventBus.cs | 509 ++++++++++-------- .../Rebus/RebusDistributedEventBus.cs | 9 +- .../Distributed/DistributedEventBusBase.cs | 236 ++++---- .../Abp/EventBus/Distributed/IEventOutbox.cs | 2 + .../Distributed/ISupportsEventBoxes.cs | 6 + .../EventBus/Distributed/InboxProcessor.cs | 4 +- .../Abp/EventBus/Distributed/OutboxSender.cs | 21 +- .../MultipleOutgoingEventPublishResult.cs | 14 + .../MongoDbContextEventOutbox.cs | 14 + 13 files changed, 488 insertions(+), 358 deletions(-) create mode 100644 framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/MultipleOutgoingEventPublishResult.cs diff --git a/framework/src/Volo.Abp.EntityFrameworkCore.PostgreSql/Volo/Abp/EntityFrameworkCore/DistributedEvents/PostgreSqlDbContextEventOutbox.cs b/framework/src/Volo.Abp.EntityFrameworkCore.PostgreSql/Volo/Abp/EntityFrameworkCore/DistributedEvents/PostgreSqlDbContextEventOutbox.cs index e5f2e0c4b6..7c5ef12fa7 100644 --- a/framework/src/Volo.Abp.EntityFrameworkCore.PostgreSql/Volo/Abp/EntityFrameworkCore/DistributedEvents/PostgreSqlDbContextEventOutbox.cs +++ b/framework/src/Volo.Abp.EntityFrameworkCore.PostgreSql/Volo/Abp/EntityFrameworkCore/DistributedEvents/PostgreSqlDbContextEventOutbox.cs @@ -13,7 +13,7 @@ public class PostgreSqlDbContextEventOutbox : DbContextEventOutbox : IDbContextEventOutbox ids) + { + var dbContext = (IHasEventOutbox)await DbContextProvider.GetDbContextAsync(); + var outgoingEvents = await dbContext.OutgoingEvents.Where(x => ids.Contains(x.Id)).ToListAsync(); + if (outgoingEvents.Any()) + { + dbContext.RemoveRange(outgoingEvents); + } + } } diff --git a/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs b/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs index 49add5df84..b253164161 100644 --- a/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs +++ b/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs @@ -85,12 +85,17 @@ public class AzureDistributedEventBus : DistributedEventBusBase, ISingletonDepen await TriggerHandlersAsync(eventType, eventData); } - public override async Task PublishFromOutboxAsync(OutgoingEventInfo outgoingEvent, OutboxConfig outboxConfig) + public async override Task PublishFromOutboxAsync(OutgoingEventInfo outgoingEvent, OutboxConfig outboxConfig) { await PublishAsync(outgoingEvent.EventName, outgoingEvent.EventData); } - public override async Task ProcessFromInboxAsync(IncomingEventInfo incomingEvent, InboxConfig inboxConfig) + public override Task PublishManyFromOutboxAsync(IEnumerable outgoingEvents, OutboxConfig outboxConfig) + { + throw new NotImplementedException(); + } + + public async override Task ProcessFromInboxAsync(IncomingEventInfo incomingEvent, InboxConfig inboxConfig) { var eventType = _eventTypes.GetOrDefault(incomingEvent.EventName); if (eventType == null) diff --git a/framework/src/Volo.Abp.EventBus.Kafka/Volo/Abp/EventBus/Kafka/KafkaDistributedEventBus.cs b/framework/src/Volo.Abp.EventBus.Kafka/Volo/Abp/EventBus/Kafka/KafkaDistributedEventBus.cs index ec1901161a..23ae188170 100644 --- a/framework/src/Volo.Abp.EventBus.Kafka/Volo/Abp/EventBus/Kafka/KafkaDistributedEventBus.cs +++ b/framework/src/Volo.Abp.EventBus.Kafka/Volo/Abp/EventBus/Kafka/KafkaDistributedEventBus.cs @@ -162,7 +162,7 @@ public class KafkaDistributedEventBus : DistributedEventBusBase, ISingletonDepen GetOrCreateHandlerFactories(eventType).Locking(factories => factories.Clear()); } - protected override async Task PublishToEventBusAsync(Type eventType, object eventData) + protected async override Task PublishToEventBusAsync(Type eventType, object eventData) { await PublishAsync( eventType, @@ -196,7 +196,12 @@ public class KafkaDistributedEventBus : DistributedEventBusBase, ISingletonDepen ); } - public override async Task ProcessFromInboxAsync( + public override Task PublishManyFromOutboxAsync(IEnumerable outgoingEvents, OutboxConfig outboxConfig) + { + throw new NotImplementedException(); + } + + public async override Task ProcessFromInboxAsync( IncomingEventInfo incomingEvent, InboxConfig inboxConfig) { diff --git a/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs b/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs index 1564cfd213..c65e9a60d1 100644 --- a/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs +++ b/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs @@ -16,225 +16,290 @@ using Volo.Abp.Threading; using Volo.Abp.Timing; using Volo.Abp.Uow; -namespace Volo.Abp.EventBus.RabbitMq; - -/* TODO: How to handle unsubscribe to unbind on RabbitMq (may not be possible for) - */ -[Dependency(ReplaceServices = true)] -[ExposeServices(typeof(IDistributedEventBus), typeof(RabbitMqDistributedEventBus))] -public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDependency +namespace Volo.Abp.EventBus.RabbitMq { - protected AbpRabbitMqEventBusOptions AbpRabbitMqEventBusOptions { get; } - protected IConnectionPool ConnectionPool { get; } - protected IRabbitMqSerializer Serializer { get; } - - //TODO: Accessing to the List may not be thread-safe! - protected ConcurrentDictionary> HandlerFactories { get; } - protected ConcurrentDictionary EventTypes { get; } - protected IRabbitMqMessageConsumerFactory MessageConsumerFactory { get; } - protected IRabbitMqMessageConsumer Consumer { get; private set; } - - public RabbitMqDistributedEventBus( - IOptions options, - IConnectionPool connectionPool, - IRabbitMqSerializer serializer, - IServiceScopeFactory serviceScopeFactory, - IOptions distributedEventBusOptions, - IRabbitMqMessageConsumerFactory messageConsumerFactory, - ICurrentTenant currentTenant, - IUnitOfWorkManager unitOfWorkManager, - IGuidGenerator guidGenerator, - IClock clock) - : base( - serviceScopeFactory, - currentTenant, - unitOfWorkManager, - distributedEventBusOptions, - guidGenerator, - clock) + /* TODO: How to handle unsubscribe to unbind on RabbitMq (may not be possible for) + */ + [Dependency(ReplaceServices = true)] + [ExposeServices(typeof(IDistributedEventBus), typeof(RabbitMqDistributedEventBus))] + public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDependency { - ConnectionPool = connectionPool; - Serializer = serializer; - MessageConsumerFactory = messageConsumerFactory; - AbpRabbitMqEventBusOptions = options.Value; - - HandlerFactories = new ConcurrentDictionary>(); - EventTypes = new ConcurrentDictionary(); - } + protected AbpRabbitMqEventBusOptions AbpRabbitMqEventBusOptions { get; } + protected IConnectionPool ConnectionPool { get; } + protected IRabbitMqSerializer Serializer { get; } + + //TODO: Accessing to the List may not be thread-safe! + protected ConcurrentDictionary> HandlerFactories { get; } + protected ConcurrentDictionary EventTypes { get; } + protected IRabbitMqMessageConsumerFactory MessageConsumerFactory { get; } + protected IRabbitMqMessageConsumer Consumer { get; private set; } + + public RabbitMqDistributedEventBus( + IOptions options, + IConnectionPool connectionPool, + IRabbitMqSerializer serializer, + IServiceScopeFactory serviceScopeFactory, + IOptions distributedEventBusOptions, + IRabbitMqMessageConsumerFactory messageConsumerFactory, + ICurrentTenant currentTenant, + IUnitOfWorkManager unitOfWorkManager, + IGuidGenerator guidGenerator, + IClock clock) + : base( + serviceScopeFactory, + currentTenant, + unitOfWorkManager, + distributedEventBusOptions, + guidGenerator, + clock) + { + ConnectionPool = connectionPool; + Serializer = serializer; + MessageConsumerFactory = messageConsumerFactory; + AbpRabbitMqEventBusOptions = options.Value; - public void Initialize() - { - Consumer = MessageConsumerFactory.Create( - new ExchangeDeclareConfiguration( - AbpRabbitMqEventBusOptions.ExchangeName, - type: "direct", - durable: true - ), - new QueueDeclareConfiguration( - AbpRabbitMqEventBusOptions.ClientName, - durable: true, - exclusive: false, - autoDelete: false - ), - AbpRabbitMqEventBusOptions.ConnectionName - ); - - Consumer.OnMessageReceived(ProcessEventAsync); - - SubscribeHandlers(AbpDistributedEventBusOptions.Handlers); - } + HandlerFactories = new ConcurrentDictionary>(); + EventTypes = new ConcurrentDictionary(); + } - private async Task ProcessEventAsync(IModel channel, BasicDeliverEventArgs ea) - { - var eventName = ea.RoutingKey; - var eventType = EventTypes.GetOrDefault(eventName); - if (eventType == null) + public void Initialize() { - return; + Consumer = MessageConsumerFactory.Create( + new ExchangeDeclareConfiguration( + AbpRabbitMqEventBusOptions.ExchangeName, + type: "direct", + durable: true + ), + new QueueDeclareConfiguration( + AbpRabbitMqEventBusOptions.ClientName, + durable: true, + exclusive: false, + autoDelete: false + ), + AbpRabbitMqEventBusOptions.ConnectionName + ); + + Consumer.OnMessageReceived(ProcessEventAsync); + + SubscribeHandlers(AbpDistributedEventBusOptions.Handlers); } - var eventBytes = ea.Body.ToArray(); + private async Task ProcessEventAsync(IModel channel, BasicDeliverEventArgs ea) + { + var eventName = ea.RoutingKey; + var eventType = EventTypes.GetOrDefault(eventName); + if (eventType == null) + { + return; + } + + var eventBytes = ea.Body.ToArray(); + + if (await AddToInboxAsync(ea.BasicProperties.MessageId, eventName, eventType, eventBytes)) + { + return; + } + + var eventData = Serializer.Deserialize(eventBytes, eventType); + + await TriggerHandlersAsync(eventType, eventData); + } - if (await AddToInboxAsync(ea.BasicProperties.MessageId, eventName, eventType, eventBytes)) + public override IDisposable Subscribe(Type eventType, IEventHandlerFactory factory) { - return; + var handlerFactories = GetOrCreateHandlerFactories(eventType); + + if (factory.IsInFactories(handlerFactories)) + { + return NullDisposable.Instance; + } + + handlerFactories.Add(factory); + + if (handlerFactories.Count == 1) //TODO: Multi-threading! + { + Consumer.BindAsync(EventNameAttribute.GetNameOrDefault(eventType)); + } + + return new EventHandlerFactoryUnregistrar(this, eventType, factory); } - var eventData = Serializer.Deserialize(eventBytes, eventType); + /// + public override void Unsubscribe(Func action) + { + Check.NotNull(action, nameof(action)); - await TriggerHandlersAsync(eventType, eventData); - } + GetOrCreateHandlerFactories(typeof(TEvent)) + .Locking(factories => + { + factories.RemoveAll( + factory => + { + var singleInstanceFactory = factory as SingleInstanceHandlerFactory; + if (singleInstanceFactory == null) + { + return false; + } + + var actionHandler = singleInstanceFactory.HandlerInstance as ActionEventHandler; + if (actionHandler == null) + { + return false; + } + + return actionHandler.Action == action; + }); + }); + } - public override IDisposable Subscribe(Type eventType, IEventHandlerFactory factory) - { - var handlerFactories = GetOrCreateHandlerFactories(eventType); + /// + public override void Unsubscribe(Type eventType, IEventHandler handler) + { + GetOrCreateHandlerFactories(eventType) + .Locking(factories => + { + factories.RemoveAll( + factory => + factory is SingleInstanceHandlerFactory && + (factory as SingleInstanceHandlerFactory).HandlerInstance == handler + ); + }); + } - if (factory.IsInFactories(handlerFactories)) + /// + public override void Unsubscribe(Type eventType, IEventHandlerFactory factory) { - return NullDisposable.Instance; + GetOrCreateHandlerFactories(eventType).Locking(factories => factories.Remove(factory)); } - handlerFactories.Add(factory); + /// + public override void UnsubscribeAll(Type eventType) + { + GetOrCreateHandlerFactories(eventType).Locking(factories => factories.Clear()); + } - if (handlerFactories.Count == 1) //TODO: Multi-threading! + protected async override Task PublishToEventBusAsync(Type eventType, object eventData) { - Consumer.BindAsync(EventNameAttribute.GetNameOrDefault(eventType)); + await PublishAsync(eventType, eventData, null); } - return new EventHandlerFactoryUnregistrar(this, eventType, factory); - } + protected override void AddToUnitOfWork(IUnitOfWork unitOfWork, UnitOfWorkEventRecord eventRecord) + { + unitOfWork.AddOrReplaceDistributedEvent(eventRecord); + } - /// - public override void Unsubscribe(Func action) - { - Check.NotNull(action, nameof(action)); + public override Task PublishFromOutboxAsync( + OutgoingEventInfo outgoingEvent, + OutboxConfig outboxConfig) + { + return PublishAsync(outgoingEvent.EventName, outgoingEvent.EventData, null, eventId: outgoingEvent.Id); + } - GetOrCreateHandlerFactories(typeof(TEvent)) - .Locking(factories => + public async override Task PublishManyFromOutboxAsync( + IEnumerable outgoingEvents, + OutboxConfig outboxConfig) + { + using (var channel = ConnectionPool.Get(AbpRabbitMqEventBusOptions.ConnectionName).CreateModel()) { - factories.RemoveAll( - factory => + var outgoingEventArray = outgoingEvents.ToArray(); + channel.ConfirmSelect(); + + var pendingConfirms = new ConcurrentDictionary(); + var failures = new ConcurrentBag(); + + void CleanPendingConfirms(ulong sequenceNumber, bool multiple, bool ack) + { + if (multiple) { - var singleInstanceFactory = factory as SingleInstanceHandlerFactory; - if (singleInstanceFactory == null) + var confirmed = pendingConfirms.Where(x => x.Key <= sequenceNumber); + foreach (var entry in confirmed) { - return false; + pendingConfirms.TryRemove(entry.Key, out var eventId); + + if (!ack) + { + failures.Add(eventId); + } } + } + else + { + pendingConfirms.TryRemove(sequenceNumber, out var eventId); - var actionHandler = singleInstanceFactory.HandlerInstance as ActionEventHandler; - if (actionHandler == null) + if (!ack) { - return false; + failures.Add(eventId); } + } + } - return actionHandler.Action == action; - }); - }); - } + foreach (var outgoingEvent in outgoingEventArray) + { + pendingConfirms.TryAdd(channel.NextPublishSeqNo, outgoingEvent.Id); + await PublishInternalAsync(channel, outgoingEvent.EventName, outgoingEvent.EventData, null, eventId: outgoingEvent.Id); + } - /// - public override void Unsubscribe(Type eventType, IEventHandler handler) - { - GetOrCreateHandlerFactories(eventType) - .Locking(factories => - { - factories.RemoveAll( - factory => - factory is SingleInstanceHandlerFactory && - (factory as SingleInstanceHandlerFactory).HandlerInstance == handler - ); - }); - } + channel.BasicAcks += (_, ea) => CleanPendingConfirms(ea.DeliveryTag, ea.Multiple, true); + channel.BasicNacks += (_, ea) => CleanPendingConfirms(ea.DeliveryTag, ea.Multiple, false); - /// - public override void Unsubscribe(Type eventType, IEventHandlerFactory factory) - { - GetOrCreateHandlerFactories(eventType).Locking(factories => factories.Remove(factory)); - } - - /// - public override void UnsubscribeAll(Type eventType) - { - GetOrCreateHandlerFactories(eventType).Locking(factories => factories.Clear()); - } + channel.WaitForConfirms(); - protected override async Task PublishToEventBusAsync(Type eventType, object eventData) - { - await PublishAsync(eventType, eventData, null); - } - - protected override void AddToUnitOfWork(IUnitOfWork unitOfWork, UnitOfWorkEventRecord eventRecord) - { - unitOfWork.AddOrReplaceDistributedEvent(eventRecord); - } - - public override Task PublishFromOutboxAsync( - OutgoingEventInfo outgoingEvent, - OutboxConfig outboxConfig) - { - return PublishAsync(outgoingEvent.EventName, outgoingEvent.EventData, null, eventId: outgoingEvent.Id); - } + return new MultipleOutgoingEventPublishResult(outgoingEventArray.Where(x => !failures.Contains(x.Id)).ToList()); + } + } - public override async Task ProcessFromInboxAsync( - IncomingEventInfo incomingEvent, - InboxConfig inboxConfig) - { - var eventType = EventTypes.GetOrDefault(incomingEvent.EventName); - if (eventType == null) + public async override Task ProcessFromInboxAsync( + IncomingEventInfo incomingEvent, + InboxConfig inboxConfig) { - return; + var eventType = EventTypes.GetOrDefault(incomingEvent.EventName); + if (eventType == null) + { + return; + } + + var eventData = Serializer.Deserialize(incomingEvent.EventData, eventType); + var exceptions = new List(); + await TriggerHandlersAsync(eventType, eventData, exceptions, inboxConfig); + if (exceptions.Any()) + { + ThrowOriginalExceptions(eventType, exceptions); + } } - var eventData = Serializer.Deserialize(incomingEvent.EventData, eventType); - var exceptions = new List(); - await TriggerHandlersAsync(eventType, eventData, exceptions, inboxConfig); - if (exceptions.Any()) + protected override byte[] Serialize(object eventData) { - ThrowOriginalExceptions(eventType, exceptions); + return Serializer.Serialize(eventData); } - } - protected override byte[] Serialize(object eventData) - { - return Serializer.Serialize(eventData); - } + public virtual Task PublishAsync(Type eventType, object eventData, IBasicProperties properties, Dictionary headersArguments = null) + { + var eventName = EventNameAttribute.GetNameOrDefault(eventType); + var body = Serializer.Serialize(eventData); - public Task PublishAsync(Type eventType, object eventData, IBasicProperties properties, Dictionary headersArguments = null) - { - var eventName = EventNameAttribute.GetNameOrDefault(eventType); - var body = Serializer.Serialize(eventData); + return PublishAsync(eventName, body, properties, headersArguments); + } - return PublishAsync(eventName, body, properties, headersArguments); - } + protected virtual Task PublishAsync( + string eventName, + byte[] body, + IBasicProperties properties, + Dictionary headersArguments = null, + Guid? eventId = null) + { + using (var channel = ConnectionPool.Get(AbpRabbitMqEventBusOptions.ConnectionName).CreateModel()) + { + return PublishInternalAsync(channel, eventName, body, properties, headersArguments, eventId); + } + } - protected Task PublishAsync( - string eventName, - byte[] body, - IBasicProperties properties, - Dictionary headersArguments = null, - Guid? eventId = null) - { - using (var channel = ConnectionPool.Get(AbpRabbitMqEventBusOptions.ConnectionName).CreateModel()) + protected virtual Task PublishInternalAsync( + IModel channel, + string eventName, + byte[] body, + IBasicProperties properties, + Dictionary headersArguments = null, + Guid? eventId = null) { channel.ExchangeDeclare( AbpRabbitMqEventBusOptions.ExchangeName, @@ -262,68 +327,68 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe basicProperties: properties, body: body ); - } - return Task.CompletedTask; - } - - private void SetEventMessageHeaders(IBasicProperties properties, Dictionary headersArguments) - { - if (headersArguments == null) - { - return; + return Task.CompletedTask; } - properties.Headers ??= new Dictionary(); - - foreach (var header in headersArguments) + private void SetEventMessageHeaders(IBasicProperties properties, Dictionary headersArguments) { - properties.Headers[header.Key] = header.Value; - } - } - - private List GetOrCreateHandlerFactories(Type eventType) - { - return HandlerFactories.GetOrAdd( - eventType, - type => + if (headersArguments == null) { - var eventName = EventNameAttribute.GetNameOrDefault(type); - EventTypes[eventName] = type; - return new List(); + return; } - ); - } - protected override IEnumerable GetHandlerFactories(Type eventType) - { - var handlerFactoryList = new List(); + properties.Headers ??= new Dictionary(); - foreach (var handlerFactory in - HandlerFactories.Where(hf => ShouldTriggerEventForHandler(eventType, hf.Key))) - { - handlerFactoryList.Add( - new EventTypeWithEventHandlerFactories(handlerFactory.Key, handlerFactory.Value)); + foreach (var header in headersArguments) + { + properties.Headers[header.Key] = header.Value; + } } - return handlerFactoryList.ToArray(); - } - - private static bool ShouldTriggerEventForHandler(Type targetEventType, Type handlerEventType) - { - //Should trigger same type - if (handlerEventType == targetEventType) + private List GetOrCreateHandlerFactories(Type eventType) { - return true; + return HandlerFactories.GetOrAdd( + eventType, + type => + { + var eventName = EventNameAttribute.GetNameOrDefault(type); + EventTypes[eventName] = type; + return new List(); + } + ); } - //TODO: Support inheritance? But it does not support on subscription to RabbitMq! - //Should trigger for inherited types - if (handlerEventType.IsAssignableFrom(targetEventType)) + protected override IEnumerable GetHandlerFactories(Type eventType) { - return true; + var handlerFactoryList = new List(); + + foreach (var handlerFactory in + HandlerFactories.Where(hf => ShouldTriggerEventForHandler(eventType, hf.Key))) + { + handlerFactoryList.Add( + new EventTypeWithEventHandlerFactories(handlerFactory.Key, handlerFactory.Value)); + } + + return handlerFactoryList.ToArray(); } - return false; + private static bool ShouldTriggerEventForHandler(Type targetEventType, Type handlerEventType) + { + //Should trigger same type + if (handlerEventType == targetEventType) + { + return true; + } + + //TODO: Support inheritance? But it does not support on subscription to RabbitMq! + //Should trigger for inherited types + if (handlerEventType.IsAssignableFrom(targetEventType)) + { + return true; + } + + return false; + } } } diff --git a/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusDistributedEventBus.cs b/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusDistributedEventBus.cs index 0542a705eb..aad844a4d7 100644 --- a/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusDistributedEventBus.cs +++ b/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusDistributedEventBus.cs @@ -146,7 +146,7 @@ public class RebusDistributedEventBus : DistributedEventBusBase, ISingletonDepen await TriggerHandlersAsync(eventType, eventData); } - protected override async Task PublishToEventBusAsync(Type eventType, object eventData) + protected async override Task PublishToEventBusAsync(Type eventType, object eventData) { await AbpRebusEventBusOptions.Publish(Rebus, eventType, eventData); } @@ -210,7 +210,12 @@ public class RebusDistributedEventBus : DistributedEventBusBase, ISingletonDepen return PublishToEventBusAsync(eventType, eventData); } - public override async Task ProcessFromInboxAsync( + public override Task PublishManyFromOutboxAsync(IEnumerable outgoingEvents, OutboxConfig outboxConfig) + { + throw new NotImplementedException(); + } + + public async override Task ProcessFromInboxAsync( IncomingEventInfo incomingEvent, InboxConfig inboxConfig) { diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/DistributedEventBusBase.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/DistributedEventBusBase.cs index dee9d1f7d3..178dea1340 100644 --- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/DistributedEventBusBase.cs +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/DistributedEventBusBase.cs @@ -8,156 +8,162 @@ using Volo.Abp.MultiTenancy; using Volo.Abp.Timing; using Volo.Abp.Uow; -namespace Volo.Abp.EventBus.Distributed; - -public abstract class DistributedEventBusBase : EventBusBase, IDistributedEventBus, ISupportsEventBoxes +namespace Volo.Abp.EventBus.Distributed { - protected IGuidGenerator GuidGenerator { get; } - protected IClock Clock { get; } - protected AbpDistributedEventBusOptions AbpDistributedEventBusOptions { get; } - - protected DistributedEventBusBase( - IServiceScopeFactory serviceScopeFactory, - ICurrentTenant currentTenant, - IUnitOfWorkManager unitOfWorkManager, - IOptions abpDistributedEventBusOptions, - IGuidGenerator guidGenerator, - IClock clock - ) : base( - serviceScopeFactory, - currentTenant, - unitOfWorkManager) - { - GuidGenerator = guidGenerator; - Clock = clock; - AbpDistributedEventBusOptions = abpDistributedEventBusOptions.Value; - } - - public IDisposable Subscribe(IDistributedEventHandler handler) where TEvent : class + public abstract class DistributedEventBusBase : EventBusBase, IDistributedEventBus, ISupportsEventBoxes { - return Subscribe(typeof(TEvent), handler); - } + protected IGuidGenerator GuidGenerator { get; } + protected IClock Clock { get; } + protected AbpDistributedEventBusOptions AbpDistributedEventBusOptions { get; } + + protected DistributedEventBusBase( + IServiceScopeFactory serviceScopeFactory, + ICurrentTenant currentTenant, + IUnitOfWorkManager unitOfWorkManager, + IOptions abpDistributedEventBusOptions, + IGuidGenerator guidGenerator, + IClock clock + ) : base( + serviceScopeFactory, + currentTenant, + unitOfWorkManager) + { + GuidGenerator = guidGenerator; + Clock = clock; + AbpDistributedEventBusOptions = abpDistributedEventBusOptions.Value; + } - public override Task PublishAsync(Type eventType, object eventData, bool onUnitOfWorkComplete = true) - { - return PublishAsync(eventType, eventData, onUnitOfWorkComplete, useOutbox: true); - } + public IDisposable Subscribe(IDistributedEventHandler handler) where TEvent : class + { + return Subscribe(typeof(TEvent), handler); + } - public Task PublishAsync( - TEvent eventData, - bool onUnitOfWorkComplete = true, - bool useOutbox = true) - where TEvent : class - { - return PublishAsync(typeof(TEvent), eventData, onUnitOfWorkComplete, useOutbox); - } + public override Task PublishAsync(Type eventType, object eventData, bool onUnitOfWorkComplete = true) + { + return PublishAsync(eventType, eventData, onUnitOfWorkComplete, useOutbox: true); + } - public async Task PublishAsync( - Type eventType, - object eventData, - bool onUnitOfWorkComplete = true, - bool useOutbox = true) - { - if (onUnitOfWorkComplete && UnitOfWorkManager.Current != null) + public Task PublishAsync( + TEvent eventData, + bool onUnitOfWorkComplete = true, + bool useOutbox = true) + where TEvent : class { - AddToUnitOfWork( - UnitOfWorkManager.Current, - new UnitOfWorkEventRecord(eventType, eventData, EventOrderGenerator.GetNext(), useOutbox) - ); - return; + return PublishAsync(typeof(TEvent), eventData, onUnitOfWorkComplete, useOutbox); } - if (useOutbox) + public async Task PublishAsync( + Type eventType, + object eventData, + bool onUnitOfWorkComplete = true, + bool useOutbox = true) { - if (await AddToOutboxAsync(eventType, eventData)) + if (onUnitOfWorkComplete && UnitOfWorkManager.Current != null) { + AddToUnitOfWork( + UnitOfWorkManager.Current, + new UnitOfWorkEventRecord(eventType, eventData, EventOrderGenerator.GetNext(), useOutbox) + ); return; } - } - await PublishToEventBusAsync(eventType, eventData); - } + if (useOutbox) + { + if (await AddToOutboxAsync(eventType, eventData)) + { + return; + } + } - public abstract Task PublishFromOutboxAsync( - OutgoingEventInfo outgoingEvent, - OutboxConfig outboxConfig - ); + await PublishToEventBusAsync(eventType, eventData); + } - public abstract Task ProcessFromInboxAsync( - IncomingEventInfo incomingEvent, - InboxConfig inboxConfig); + public abstract Task PublishFromOutboxAsync( + OutgoingEventInfo outgoingEvent, + OutboxConfig outboxConfig + ); - private async Task AddToOutboxAsync(Type eventType, object eventData) - { - var unitOfWork = UnitOfWorkManager.Current; - if (unitOfWork == null) - { - return false; - } + public abstract Task PublishManyFromOutboxAsync( + IEnumerable outgoingEvents, + OutboxConfig outboxConfig + ); + + public abstract Task ProcessFromInboxAsync( + IncomingEventInfo incomingEvent, + InboxConfig inboxConfig); - foreach (var outboxConfig in AbpDistributedEventBusOptions.Outboxes.Values) + private async Task AddToOutboxAsync(Type eventType, object eventData) { - if (outboxConfig.Selector == null || outboxConfig.Selector(eventType)) + var unitOfWork = UnitOfWorkManager.Current; + if (unitOfWork == null) { - var eventOutbox = (IEventOutbox)unitOfWork.ServiceProvider.GetRequiredService(outboxConfig.ImplementationType); - var eventName = EventNameAttribute.GetNameOrDefault(eventType); - await eventOutbox.EnqueueAsync( - new OutgoingEventInfo( - GuidGenerator.Create(), - eventName, - Serialize(eventData), - Clock.Now - ) - ); - return true; + return false; } - } - return false; - } + foreach (var outboxConfig in AbpDistributedEventBusOptions.Outboxes.Values) + { + if (outboxConfig.Selector == null || outboxConfig.Selector(eventType)) + { + var eventOutbox = (IEventOutbox)unitOfWork.ServiceProvider.GetRequiredService(outboxConfig.ImplementationType); + var eventName = EventNameAttribute.GetNameOrDefault(eventType); + await eventOutbox.EnqueueAsync( + new OutgoingEventInfo( + GuidGenerator.Create(), + eventName, + Serialize(eventData), + Clock.Now + ) + ); + return true; + } + } - protected async Task AddToInboxAsync( - string messageId, - string eventName, - Type eventType, - byte[] eventBytes) - { - if (AbpDistributedEventBusOptions.Inboxes.Count <= 0) - { return false; } - using (var scope = ServiceScopeFactory.CreateScope()) + protected async Task AddToInboxAsync( + string messageId, + string eventName, + Type eventType, + byte[] eventBytes) { - foreach (var inboxConfig in AbpDistributedEventBusOptions.Inboxes.Values) + if (AbpDistributedEventBusOptions.Inboxes.Count <= 0) { - if (inboxConfig.EventSelector == null || inboxConfig.EventSelector(eventType)) - { - var eventInbox = (IEventInbox)scope.ServiceProvider.GetRequiredService(inboxConfig.ImplementationType); + return false; + } - if (!messageId.IsNullOrEmpty()) + using (var scope = ServiceScopeFactory.CreateScope()) + { + foreach (var inboxConfig in AbpDistributedEventBusOptions.Inboxes.Values) + { + if (inboxConfig.EventSelector == null || inboxConfig.EventSelector(eventType)) { - if (await eventInbox.ExistsByMessageIdAsync(messageId)) + var eventInbox = (IEventInbox) scope.ServiceProvider.GetRequiredService(inboxConfig.ImplementationType); + + if (!messageId.IsNullOrEmpty()) { - continue; + if (await eventInbox.ExistsByMessageIdAsync(messageId)) + { + continue; + } } - } - await eventInbox.EnqueueAsync( - new IncomingEventInfo( - GuidGenerator.Create(), - messageId, - eventName, - eventBytes, - Clock.Now - ) - ); + await eventInbox.EnqueueAsync( + new IncomingEventInfo( + GuidGenerator.Create(), + messageId, + eventName, + eventBytes, + Clock.Now + ) + ); + } } } + + return true; } - return true; + protected abstract byte[] Serialize(object eventData); } - - protected abstract byte[] Serialize(object eventData); } diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/IEventOutbox.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/IEventOutbox.cs index 6ecefcf002..018747945c 100644 --- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/IEventOutbox.cs +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/IEventOutbox.cs @@ -12,4 +12,6 @@ public interface IEventOutbox Task> GetWaitingEventsAsync(int maxCount, CancellationToken cancellationToken = default); Task DeleteAsync(Guid id); + + Task DeleteManyAsync(IEnumerable ids); } diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/ISupportsEventBoxes.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/ISupportsEventBoxes.cs index 37405fb58a..7c23be5ecd 100644 --- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/ISupportsEventBoxes.cs +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/ISupportsEventBoxes.cs @@ -1,3 +1,4 @@ +using System.Collections.Generic; using System.Threading.Tasks; namespace Volo.Abp.EventBus.Distributed; @@ -9,6 +10,11 @@ public interface ISupportsEventBoxes OutboxConfig outboxConfig ); + Task PublishManyFromOutboxAsync( + IEnumerable outgoingEvents, + OutboxConfig outboxConfig + ); + Task ProcessFromInboxAsync( IncomingEventInfo incomingEvent, InboxConfig inboxConfig diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/InboxProcessor.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/InboxProcessor.cs index 50c1a4e009..0d22245d13 100644 --- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/InboxProcessor.cs +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/InboxProcessor.cs @@ -98,7 +98,7 @@ public class InboxProcessor : IInboxProcessor, ITransientDependency break; } - Logger.LogInformation($"Found {waitingEvents.Count} events in the inbox."); + //Logger.LogInformation($"Found {waitingEvents.Count} events in the inbox."); foreach (var waitingEvent in waitingEvents) { @@ -113,7 +113,7 @@ public class InboxProcessor : IInboxProcessor, ITransientDependency await uow.CompleteAsync(); } - Logger.LogInformation($"Processed the incoming event with id = {waitingEvent.Id:N}"); + //Logger.LogInformation($"Processed the incoming event with id = {waitingEvent.Id:N}"); } } } diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/OutboxSender.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/OutboxSender.cs index 6ed60ba34c..d42110e4ea 100644 --- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/OutboxSender.cs +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/OutboxSender.cs @@ -1,4 +1,6 @@ using System; +using System.Diagnostics; +using System.Linq; using System.Threading; using System.Threading.Tasks; using Microsoft.Extensions.DependencyInjection; @@ -76,25 +78,20 @@ public class OutboxSender : IOutboxSender, ITransientDependency while (true) { var waitingEvents = await Outbox.GetWaitingEventsAsync(EventBusBoxesOptions.OutboxWaitingEventMaxCount, StoppingToken); - if (waitingEvents.Count <= 0) + if (waitingEvents.Count < 1000) { break; } Logger.LogInformation($"Found {waitingEvents.Count} events in the outbox."); - foreach (var waitingEvent in waitingEvents) - { - await DistributedEventBus - .AsSupportsEventBoxes() - .PublishFromOutboxAsync( - waitingEvent, - OutboxConfig - ); + var result = await DistributedEventBus + .AsSupportsEventBoxes() + .PublishManyFromOutboxAsync(waitingEvents, OutboxConfig); - await Outbox.DeleteAsync(waitingEvent.Id); - Logger.LogInformation($"Sent the event to the message broker with id = {waitingEvent.Id:N}"); - } + await Outbox.DeleteManyAsync(result.PublishedOutgoingEvents.Select(x => x.Id).ToArray()); + + Logger.LogInformation($"Sent {result.PublishedOutgoingEvents.Count} events to message broker"); } } else diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/MultipleOutgoingEventPublishResult.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/MultipleOutgoingEventPublishResult.cs new file mode 100644 index 0000000000..515dec1ae6 --- /dev/null +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/MultipleOutgoingEventPublishResult.cs @@ -0,0 +1,14 @@ +using System.Collections.Generic; +using Volo.Abp.EventBus.Distributed; + +namespace Volo.Abp.EventBus; + +public class MultipleOutgoingEventPublishResult +{ + public IReadOnlyList PublishedOutgoingEvents { get; } + + public MultipleOutgoingEventPublishResult(IReadOnlyList outgoingEvents) + { + PublishedOutgoingEvents = outgoingEvents; + } +} diff --git a/framework/src/Volo.Abp.MongoDB/Volo/Abp/MongoDB/DistributedEvents/MongoDbContextEventOutbox.cs b/framework/src/Volo.Abp.MongoDB/Volo/Abp/MongoDB/DistributedEvents/MongoDbContextEventOutbox.cs index 1f3ae999b0..fae0709953 100644 --- a/framework/src/Volo.Abp.MongoDB/Volo/Abp/MongoDB/DistributedEvents/MongoDbContextEventOutbox.cs +++ b/framework/src/Volo.Abp.MongoDB/Volo/Abp/MongoDB/DistributedEvents/MongoDbContextEventOutbox.cs @@ -68,4 +68,18 @@ public class MongoDbContextEventOutbox : IMongoDbContextEventOu await dbContext.OutgoingEvents.DeleteOneAsync(x => x.Id.Equals(id)); } } + + [UnitOfWork] + public async Task DeleteManyAsync(IEnumerable ids) + { + var dbContext = (IHasEventOutbox)await MongoDbContextProvider.GetDbContextAsync(); + if (dbContext.SessionHandle != null) + { + await dbContext.OutgoingEvents.DeleteManyAsync(dbContext.SessionHandle, x => ids.Contains(x.Id)); + } + else + { + await dbContext.OutgoingEvents.DeleteManyAsync(x => ids.Contains(x.Id)); + } + } } From 5ee59194637c187ecfee34aab7163c801e182044 Mon Sep 17 00:00:00 2001 From: liangshiwei Date: Sun, 9 Jan 2022 20:24:39 +0800 Subject: [PATCH 02/11] Batch publish events from outbox to the kafka event bus --- .../Kafka/KafkaDistributedEventBus.cs | 44 ++++++++++++++----- .../Abp/EventBus/Kafka/MessageExtensions.cs | 18 ++++++++ .../EventBus/Distributed/InboxProcessor.cs | 4 +- .../Abp/EventBus/Distributed/OutboxSender.cs | 2 +- 4 files changed, 55 insertions(+), 13 deletions(-) create mode 100644 framework/src/Volo.Abp.EventBus.Kafka/Volo/Abp/EventBus/Kafka/MessageExtensions.cs diff --git a/framework/src/Volo.Abp.EventBus.Kafka/Volo/Abp/EventBus/Kafka/KafkaDistributedEventBus.cs b/framework/src/Volo.Abp.EventBus.Kafka/Volo/Abp/EventBus/Kafka/KafkaDistributedEventBus.cs index 23ae188170..aef7662095 100644 --- a/framework/src/Volo.Abp.EventBus.Kafka/Volo/Abp/EventBus/Kafka/KafkaDistributedEventBus.cs +++ b/framework/src/Volo.Abp.EventBus.Kafka/Volo/Abp/EventBus/Kafka/KafkaDistributedEventBus.cs @@ -77,12 +77,7 @@ public class KafkaDistributedEventBus : DistributedEventBusBase, ISingletonDepen return; } - string messageId = null; - - if (message.Headers.TryGetLastBytes("messageId", out var messageIdBytes)) - { - messageId = System.Text.Encoding.UTF8.GetString(messageIdBytes); - } + var messageId = message.GetMessageId(); if (await AddToInboxAsync(messageId, eventName, eventType, message.Value)) { @@ -196,9 +191,38 @@ public class KafkaDistributedEventBus : DistributedEventBusBase, ISingletonDepen ); } - public override Task PublishManyFromOutboxAsync(IEnumerable outgoingEvents, OutboxConfig outboxConfig) + public async override Task PublishManyFromOutboxAsync(IEnumerable outgoingEvents, OutboxConfig outboxConfig) { - throw new NotImplementedException(); + var pendingConfirms = new ConcurrentDictionary(); + var outgoingEventArray = outgoingEvents.ToArray(); + + var tasks = new List(); + foreach (var outgoingEvent in outgoingEventArray) + { + var messageId = outgoingEvent.Id.ToString("N"); + pendingConfirms.TryAdd(messageId, outgoingEvent.Id); + + var task = PublishAsync( + AbpKafkaEventBusOptions.TopicName, + outgoingEvent.EventName, + outgoingEvent.EventData, + new Headers { { "messageId", System.Text.Encoding.UTF8.GetBytes(messageId)} }, + null + ); + + tasks.Add(task.ContinueWith(t => + { + if (!t.IsFaulted) + { + var message = t.Result.Message; + pendingConfirms.TryRemove(message.GetMessageId(), out _); + } + })); + } + + await Task.WhenAll(tasks); + + return new MultipleOutgoingEventPublishResult(outgoingEventArray.Where(x => !pendingConfirms.Select(p => p.Value).Contains(x.Id)).ToList()); } public async override Task ProcessFromInboxAsync( @@ -244,13 +268,13 @@ public class KafkaDistributedEventBus : DistributedEventBusBase, ISingletonDepen return PublishAsync(topicName, eventName, body, headers, headersArguments); } - private async Task PublishAsync(string topicName, string eventName, byte[] body, Headers headers, Dictionary headersArguments) + private Task> PublishAsync(string topicName, string eventName, byte[] body, Headers headers, Dictionary headersArguments) { var producer = ProducerPool.Get(AbpKafkaEventBusOptions.ConnectionName); SetEventMessageHeaders(headers, headersArguments); - await producer.ProduceAsync( + return producer.ProduceAsync( topicName, new Message { diff --git a/framework/src/Volo.Abp.EventBus.Kafka/Volo/Abp/EventBus/Kafka/MessageExtensions.cs b/framework/src/Volo.Abp.EventBus.Kafka/Volo/Abp/EventBus/Kafka/MessageExtensions.cs new file mode 100644 index 0000000000..17a80ec87c --- /dev/null +++ b/framework/src/Volo.Abp.EventBus.Kafka/Volo/Abp/EventBus/Kafka/MessageExtensions.cs @@ -0,0 +1,18 @@ +using Confluent.Kafka; + +namespace Volo.Abp.EventBus.Kafka; + +public static class MessageExtensions +{ + public static string GetMessageId(this Message message) + { + string messageId = null; + + if (message.Headers.TryGetLastBytes("messageId", out var messageIdBytes)) + { + messageId = System.Text.Encoding.UTF8.GetString(messageIdBytes); + } + + return messageId; + } +} diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/InboxProcessor.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/InboxProcessor.cs index 0d22245d13..50c1a4e009 100644 --- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/InboxProcessor.cs +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/InboxProcessor.cs @@ -98,7 +98,7 @@ public class InboxProcessor : IInboxProcessor, ITransientDependency break; } - //Logger.LogInformation($"Found {waitingEvents.Count} events in the inbox."); + Logger.LogInformation($"Found {waitingEvents.Count} events in the inbox."); foreach (var waitingEvent in waitingEvents) { @@ -113,7 +113,7 @@ public class InboxProcessor : IInboxProcessor, ITransientDependency await uow.CompleteAsync(); } - //Logger.LogInformation($"Processed the incoming event with id = {waitingEvent.Id:N}"); + Logger.LogInformation($"Processed the incoming event with id = {waitingEvent.Id:N}"); } } } diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/OutboxSender.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/OutboxSender.cs index d42110e4ea..17dec0a810 100644 --- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/OutboxSender.cs +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/OutboxSender.cs @@ -78,7 +78,7 @@ public class OutboxSender : IOutboxSender, ITransientDependency while (true) { var waitingEvents = await Outbox.GetWaitingEventsAsync(EventBusBoxesOptions.OutboxWaitingEventMaxCount, StoppingToken); - if (waitingEvents.Count < 1000) + if (waitingEvents.Count <= 0) { break; } From f36c44fc78203c4f520cfce62d99fce82ecb24c4 Mon Sep 17 00:00:00 2001 From: liangshiwei Date: Sun, 9 Jan 2022 20:50:17 +0800 Subject: [PATCH 03/11] Batch publish events from outbox to the azure event bus --- .../Azure/AzureDistributedEventBus.cs | 35 +++++++++++++++++-- 1 file changed, 32 insertions(+), 3 deletions(-) diff --git a/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs b/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs index b253164161..a253a8ba96 100644 --- a/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs +++ b/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs @@ -90,9 +90,38 @@ public class AzureDistributedEventBus : DistributedEventBusBase, ISingletonDepen await PublishAsync(outgoingEvent.EventName, outgoingEvent.EventData); } - public override Task PublishManyFromOutboxAsync(IEnumerable outgoingEvents, OutboxConfig outboxConfig) + public async override Task PublishManyFromOutboxAsync(IEnumerable outgoingEvents, OutboxConfig outboxConfig) { - throw new NotImplementedException(); + var outgoingEventArray = outgoingEvents.ToArray(); + var failures = new List(); + + var publisher = await _publisherPool.GetAsync( + _options.TopicName, + _options.ConnectionName); + + using var messageBatch = await publisher.CreateMessageBatchAsync(); + + var failed = false; + foreach (var outgoingEvent in outgoingEventArray) + { + if (failed) + { + failures.Add(outgoingEvent.Id); + continue; + } + + var body = _serializer.Serialize(outgoingEvent.EventData); + + if (!messageBatch.TryAddMessage(new ServiceBusMessage(body) { Subject = outgoingEvent.EventName })) + { + failed = true; + failures.Add(outgoingEvent.Id); + } + } + + await publisher.SendMessagesAsync(messageBatch); + + return new MultipleOutgoingEventPublishResult(outgoingEventArray.Where(x => !failures.Contains(x.Id)).ToList()); } public async override Task ProcessFromInboxAsync(IncomingEventInfo incomingEvent, InboxConfig inboxConfig) @@ -183,7 +212,7 @@ public class AzureDistributedEventBus : DistributedEventBusBase, ISingletonDepen .Locking(factories => factories.Clear()); } - protected override async Task PublishToEventBusAsync(Type eventType, object eventData) + protected async override Task PublishToEventBusAsync(Type eventType, object eventData) { await PublishAsync(EventNameAttribute.GetNameOrDefault(eventType), eventData); } From 9ec54780d475ca56c2c6086039a9bb4d42a4371c Mon Sep 17 00:00:00 2001 From: liangshiwei Date: Sun, 9 Jan 2022 21:18:48 +0800 Subject: [PATCH 04/11] Batch publish events from outbox to the rebus event bus --- .../EventBus/Rebus/AbpRebusEventBusOptions.cs | 13 +--------- .../Rebus/RebusDistributedEventBus.cs | 24 ++++++++++++++++--- 2 files changed, 22 insertions(+), 15 deletions(-) diff --git a/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/AbpRebusEventBusOptions.cs b/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/AbpRebusEventBusOptions.cs index 5238cdd8bb..4c39b9bf1d 100644 --- a/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/AbpRebusEventBusOptions.cs +++ b/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/AbpRebusEventBusOptions.cs @@ -20,24 +20,13 @@ public class AbpRebusEventBusOptions } private Action _configurer; - [NotNull] - public Func Publish { - get => _publish; - set => _publish = Check.NotNull(value, nameof(value)); - } - private Func _publish; + public Func Publish { get; set; } public AbpRebusEventBusOptions() { - _publish = DefaultPublish; _configurer = DefaultConfigure; } - private async Task DefaultPublish(IBus bus, Type eventType, object eventData) - { - await bus.Advanced.Routing.Send(InputQueueName, eventData); - } - private void DefaultConfigure(RebusConfigurer configure) { configure.Subscriptions(s => s.StoreInMemory()); diff --git a/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusDistributedEventBus.cs b/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusDistributedEventBus.cs index aad844a4d7..fc6a0667af 100644 --- a/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusDistributedEventBus.cs +++ b/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusDistributedEventBus.cs @@ -148,7 +148,18 @@ public class RebusDistributedEventBus : DistributedEventBusBase, ISingletonDepen protected async override Task PublishToEventBusAsync(Type eventType, object eventData) { - await AbpRebusEventBusOptions.Publish(Rebus, eventType, eventData); + await PublishAsync(eventType, eventData); + } + + protected virtual async Task PublishAsync(Type eventType, object eventData) + { + if (AbpRebusEventBusOptions.Publish != null) + { + await AbpRebusEventBusOptions.Publish(Rebus, eventType, eventData); + return; + } + + await Rebus.Advanced.Routing.Send(AbpRebusEventBusOptions.InputQueueName, eventData); } protected override void AddToUnitOfWork(IUnitOfWork unitOfWork, UnitOfWorkEventRecord eventRecord) @@ -210,9 +221,16 @@ public class RebusDistributedEventBus : DistributedEventBusBase, ISingletonDepen return PublishToEventBusAsync(eventType, eventData); } - public override Task PublishManyFromOutboxAsync(IEnumerable outgoingEvents, OutboxConfig outboxConfig) + public async override Task PublishManyFromOutboxAsync(IEnumerable outgoingEvents, OutboxConfig outboxConfig) { - throw new NotImplementedException(); + var outgoingEventArray = outgoingEvents.ToArray(); + + foreach (var outgoingEvent in outgoingEventArray) + { + await PublishFromOutboxAsync(outgoingEvent, outboxConfig); + } + + return new MultipleOutgoingEventPublishResult(outgoingEventArray); } public async override Task ProcessFromInboxAsync( From 325f82a3d7a7448f7e7b3b73438f965810b93500 Mon Sep 17 00:00:00 2001 From: liangshiwei Date: Mon, 10 Jan 2022 17:11:06 +0800 Subject: [PATCH 05/11] Update AzureDistributedEventBus --- .../Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs b/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs index ffcf6df8c7..5e91a224c5 100644 --- a/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs +++ b/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs @@ -110,9 +110,7 @@ public class AzureDistributedEventBus : DistributedEventBusBase, ISingletonDepen continue; } - var body = _serializer.Serialize(outgoingEvent.EventData); - - if (!messageBatch.TryAddMessage(new ServiceBusMessage(body) { Subject = outgoingEvent.EventName })) + if (!messageBatch.TryAddMessage(new ServiceBusMessage(outgoingEvent.EventData) { Subject = outgoingEvent.EventName })) { failed = true; failures.Add(outgoingEvent.Id); From 3364e7e2273b395b87a5395513a3b5585ac5ddf9 Mon Sep 17 00:00:00 2001 From: liangshiwei Date: Fri, 14 Jan 2022 14:28:30 +0800 Subject: [PATCH 06/11] Update RabbitMqDistributedEventBus.cs --- .../Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs b/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs index a9cfb4dc23..a9d6a42207 100644 --- a/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs +++ b/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs @@ -238,7 +238,7 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe foreach (var outgoingEvent in outgoingEventArray) { pendingConfirms.TryAdd(channel.NextPublishSeqNo, outgoingEvent.Id); - await PublishInternalAsync(channel, outgoingEvent.EventName, outgoingEvent.EventData, null, + await PublishAsync(channel, outgoingEvent.EventName, outgoingEvent.EventData, null, eventId: outgoingEvent.Id); } @@ -293,11 +293,11 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe { using (var channel = ConnectionPool.Get(AbpRabbitMqEventBusOptions.ConnectionName).CreateModel()) { - return PublishInternalAsync(channel, eventName, body, properties, headersArguments, eventId); + return PublishAsync(channel, eventName, body, properties, headersArguments, eventId); } } - protected virtual Task PublishInternalAsync( + protected virtual Task PublishAsync( IModel channel, string eventName, byte[] body, From 0c45dccd54f9c70b6fb2008df81b22655a5f2d20 Mon Sep 17 00:00:00 2001 From: liangshiwei Date: Tue, 25 Jan 2022 14:17:09 +0800 Subject: [PATCH 07/11] Update Rebus --- .../Abp/EventBus/Rebus/RebusDistributedEventBus.cs | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusDistributedEventBus.cs b/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusDistributedEventBus.cs index 81611e180a..a5cac31235 100644 --- a/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusDistributedEventBus.cs +++ b/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusDistributedEventBus.cs @@ -7,6 +7,7 @@ using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Options; using Rebus.Bus; using Rebus.Pipeline; +using Rebus.Transport; using Volo.Abp.DependencyInjection; using Volo.Abp.EventBus.Distributed; using Volo.Abp.Guids; @@ -227,9 +228,14 @@ public class RebusDistributedEventBus : DistributedEventBusBase, ISingletonDepen { var outgoingEventArray = outgoingEvents.ToArray(); - foreach (var outgoingEvent in outgoingEventArray) + using (var scope = new RebusTransactionScope()) { - await PublishFromOutboxAsync(outgoingEvent, outboxConfig); + foreach (var outgoingEvent in outgoingEventArray) + { + await PublishFromOutboxAsync(outgoingEvent, outboxConfig); + } + + await scope.CompleteAsync(); } return new MultipleOutgoingEventPublishResult(outgoingEventArray); From 282d9ada01463503afbfce2bcbe34f68b9cb48db Mon Sep 17 00:00:00 2001 From: liangshiwei Date: Tue, 25 Jan 2022 14:20:56 +0800 Subject: [PATCH 08/11] Update RabbitMqMessageConsumer.cs --- .../Volo/Abp/RabbitMQ/RabbitMqMessageConsumer.cs | 8 -------- 1 file changed, 8 deletions(-) diff --git a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/RabbitMqMessageConsumer.cs b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/RabbitMqMessageConsumer.cs index 049e8da712..8ce184295a 100644 --- a/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/RabbitMqMessageConsumer.cs +++ b/framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/RabbitMqMessageConsumer.cs @@ -176,14 +176,6 @@ public class RabbitMqMessageConsumer : IRabbitMqMessageConsumer, ITransientDepen } catch (Exception ex) { - if (ex is OperationInterruptedException operationInterruptedException && - operationInterruptedException.ShutdownReason.ReplyCode == 406 && - operationInterruptedException.Message.Contains("arg 'x-dead-letter-exchange'")) - { - Logger.LogException(ex, LogLevel.Warning); - await ExceptionNotifier.NotifyAsync(ex, logLevel: LogLevel.Warning); - } - Logger.LogException(ex, LogLevel.Warning); await ExceptionNotifier.NotifyAsync(ex, logLevel: LogLevel.Warning); } From 7221600e75719a494447953b789e79539fb7f10e Mon Sep 17 00:00:00 2001 From: liangshiwei Date: Thu, 17 Mar 2022 21:55:33 +0800 Subject: [PATCH 09/11] Refactor --- .../Azure/AzureDistributedEventBus.cs | 15 +--- .../Kafka/KafkaDistributedEventBus.cs | 76 ++++++++++++------- .../RabbitMq/RabbitMqDistributedEventBus.cs | 69 +++++++---------- .../Rebus/RebusDistributedEventBus.cs | 8 +- .../Distributed/AbpEventBusBoxesOptions.cs | 6 ++ .../Distributed/DistributedEventBusBase.cs | 2 +- .../Distributed/ISupportsEventBoxes.cs | 2 +- .../Abp/EventBus/Distributed/OutboxSender.cs | 46 +++++++++-- .../MultipleOutgoingEventPublishResult.cs | 14 ---- .../Volo/Abp/Kafka/ConsumerPool.cs | 1 + .../Volo/Abp/Kafka/KafkaMessageConsumer.cs | 21 +++-- .../Volo/Abp/Kafka/ProducerPool.cs | 31 +++++--- 12 files changed, 161 insertions(+), 130 deletions(-) delete mode 100644 framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/MultipleOutgoingEventPublishResult.cs diff --git a/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs b/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs index add6670bd4..fa6e155081 100644 --- a/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs +++ b/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs @@ -92,10 +92,9 @@ public class AzureDistributedEventBus : DistributedEventBusBase, ISingletonDepen await PublishAsync(outgoingEvent.EventName, outgoingEvent.EventData, outgoingEvent.Id); } - public async override Task PublishManyFromOutboxAsync(IEnumerable outgoingEvents, OutboxConfig outboxConfig) + public async override Task PublishManyFromOutboxAsync(IEnumerable outgoingEvents, OutboxConfig outboxConfig) { var outgoingEventArray = outgoingEvents.ToArray(); - var failures = new List(); var publisher = await _publisherPool.GetAsync( _options.TopicName, @@ -103,25 +102,15 @@ public class AzureDistributedEventBus : DistributedEventBusBase, ISingletonDepen using var messageBatch = await publisher.CreateMessageBatchAsync(); - var failed = false; foreach (var outgoingEvent in outgoingEventArray) { - if (failed) - { - failures.Add(outgoingEvent.Id); - continue; - } - if (!messageBatch.TryAddMessage(new ServiceBusMessage(outgoingEvent.EventData) { Subject = outgoingEvent.EventName })) { - failed = true; - failures.Add(outgoingEvent.Id); + throw new AbpException("The message is too large to fit in the batch. Set AbpEventBusBoxesOptions.OutboxWaitingEventMaxCount to reduce the number"); } } await publisher.SendMessagesAsync(messageBatch); - - return new MultipleOutgoingEventPublishResult(outgoingEventArray.Where(x => !failures.Contains(x.Id)).ToList()); } public async override Task ProcessFromInboxAsync(IncomingEventInfo incomingEvent, InboxConfig inboxConfig) diff --git a/framework/src/Volo.Abp.EventBus.Kafka/Volo/Abp/EventBus/Kafka/KafkaDistributedEventBus.cs b/framework/src/Volo.Abp.EventBus.Kafka/Volo/Abp/EventBus/Kafka/KafkaDistributedEventBus.cs index e271ad502c..4dba728c51 100644 --- a/framework/src/Volo.Abp.EventBus.Kafka/Volo/Abp/EventBus/Kafka/KafkaDistributedEventBus.cs +++ b/framework/src/Volo.Abp.EventBus.Kafka/Volo/Abp/EventBus/Kafka/KafkaDistributedEventBus.cs @@ -193,38 +193,40 @@ public class KafkaDistributedEventBus : DistributedEventBusBase, ISingletonDepen ); } - public async override Task PublishManyFromOutboxAsync(IEnumerable outgoingEvents, OutboxConfig outboxConfig) + public override Task PublishManyFromOutboxAsync(IEnumerable outgoingEvents, OutboxConfig outboxConfig) { - var pendingConfirms = new ConcurrentDictionary(); + var producer = ProducerPool.Get(); var outgoingEventArray = outgoingEvents.ToArray(); - - var tasks = new List(); - foreach (var outgoingEvent in outgoingEventArray) + producer.BeginTransaction(); + try { - var messageId = outgoingEvent.Id.ToString("N"); - pendingConfirms.TryAdd(messageId, outgoingEvent.Id); - - var task = PublishAsync( - AbpKafkaEventBusOptions.TopicName, - outgoingEvent.EventName, - outgoingEvent.EventData, - new Headers { { "messageId", System.Text.Encoding.UTF8.GetBytes(messageId)} }, - null - ); - - tasks.Add(task.ContinueWith(t => - { - if (!t.IsFaulted) - { - var message = t.Result.Message; - pendingConfirms.TryRemove(message.GetMessageId(), out _); - } - })); + foreach (var outgoingEvent in outgoingEventArray) + { + var messageId = outgoingEvent.Id.ToString("N"); + var headers = new Headers + { + { "messageId", System.Text.Encoding.UTF8.GetBytes(messageId)} + }; + + producer.Produce( + AbpKafkaEventBusOptions.TopicName, + new Message + { + Key = outgoingEvent.EventName, + Value = outgoingEvent.EventData, + Headers = headers + }); + } + + producer.CommitTransaction(); } - - await Task.WhenAll(tasks); - - return new MultipleOutgoingEventPublishResult(outgoingEventArray.Where(x => !pendingConfirms.Select(p => p.Value).Contains(x.Id)).ToList()); + catch (Exception e) + { + producer.AbortTransaction(); + throw; + } + + return Task.CompletedTask; } public async override Task ProcessFromInboxAsync( @@ -270,10 +272,26 @@ public class KafkaDistributedEventBus : DistributedEventBusBase, ISingletonDepen return PublishAsync(topicName, eventName, body, headers, headersArguments); } - private Task> PublishAsync(string topicName, string eventName, byte[] body, Headers headers, Dictionary headersArguments) + private Task> PublishAsync( + string topicName, + string eventName, + byte[] body, + Headers headers, + Dictionary headersArguments) { var producer = ProducerPool.Get(AbpKafkaEventBusOptions.ConnectionName); + return PublishAsync(producer, topicName, eventName, body, headers, headersArguments); + } + + private Task> PublishAsync( + IProducer producer, + string topicName, + string eventName, + byte[] body, + Headers headers, + Dictionary headersArguments) + { SetEventMessageHeaders(headers, headersArguments); return producer.ProduceAsync( diff --git a/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs b/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs index 9b3bc5b76e..b9210a6eb8 100644 --- a/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs +++ b/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs @@ -34,6 +34,8 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe protected IRabbitMqMessageConsumerFactory MessageConsumerFactory { get; } protected IRabbitMqMessageConsumer Consumer { get; private set; } + private bool _exchangeCreated; + public RabbitMqDistributedEventBus( IOptions options, IConnectionPool connectionPool, @@ -84,6 +86,8 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe Consumer.OnMessageReceived(ProcessEventAsync); SubscribeHandlers(AbpDistributedEventBusOptions.Handlers); + + } private async Task ProcessEventAsync(IModel channel, BasicDeliverEventArgs ea) @@ -197,7 +201,7 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe return PublishAsync(outgoingEvent.EventName, outgoingEvent.EventData, null, eventId: outgoingEvent.Id); } - public async override Task PublishManyFromOutboxAsync( + public async override Task PublishManyFromOutboxAsync( IEnumerable outgoingEvents, OutboxConfig outboxConfig) { @@ -206,48 +210,17 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe var outgoingEventArray = outgoingEvents.ToArray(); channel.ConfirmSelect(); - var pendingConfirms = new ConcurrentDictionary(); - var failures = new ConcurrentBag(); - - void CleanPendingConfirms(ulong sequenceNumber, bool multiple, bool ack) - { - if (multiple) - { - var confirmed = pendingConfirms.Where(x => x.Key <= sequenceNumber); - foreach (var entry in confirmed) - { - pendingConfirms.TryRemove(entry.Key, out var eventId); - - if (!ack) - { - failures.Add(eventId); - } - } - } - else - { - pendingConfirms.TryRemove(sequenceNumber, out var eventId); - - if (!ack) - { - failures.Add(eventId); - } - } - } - foreach (var outgoingEvent in outgoingEventArray) { - pendingConfirms.TryAdd(channel.NextPublishSeqNo, outgoingEvent.Id); - await PublishAsync(channel, outgoingEvent.EventName, outgoingEvent.EventData, null, + await PublishAsync( + channel, + outgoingEvent.EventName, + outgoingEvent.EventData, + properties: null, eventId: outgoingEvent.Id); } - channel.BasicAcks += (_, ea) => CleanPendingConfirms(ea.DeliveryTag, ea.Multiple, true); - channel.BasicNacks += (_, ea) => CleanPendingConfirms(ea.DeliveryTag, ea.Multiple, false); - - channel.WaitForConfirms(); - - return new MultipleOutgoingEventPublishResult(outgoingEventArray.Where(x => !failures.Contains(x.Id)).ToList()); + channel.WaitForConfirmsOrDie(); } } @@ -308,11 +281,7 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe Dictionary headersArguments = null, Guid? eventId = null) { - channel.ExchangeDeclare( - AbpRabbitMqEventBusOptions.ExchangeName, - "direct", - durable: true - ); + EnsureExchangeExists(channel); if (properties == null) { @@ -338,6 +307,20 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe return Task.CompletedTask; } + private void EnsureExchangeExists(IModel channel) + { + if (_exchangeCreated) + { + return; + } + + channel.ExchangeDeclare( + AbpRabbitMqEventBusOptions.ExchangeName, + "direct", + durable: true + ); + } + private void SetEventMessageHeaders(IBasicProperties properties, Dictionary headersArguments) { if (headersArguments == null) diff --git a/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusDistributedEventBus.cs b/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusDistributedEventBus.cs index a5cac31235..4ccd72ba15 100644 --- a/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusDistributedEventBus.cs +++ b/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusDistributedEventBus.cs @@ -162,7 +162,7 @@ public class RebusDistributedEventBus : DistributedEventBusBase, ISingletonDepen return; } - await Rebus.Advanced.Routing.Send(AbpRebusEventBusOptions.InputQueueName, eventData); + await Rebus.Publish(eventData); } protected override void AddToUnitOfWork(IUnitOfWork unitOfWork, UnitOfWorkEventRecord eventRecord) @@ -224,7 +224,7 @@ public class RebusDistributedEventBus : DistributedEventBusBase, ISingletonDepen return PublishToEventBusAsync(eventType, eventData); } - public async override Task PublishManyFromOutboxAsync(IEnumerable outgoingEvents, OutboxConfig outboxConfig) + public async override Task PublishManyFromOutboxAsync(IEnumerable outgoingEvents, OutboxConfig outboxConfig) { var outgoingEventArray = outgoingEvents.ToArray(); @@ -234,11 +234,9 @@ public class RebusDistributedEventBus : DistributedEventBusBase, ISingletonDepen { await PublishFromOutboxAsync(outgoingEvent, outboxConfig); } - + await scope.CompleteAsync(); } - - return new MultipleOutgoingEventPublishResult(outgoingEventArray); } public async override Task ProcessFromInboxAsync( diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/AbpEventBusBoxesOptions.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/AbpEventBusBoxesOptions.cs index a46fc76238..749aa3d6d2 100644 --- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/AbpEventBusBoxesOptions.cs +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/AbpEventBusBoxesOptions.cs @@ -34,6 +34,11 @@ public class AbpEventBusBoxesOptions /// Default: 2 hours /// public TimeSpan WaitTimeToDeleteProcessedInboxEvents { get; set; } + + /// + /// Default: false + /// + public bool OutboxPublishInBatch { get; set; } public AbpEventBusBoxesOptions() { @@ -43,5 +48,6 @@ public class AbpEventBusBoxesOptions PeriodTimeSpan = TimeSpan.FromSeconds(2); DistributedLockWaitDuration = TimeSpan.FromSeconds(15); WaitTimeToDeleteProcessedInboxEvents = TimeSpan.FromHours(2); + OutboxPublishInBatch = false; } } diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/DistributedEventBusBase.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/DistributedEventBusBase.cs index 0add75f9c4..69ae91505d 100644 --- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/DistributedEventBusBase.cs +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/DistributedEventBusBase.cs @@ -85,7 +85,7 @@ public abstract class DistributedEventBusBase : EventBusBase, IDistributedEventB OutboxConfig outboxConfig ); - public abstract Task PublishManyFromOutboxAsync( + public abstract Task PublishManyFromOutboxAsync( IEnumerable outgoingEvents, OutboxConfig outboxConfig ); diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/ISupportsEventBoxes.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/ISupportsEventBoxes.cs index 7c23be5ecd..73f74a68ad 100644 --- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/ISupportsEventBoxes.cs +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/ISupportsEventBoxes.cs @@ -10,7 +10,7 @@ public interface ISupportsEventBoxes OutboxConfig outboxConfig ); - Task PublishManyFromOutboxAsync( + Task PublishManyFromOutboxAsync( IEnumerable outgoingEvents, OutboxConfig outboxConfig ); diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/OutboxSender.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/OutboxSender.cs index 17dec0a810..ac92740e86 100644 --- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/OutboxSender.cs +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/OutboxSender.cs @@ -1,4 +1,5 @@ using System; +using System.Collections.Generic; using System.Diagnostics; using System.Linq; using System.Threading; @@ -84,14 +85,15 @@ public class OutboxSender : IOutboxSender, ITransientDependency } Logger.LogInformation($"Found {waitingEvents.Count} events in the outbox."); - - var result = await DistributedEventBus - .AsSupportsEventBoxes() - .PublishManyFromOutboxAsync(waitingEvents, OutboxConfig); - - await Outbox.DeleteManyAsync(result.PublishedOutgoingEvents.Select(x => x.Id).ToArray()); - - Logger.LogInformation($"Sent {result.PublishedOutgoingEvents.Count} events to message broker"); + + if (EventBusBoxesOptions.OutboxPublishInBatch) + { + await PublishOutgoingMessagesInBatchAsync(waitingEvents); + } + else + { + await PublishOutgoingMessagesAsync(waitingEvents); + } } } else @@ -105,4 +107,32 @@ public class OutboxSender : IOutboxSender, ITransientDependency } } } + + protected virtual async Task PublishOutgoingMessagesAsync(List waitingEvents) + { + foreach (var waitingEvent in waitingEvents) + { + await DistributedEventBus + .AsSupportsEventBoxes() + .PublishFromOutboxAsync( + waitingEvent, + OutboxConfig + ); + + await Outbox.DeleteAsync(waitingEvent.Id); + + Logger.LogInformation($"Sent the event to the message broker with id = {waitingEvent.Id:N}"); + } + } + + protected virtual async Task PublishOutgoingMessagesInBatchAsync(List waitingEvents) + { + await DistributedEventBus + .AsSupportsEventBoxes() + .PublishManyFromOutboxAsync(waitingEvents, OutboxConfig); + + await Outbox.DeleteManyAsync(waitingEvents.Select(x => x.Id).ToArray()); + + Logger.LogInformation($"Sent {waitingEvents.Count} events to message broker"); + } } diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/MultipleOutgoingEventPublishResult.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/MultipleOutgoingEventPublishResult.cs deleted file mode 100644 index 515dec1ae6..0000000000 --- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/MultipleOutgoingEventPublishResult.cs +++ /dev/null @@ -1,14 +0,0 @@ -using System.Collections.Generic; -using Volo.Abp.EventBus.Distributed; - -namespace Volo.Abp.EventBus; - -public class MultipleOutgoingEventPublishResult -{ - public IReadOnlyList PublishedOutgoingEvents { get; } - - public MultipleOutgoingEventPublishResult(IReadOnlyList outgoingEvents) - { - PublishedOutgoingEvents = outgoingEvents; - } -} diff --git a/framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/ConsumerPool.cs b/framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/ConsumerPool.cs index 81a5c195c1..7147c81271 100644 --- a/framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/ConsumerPool.cs +++ b/framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/ConsumerPool.cs @@ -76,6 +76,7 @@ public class ConsumerPool : IConsumerPool, ISingletonDependency try { + consumer.Value.Unsubscribe(); consumer.Value.Close(); consumer.Value.Dispose(); } diff --git a/framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/KafkaMessageConsumer.cs b/framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/KafkaMessageConsumer.cs index da21f48883..704c15845f 100644 --- a/framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/KafkaMessageConsumer.cs +++ b/framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/KafkaMessageConsumer.cs @@ -80,9 +80,11 @@ public class KafkaMessageConsumer : IKafkaMessageConsumer, ITransientDependency, protected virtual async Task Timer_Elapsed(AbpAsyncTimer timer) { - await CreateTopicAsync(); - Consume(); - Timer.Stop(); + if (Consumer == null) + { + await CreateTopicAsync(); + Consume(); + } } protected virtual async Task CreateTopicAsync() @@ -164,12 +166,21 @@ public class KafkaMessageConsumer : IKafkaMessageConsumer, ITransientDependency, public virtual void Dispose() { + Timer.Stop(); if (Consumer == null) { return; } - Consumer.Close(); - Consumer.Dispose(); + try + { + Consumer.Unsubscribe(); + Consumer.Close(); + Consumer.Dispose(); + Consumer = null; + } + catch (ObjectDisposedException) + { + } } } diff --git a/framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/ProducerPool.cs b/framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/ProducerPool.cs index 49287a20e9..0a31a3483a 100644 --- a/framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/ProducerPool.cs +++ b/framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/ProducerPool.cs @@ -17,6 +17,8 @@ public class ProducerPool : IProducerPool, ISingletonDependency protected ConcurrentDictionary>> Producers { get; } protected TimeSpan TotalDisposeWaitDuration { get; set; } = TimeSpan.FromSeconds(10); + + protected TimeSpan DefaultTransactionsWaitDuration { get; set; } = TimeSpan.FromSeconds(30); public ILogger Logger { get; set; } @@ -37,11 +39,18 @@ public class ProducerPool : IProducerPool, ISingletonDependency return Producers.GetOrAdd( connectionName, connection => new Lazy>(() => { - var config = Options.Connections.GetOrDefault(connection); - - Options.ConfigureProducer?.Invoke(new ProducerConfig(config)); - - return new ProducerBuilder(config).Build(); + var producerConfig = new ProducerConfig(Options.Connections.GetOrDefault(connection)); + Options.ConfigureProducer?.Invoke(producerConfig); + + if (producerConfig.TransactionalId.IsNullOrWhiteSpace()) + { + producerConfig.TransactionalId = Guid.NewGuid().ToString(); + } + + var producer = new ProducerBuilder(producerConfig).Build(); + producer.InitTransactions(DefaultTransactionsWaitDuration); + + return producer; })).Value; } @@ -69,7 +78,7 @@ public class ProducerPool : IProducerPool, ISingletonDependency foreach (var producer in Producers.Values) { var poolItemDisposeStopwatch = Stopwatch.StartNew(); - + try { producer.Value.Dispose(); @@ -77,19 +86,19 @@ public class ProducerPool : IProducerPool, ISingletonDependency catch { } - + poolItemDisposeStopwatch.Stop(); - + remainingWaitDuration = remainingWaitDuration > poolItemDisposeStopwatch.Elapsed ? remainingWaitDuration.Subtract(poolItemDisposeStopwatch.Elapsed) : TimeSpan.Zero; } - + poolDisposeStopwatch.Stop(); - + Logger.LogInformation( $"Disposed Kafka Producer Pool ({Producers.Count} producers in {poolDisposeStopwatch.Elapsed.TotalMilliseconds:0.00} ms)."); - + if (poolDisposeStopwatch.Elapsed.TotalSeconds > 5.0) { Logger.LogWarning( From f1cbcb16c7e7fa5f4b21603d981fdbf9619dc223 Mon Sep 17 00:00:00 2001 From: liangshiwei Date: Fri, 18 Mar 2022 11:55:14 +0800 Subject: [PATCH 10/11] Update RabbitMqDistributedEventBus --- .../RabbitMq/RabbitMqDistributedEventBus.cs | 22 ++++++++++++------- 1 file changed, 14 insertions(+), 8 deletions(-) diff --git a/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs b/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs index b9210a6eb8..73ca0cbb61 100644 --- a/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs +++ b/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs @@ -86,8 +86,6 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe Consumer.OnMessageReceived(ProcessEventAsync); SubscribeHandlers(AbpDistributedEventBusOptions.Handlers); - - } private async Task ProcessEventAsync(IModel channel, BasicDeliverEventArgs ea) @@ -313,12 +311,20 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe { return; } - - channel.ExchangeDeclare( - AbpRabbitMqEventBusOptions.ExchangeName, - "direct", - durable: true - ); + + try + { + channel.ExchangeDeclarePassive(AbpRabbitMqEventBusOptions.ExchangeName); + } + catch (Exception) + { + channel.ExchangeDeclare( + AbpRabbitMqEventBusOptions.ExchangeName, + "direct", + durable: true + ); + } + _exchangeCreated = true; } private void SetEventMessageHeaders(IBasicProperties properties, Dictionary headersArguments) From 0a8b45e09d3c594640e6d2e8e0ee8f3293c0b730 Mon Sep 17 00:00:00 2001 From: liangshiwei Date: Fri, 29 Apr 2022 10:31:30 +0800 Subject: [PATCH 11/11] Batch publish outbox events by default --- .../Abp/EventBus/Distributed/AbpEventBusBoxesOptions.cs | 6 +++--- .../Volo/Abp/EventBus/Distributed/OutboxSender.cs | 2 +- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/AbpEventBusBoxesOptions.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/AbpEventBusBoxesOptions.cs index 749aa3d6d2..d92c48db3a 100644 --- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/AbpEventBusBoxesOptions.cs +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/AbpEventBusBoxesOptions.cs @@ -36,9 +36,9 @@ public class AbpEventBusBoxesOptions public TimeSpan WaitTimeToDeleteProcessedInboxEvents { get; set; } /// - /// Default: false + /// Default: true /// - public bool OutboxPublishInBatch { get; set; } + public bool BatchPublishOutboxEvents { get; set; } public AbpEventBusBoxesOptions() { @@ -48,6 +48,6 @@ public class AbpEventBusBoxesOptions PeriodTimeSpan = TimeSpan.FromSeconds(2); DistributedLockWaitDuration = TimeSpan.FromSeconds(15); WaitTimeToDeleteProcessedInboxEvents = TimeSpan.FromHours(2); - OutboxPublishInBatch = false; + BatchPublishOutboxEvents = true; } } diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/OutboxSender.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/OutboxSender.cs index ac92740e86..851ddf3da1 100644 --- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/OutboxSender.cs +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/OutboxSender.cs @@ -86,7 +86,7 @@ public class OutboxSender : IOutboxSender, ITransientDependency Logger.LogInformation($"Found {waitingEvents.Count} events in the outbox."); - if (EventBusBoxesOptions.OutboxPublishInBatch) + if (EventBusBoxesOptions.BatchPublishOutboxEvents) { await PublishOutgoingMessagesInBatchAsync(waitingEvents); }