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 d8a3d1341a..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 @@ -87,12 +87,33 @@ 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, outgoingEvent.Id); } - public override async Task ProcessFromInboxAsync(IncomingEventInfo incomingEvent, InboxConfig inboxConfig) + public async override Task PublishManyFromOutboxAsync(IEnumerable outgoingEvents, OutboxConfig outboxConfig) + { + var outgoingEventArray = outgoingEvents.ToArray(); + + var publisher = await _publisherPool.GetAsync( + _options.TopicName, + _options.ConnectionName); + + using var messageBatch = await publisher.CreateMessageBatchAsync(); + + foreach (var outgoingEvent in outgoingEventArray) + { + if (!messageBatch.TryAddMessage(new ServiceBusMessage(outgoingEvent.EventData) { Subject = outgoingEvent.EventName })) + { + throw new AbpException("The message is too large to fit in the batch. Set AbpEventBusBoxesOptions.OutboxWaitingEventMaxCount to reduce the number"); + } + } + + await publisher.SendMessagesAsync(messageBatch); + } + + public async override Task ProcessFromInboxAsync(IncomingEventInfo incomingEvent, InboxConfig inboxConfig) { var eventType = _eventTypes.GetOrDefault(incomingEvent.EventName); if (eventType == null) @@ -180,7 +201,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); } 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 a19b01311c..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 @@ -79,12 +79,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)) { @@ -164,7 +159,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, @@ -198,7 +193,43 @@ public class KafkaDistributedEventBus : DistributedEventBusBase, ISingletonDepen ); } - public override async Task ProcessFromInboxAsync( + public override Task PublishManyFromOutboxAsync(IEnumerable outgoingEvents, OutboxConfig outboxConfig) + { + var producer = ProducerPool.Get(); + var outgoingEventArray = outgoingEvents.ToArray(); + producer.BeginTransaction(); + try + { + 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(); + } + catch (Exception e) + { + producer.AbortTransaction(); + throw; + } + + return Task.CompletedTask; + } + + public async override Task ProcessFromInboxAsync( IncomingEventInfo incomingEvent, InboxConfig inboxConfig) { @@ -241,13 +272,29 @@ 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); + 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); - 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.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs b/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs index a2932de52e..c9ec49bc7e 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, @@ -180,7 +182,7 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe 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, eventData, null); } @@ -197,7 +199,30 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe return PublishAsync(outgoingEvent.EventName, outgoingEvent.EventData, null, eventId: outgoingEvent.Id); } - public override async Task ProcessFromInboxAsync( + public async override Task PublishManyFromOutboxAsync( + IEnumerable outgoingEvents, + OutboxConfig outboxConfig) + { + using (var channel = ConnectionPool.Get(AbpRabbitMqEventBusOptions.ConnectionName).CreateModel()) + { + var outgoingEventArray = outgoingEvents.ToArray(); + channel.ConfirmSelect(); + + foreach (var outgoingEvent in outgoingEventArray) + { + await PublishAsync( + channel, + outgoingEvent.EventName, + outgoingEvent.EventData, + properties: null, + eventId: outgoingEvent.Id); + } + + channel.WaitForConfirmsOrDie(); + } + } + + public async override Task ProcessFromInboxAsync( IncomingEventInfo incomingEvent, InboxConfig inboxConfig) { @@ -233,7 +258,7 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe return PublishAsync(eventName, body, properties, headersArguments); } - protected Task PublishAsync( + protected virtual Task PublishAsync( string eventName, byte[] body, IBasicProperties properties, @@ -242,37 +267,66 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe { using (var channel = ConnectionPool.Get(AbpRabbitMqEventBusOptions.ConnectionName).CreateModel()) { - channel.ExchangeDeclare( - AbpRabbitMqEventBusOptions.ExchangeName, - AbpRabbitMqEventBusOptions.GetExchangeTypeOrDefault(), - durable: true - ); - - if (properties == null) - { - properties = channel.CreateBasicProperties(); - properties.DeliveryMode = RabbitMqConsts.DeliveryModes.Persistent; - } + return PublishAsync(channel, eventName, body, properties, headersArguments, eventId); + } + } - if (properties.MessageId.IsNullOrEmpty()) - { - properties.MessageId = (eventId ?? GuidGenerator.Create()).ToString("N"); - } + protected virtual Task PublishAsync( + IModel channel, + string eventName, + byte[] body, + IBasicProperties properties, + Dictionary headersArguments = null, + Guid? eventId = null) + { + EnsureExchangeExists(channel); - SetEventMessageHeaders(properties, headersArguments); + if (properties == null) + { + properties = channel.CreateBasicProperties(); + properties.DeliveryMode = RabbitMqConsts.DeliveryModes.Persistent; + } - channel.BasicPublish( - exchange: AbpRabbitMqEventBusOptions.ExchangeName, - routingKey: eventName, - mandatory: true, - basicProperties: properties, - body: body - ); + if (properties.MessageId.IsNullOrEmpty()) + { + properties.MessageId = (eventId ?? GuidGenerator.Create()).ToString("N"); } + SetEventMessageHeaders(properties, headersArguments); + + channel.BasicPublish( + exchange: AbpRabbitMqEventBusOptions.ExchangeName, + routingKey: eventName, + mandatory: true, + basicProperties: properties, + body: body + ); + return Task.CompletedTask; } + private void EnsureExchangeExists(IModel channel) + { + if (_exchangeCreated) + { + return; + } + + try + { + channel.ExchangeDeclarePassive(AbpRabbitMqEventBusOptions.ExchangeName); + } + catch (Exception) + { + channel.ExchangeDeclare( + AbpRabbitMqEventBusOptions.ExchangeName, + AbpRabbitMqEventBusOptions.GetExchangeTypeOrDefault(), + durable: true + ); + } + _exchangeCreated = true; + } + private void SetEventMessageHeaders(IBasicProperties properties, Dictionary headersArguments) { if (headersArguments == null) @@ -306,7 +360,7 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe var handlerFactoryList = new List(); foreach (var handlerFactory in - HandlerFactories.Where(hf => ShouldTriggerEventForHandler(eventType, hf.Key))) + HandlerFactories.Where(hf => ShouldTriggerEventForHandler(eventType, hf.Key))) { handlerFactoryList.Add( new EventTypeWithEventHandlerFactories(handlerFactory.Key, handlerFactory.Value)); 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 cf6c1d6b0f..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 @@ -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; @@ -149,6 +150,11 @@ public class RebusDistributedEventBus : DistributedEventBusBase, ISingletonDepen } protected async override Task PublishToEventBusAsync(Type eventType, object eventData) + { + await PublishAsync(eventType, eventData); + } + + protected virtual async Task PublishAsync(Type eventType, object eventData) { if (AbpRebusEventBusOptions.Publish != null) { @@ -218,7 +224,22 @@ public class RebusDistributedEventBus : DistributedEventBusBase, ISingletonDepen return PublishToEventBusAsync(eventType, eventData); } - public override async Task ProcessFromInboxAsync( + public async override Task PublishManyFromOutboxAsync(IEnumerable outgoingEvents, OutboxConfig outboxConfig) + { + var outgoingEventArray = outgoingEvents.ToArray(); + + using (var scope = new RebusTransactionScope()) + { + foreach (var outgoingEvent in outgoingEventArray) + { + await PublishFromOutboxAsync(outgoingEvent, outboxConfig); + } + + await scope.CompleteAsync(); + } + } + + public async override Task ProcessFromInboxAsync( IncomingEventInfo incomingEvent, InboxConfig inboxConfig) { 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..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 @@ -34,6 +34,11 @@ public class AbpEventBusBoxesOptions /// Default: 2 hours /// public TimeSpan WaitTimeToDeleteProcessedInboxEvents { get; set; } + + /// + /// Default: true + /// + public bool BatchPublishOutboxEvents { get; set; } public AbpEventBusBoxesOptions() { @@ -43,5 +48,6 @@ public class AbpEventBusBoxesOptions PeriodTimeSpan = TimeSpan.FromSeconds(2); DistributedLockWaitDuration = TimeSpan.FromSeconds(15); WaitTimeToDeleteProcessedInboxEvents = TimeSpan.FromHours(2); + BatchPublishOutboxEvents = true; } } 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 6547c1686c..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,6 +85,11 @@ public abstract class DistributedEventBusBase : EventBusBase, IDistributedEventB OutboxConfig outboxConfig ); + public abstract Task PublishManyFromOutboxAsync( + IEnumerable outgoingEvents, + OutboxConfig outboxConfig + ); + public abstract Task ProcessFromInboxAsync( IncomingEventInfo incomingEvent, InboxConfig inboxConfig); @@ -101,7 +106,8 @@ public abstract class DistributedEventBusBase : EventBusBase, IDistributedEventB { if (outboxConfig.Selector == null || outboxConfig.Selector(eventType)) { - var eventOutbox = (IEventOutbox)unitOfWork.ServiceProvider.GetRequiredService(outboxConfig.ImplementationType); + var eventOutbox = + (IEventOutbox)unitOfWork.ServiceProvider.GetRequiredService(outboxConfig.ImplementationType); var eventName = EventNameAttribute.GetNameOrDefault(eventType); await eventOutbox.EnqueueAsync( new OutgoingEventInfo( @@ -135,7 +141,8 @@ public abstract class DistributedEventBusBase : EventBusBase, IDistributedEventB { if (inboxConfig.EventSelector == null || inboxConfig.EventSelector(eventType)) { - var eventInbox = (IEventInbox)scope.ServiceProvider.GetRequiredService(inboxConfig.ImplementationType); + var eventInbox = + (IEventInbox)scope.ServiceProvider.GetRequiredService(inboxConfig.ImplementationType); if (!messageId.IsNullOrEmpty()) { 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..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 @@ -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/OutboxSender.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/OutboxSender.cs index 6ed60ba34c..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 @@ -1,4 +1,7 @@ using System; +using System.Collections.Generic; +using System.Diagnostics; +using System.Linq; using System.Threading; using System.Threading.Tasks; using Microsoft.Extensions.DependencyInjection; @@ -82,18 +85,14 @@ public class OutboxSender : IOutboxSender, ITransientDependency } Logger.LogInformation($"Found {waitingEvents.Count} events in the outbox."); - - foreach (var waitingEvent in waitingEvents) + + if (EventBusBoxesOptions.BatchPublishOutboxEvents) { - 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}"); + await PublishOutgoingMessagesInBatchAsync(waitingEvents); + } + else + { + await PublishOutgoingMessagesAsync(waitingEvents); } } } @@ -108,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.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( 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)); + } + } } 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); }