tsai
/
abp
mirror of https://github.com/abpframework/abp.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

Refactor inbox retry policy configuration

pull/23563/head
maliming 1 year ago
parent
79d62ccad9
commit
8e06551791
No known key found for this signature in database GPG Key ID: A646B9CB645ECEA4
5 changed files with 27 additions and 43 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 1
      framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/DbContextEventInbox.cs
  2. 10
      framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/AbpEventBusBoxesOptions.cs
  3. 15
      framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/AbpInboxProcessorOptions.cs
  4. 43
      framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/InboxProcessor.cs
  5. 1
      framework/src/Volo.Abp.MongoDB/Volo/Abp/MongoDB/DistributedEvents/MongoDbContextEventInbox.cs

1
framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/DbContextEventInbox.cs
View File

@ -51,6 +51,7 @@ public class DbContextEventInbox<TDbContext> : IDbContextEventInbox<TDbContext>
.IncomingEvents
.AsNoTracking()
.Where(x => x.Status == IncomingEventStatus.Pending)
.Where(x => x.NextRetryTime == null || x.NextRetryTime <= Clock.Now)
.WhereIf(transformedFilter != null, transformedFilter!)
.OrderBy(x => x.CreationTime)
.Take(maxCount)

10
framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/AbpEventBusBoxesOptions.cs
View File

@ -36,6 +36,16 @@ public class AbpEventBusBoxesOptions
/// </summary>
public TimeSpan PeriodTimeSpan { get; set; }
/// <summary>
/// Default: <see cref="InboxProcessorFailurePolicy.Retry"/>
/// </summary>
public InboxProcessorFailurePolicy InboxProcessorFailurePolicy { get; set; } = InboxProcessorFailurePolicy.Retry;
/// <summary>
/// Default: 10
/// </summary>
public int InboxProcessorMaxRetryCount { get; set; } = 10;
/// <summary>
/// Default: 15 seconds
/// </summary>

15
framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/AbpInboxProcessorOptions.cs
View File

@ -1,15 +0,0 @@
namespace Volo.Abp.EventBus.Distributed;
public class AbpInboxProcessorOptions
{
public InboxProcessorFailurePolicy FailurePolicy { get; set; } = InboxProcessorFailurePolicy.Retry;
/// <summary>
/// Retry intervals follow an exponential backoff strategy:
/// The delay for each retry is twice the previous one,
/// starting with an initial delay of 1 second.
/// For example: 1s, 2s, 4s, 8s, 16s, 32s ...
/// Maximum of 10 retries.
/// </summary>
public int MaxRetryCount { get; set; } = 10;
}

43
framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/InboxProcessor.cs
View File

