dotnet add package Core.TaskProcessor
Hangfire Pro Redis except:
- open source (Apache 2.0)
- exclusive redis/elasticache 6.2+ storage engine (cluster mode supported)
- multi tenancy
- global pause and resume of all processing
- every (non-recurring) task belongs to a batch
- batch cancelation can abort in process tasks "instantly"
- easy to access batch statistics per tenant
- async extension points with access to batch information
builder.Services.AddTaskProcessor(new TaskProcessorOptions
{
Redis = "localhost:6379,abortConnect=false",
Prefix = "{coretask}", // redis cluster mode needs single hash slot
Queues = new[] { "high", "default", "low" }, // pop queues from left to right - first non empty queue wins
MaxWorkers = 4, // action block concurrency limit
Retries = 3,
Invisibility = TimeSpan.FromMinutes(5), // task will be redelivered when taking longer than this
BaseFrequency = TimeSpan.FromSeconds(5), // fetches tasks when reactive events failed
PushbackFrequency = TimeSpan.FromSeconds(10), // how often to run task retry/delay pushbacks
CleanUpFrequency = TimeSpan.FromMinutes(5), // how often to run batch cleanups
Retention = TimeSpan.FromDays(7), // batch information will be kept this long
UseHostedService = true, // use supplied background worker service
OnTaskFailedDelay = (ctx, retry) => // delay retry on task failure
Task.FromResult(retry switch
{
2 => TimeSpan.FromSeconds(5),
1 => TimeSpan.FromSeconds(60),
_ => (TimeSpan?)null,
})
});var batchId = await _processor.EnqueueBatchAsync("default", "my-tenant", batch =>
{
batch.Enqueue(() => _someScopedService.DoSomethingAsync("hello", CancellationToken.None));
batch.Enqueue(() => _someScopedService.DoSomethingAsync("world", CancellationToken.None),
delayUntil: DateTimeOffset.UtcNow.AddSeconds(30));
batch.ContinueWith(() => _someScopedService.DoSomething("!"), "high");
})- static functions
- functions on types resolveable by the IServiceProvider (scoped interfaces preferred)
- return type of void or Task
- all parameters need to be json serializable with the default implementation (custom implementations possible)
- constant parameters are fastest, yet complex lists and objects are also possible (dynamic invoked)
- only the first method call will run in background - all parameters will be evaluated at enqueue time
- beware of functions in strongly named assemblys
await _processor.AppendBatchAsync("default", "my-tenant", batchId, batch =>
{
batch.Enqueue(() => _someScopedService.DoSomethingAsync("hello 2", CancellationToken.None));
batch.Enqueue(() => _someScopedService.DoSomethingAsync("world 2", CancellationToken.None));
})await _processor.CancelBatchAsync(batchId);await _processor.UpsertScheduleAsync(new ScheduleData
{
Id = "unique-schedule-id",
Tenant = "my-tenant",
Queue = "default",
Scope = "Send hourly email",
Cron = "0 */1 * * *",
Timezone = "Etc/UTC",
Unique = true // if task is enqueued from previous cycle and hasn't completed yet, this cycle will be skipped
}, () => _someScopedService.DoSomethingAsync("scheduled task", CancellationToken.None));await proc.CancelScheduleAsync("unique-schedule-id", "my-tenant");await proc.GetBatchesAsync("my-tenant", 0, 25);
await proc.GetTasksInQueueAsync("low", 0, 25);If you need to integrate another ioc solution set UseHostedService = false and provide a custom one. Executor.InvokeAsync has an Type to instance resolver callback.
class CustomExecutorService : BackgroundService
{
private readonly ITaskProcessor _processor;
private readonly IServiceProvider _serviceProvider;
public TaskExecutorService(ITaskProcessor processor, IServiceProvider serviceProvider)
{
_processor = processor;
_serviceProvider = serviceProvider;
processor.Execute = Execute;
}
private async Task Execute(TaskContext ctx)
{
await using var scope = _serviceProvider.CreateAsyncScope();
if (ctx.Topic == "internal:expression:v1")
await _processor.Executor.InvokeAsync(ctx,
type => scope.ServiceProvider.GetRequiredService(type))
.ConfigureAwait(false);
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
await _processor.RunAsync(stoppingToken);
}
}