@ -1,5 +1,6 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.DependencyInjection;
@ -25,7 +26,6 @@ public class InboxProcessor : IInboxProcessor, ITransientDependency
protected IEventInbox Inbox { get; private set; } = default!;
protected InboxConfig InboxConfig { get; private set; } = default!;
protected AbpEventBusBoxesOptions EventBusBoxesOptions { get; }
protected AbpInboxProcessorOptions AbpInboxProcessorOptions { get; set; }
protected DateTime? LastCleanTime { get; set; }
protected string DistributedLockName { get; set; } = default!;
@ -40,8 +40,7 @@ public class InboxProcessor : IInboxProcessor, ITransientDependency
IAbpDistributedLock distributedLock,
IUnitOfWorkManager unitOfWorkManager,
IClock clock,
IOptions<AbpEventBusBoxesOptions> eventBusBoxesOptions,
IOptions<AbpInboxProcessorOptions> inboxProcessorOptions)
IOptions<AbpEventBusBoxesOptions> eventBusBoxesOptions)
{
ServiceProvider = serviceProvider;
Timer = timer;
@ -50,7 +49,6 @@ public class InboxProcessor : IInboxProcessor, ITransientDependency
UnitOfWorkManager = unitOfWorkManager;
Clock = clock;
EventBusBoxesOptions = eventBusBoxesOptions.Value;
AbpInboxProcessorOptions = inboxProcessorOptions.Value;
Timer.Period = Convert.ToInt32(EventBusBoxesOptions.PeriodTimeSpan.TotalMilliseconds);
Timer.Elapsed += TimerOnElapsed;
Logger = NullLogger<InboxProcessor>.Instance;
@ -107,12 +105,6 @@ public class InboxProcessor : IInboxProcessor, ITransientDependency
{
Logger.LogInformation($"Start processing the incoming event with id = {waitingEvent.Id:N}");
if (waitingEvent.NextRetryTime.HasValue && waitingEvent.NextRetryTime.Value > Clock.Now)
{
Logger.LogInformation($"Event with id = {waitingEvent.Id:N} is not ready to be processed yet. Next retry time: {waitingEvent.NextRetryTime.Value}");
continue;
}
try
{
using (var uow = UnitOfWorkManager.Begin(isTransactional: true, requiresNew: true))
@ -130,41 +122,42 @@ public class InboxProcessor : IInboxProcessor, ITransientDependency
}
catch (Exception e)
{
Logger.LogError(e, $"An error occurred while processing the incoming event with id = {waitingEvent.Id:N}");
Logger.LogError(e, $"Event with id = {waitingEvent.Id:N} processing failed.");
if (AbpInboxProcessorOptions.FailurePolicy == InboxProcessorFailurePolicy.Retry)
if (EventBusBoxesOptions.InboxProcessorFailurePolicy == InboxProcessorFailurePolicy.Retry)
{
throw;
}
if (AbpInboxProcessorOptions.FailurePolicy == InboxProcessorFailurePolicy.RetryLater)
if (EventBusBoxesOptions.InboxProcessorFailurePolicy == InboxProcessorFailurePolicy.RetryLater)
{
using (var uow = UnitOfWorkManager.Begin(isTransactional: true, requiresNew: true))
{
if (waitingEvent.RetryCount > AbpInboxProcessorOptions.MaxRetryCount)
if (waitingEvent.RetryCount > EventBusBoxesOptions.InboxProcessorMaxRetryCount)
{
Logger.LogWarning($"Max retry count reached for event with id = {waitingEvent.Id:N}. Discarding the event.");
Logger.LogWarning($"Event with id = {waitingEvent.Id:N} has exceeded the maximum retry count. Marking it as discarded.");
await Inbox.MarkAsDiscardAsync(waitingEvent.Id);
await uow.CompleteAsync(StoppingToken);
continue;
}
Logger.LogInformation($"Retrying event with id = {waitingEvent.Id:N}. " +
$"Retry count: {waitingEvent.RetryCount}, " +
$"Next retry time: {GetNextRetryTime(waitingEvent.RetryCount, AbpInboxProcessorOptions.MaxRetryCount)}");
Logger.LogInformation($"Event with id = {waitingEvent.Id:N} will retry later. " +
$"Current retry count: {waitingEvent.RetryCount}, " +
$"Next retry time: {GetNextRetryTime(waitingEvent.RetryCount)}" +
$"Max retry count: {EventBusBoxesOptions.InboxProcessorMaxRetryCount}.");
await Inbox.RetryLaterAsync(waitingEvent.Id, ++waitingEvent.RetryCount, GetNextRetryTime(waitingEvent.RetryCount, AbpInboxProcessorOptions.MaxRetryCount));
await Inbox.RetryLaterAsync(waitingEvent.Id, ++waitingEvent.RetryCount, GetNextRetryTime(waitingEvent.RetryCount));
await uow.CompleteAsync(StoppingToken);
}
continue;
}
if (AbpInboxProcessorOptions.FailurePolicy == InboxProcessorFailurePolicy.Discard)
if (EventBusBoxesOptions.InboxProcessorFailurePolicy == InboxProcessorFailurePolicy.Discard)
{
using (var uow = UnitOfWorkManager.Begin(isTransactional: true, requiresNew: true))
{
Logger.LogInformation($"Discarding event with id = {waitingEvent.Id:N} due to an error.");
Logger.LogInformation($"Event with id = {waitingEvent.Id:N} will be discarded.");
await Inbox.MarkAsDiscardAsync(waitingEvent.Id);
await uow.CompleteAsync(StoppingToken);
@ -187,15 +180,9 @@ public class InboxProcessor : IInboxProcessor, ITransientDependency
}
}
protected virtual DateTime? GetNextRetryTime(int retryCount, int maxRetryCount)
protected virtual DateTime? GetNextRetryTime(int retryCount)
{
if (retryCount > maxRetryCount)
{
return null;
}
var delaySeconds = 1 * (int)Math.Pow(2, retryCount - 1);
return DateTime.Now.AddSeconds(delaySeconds);
}

1
framework/src/Volo.Abp.MongoDB/Volo/Abp/MongoDB/DistributedEvents/MongoDbContextEventInbox.cs
View File

@ -65,6 +65,7 @@ public class MongoDbContextEventInbox<TMongoDbContext> : IMongoDbContextEventInb
.IncomingEvents
.AsQueryable()
.Where(x => x.Status == IncomingEventStatus.Pending)
.Where(x => x.NextRetryTime == null || x.NextRetryTime <= Clock.Now)
.WhereIf(transformedFilter != null, transformedFilter!)
.OrderBy(x => x.CreationTime)
.Take(maxCount)

Write Preview
Loading…
Cancel
Save