这在很大程度上受到了skjagini's answer 中链接的documentation 的启发，并进行了一些改进。
我认为在此重申整个示例可能会有所帮助，以防链接在某些时候断开。我做了一些调整；最值得注意的是，我注入了一个IServiceScopeFactory，以允许后台进程自己安全地请求服务。我在这个答案的末尾解释了我的推理。
核心思想是创建一个任务队列，用户可以将其注入到他们的控制器中，然后分配任务。 长期运行的托管服务中存在相同的任务队列，该服务一次将一个任务出列并执行。
任务队列：
public interface IBackgroundTaskQueue
{
// Enqueues the given task.
void EnqueueTask(Func<IServiceScopeFactory, CancellationToken, Task> task);
// Dequeues and returns one task. This method blocks until a task becomes available.
Task<Func<IServiceScopeFactory, CancellationToken, Task>> DequeueAsync(CancellationToken cancellationToken);
}
public class BackgroundTaskQueue : IBackgroundTaskQueue
{
private readonly ConcurrentQueue<Func<IServiceScopeFactory, CancellationToken, Task>> _items = new();
// Holds the current count of tasks in the queue.
private readonly SemaphoreSlim _signal = new SemaphoreSlim(0);
public void EnqueueTask(Func<IServiceScopeFactory, CancellationToken, Task> task)
{
if(task == null)
throw new ArgumentNullException(nameof(task));
_items.Enqueue(task);
_signal.Release();
}
public async Task<Func<IServiceScopeFactory, CancellationToken, Task>> DequeueAsync(CancellationToken cancellationToken)
{
// Wait for task to become available
await _signal.WaitAsync(cancellationToken);
_items.TryDequeue(out var task);
return task;
}
}
在任务队列的核心，我们有一个线程安全的ConcurrentQueue<>。由于我们不想在新任务可用之前轮询队列，因此我们使用SemaphoreSlim 对象来跟踪队列中当前的任务数。每次我们调用Release，内部计数器都会递增。 WaitAsync 方法会阻塞，直到内部计数器大于 0，然后递减它。
为了出队和执行任务，我们创建了一个后台服务：
public class BackgroundQueueHostedService : BackgroundService
{
private readonly IBackgroundTaskQueue _taskQueue;
private readonly IServiceScopeFactory _serviceScopeFactory;
private readonly ILogger<BackgroundQueueHostedService> _logger;
public BackgroundQueueHostedService(IBackgroundTaskQueue taskQueue, IServiceScopeFactory serviceScopeFactory, ILogger<BackgroundQueueHostedService> logger)
{
_taskQueue = taskQueue ?? throw new ArgumentNullException(nameof(taskQueue));
_serviceScopeFactory = serviceScopeFactory ?? throw new ArgumentNullException(nameof(serviceScopeFactory));
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
// Dequeue and execute tasks until the application is stopped
while(!stoppingToken.IsCancellationRequested)
{
// Get next task
// This blocks until a task becomes available
var task = await _taskQueue.DequeueAsync(stoppingToken);
try
{
// Run task
await task(_serviceScopeFactory, stoppingToken);
}
catch(Exception ex)
{
_logger.LogError(ex, "An error occured during execution of a background task");
}
}
}
}
最后，我们需要让我们的任务队列可用于依赖注入，并启动我们的后台服务：
public void ConfigureServices(IServiceCollection services)
{
// ...
services.AddSingleton<IBackgroundTaskQueue, BackgroundTaskQueue>();
services.AddHostedService<BackgroundQueueHostedService>();
// ...
}
我们现在可以将后台任务队列注入我们的控制器并将任务入队：
public class ExampleController : Controller
{
private readonly IBackgroundTaskQueue _backgroundTaskQueue;
public ExampleController(IBackgroundTaskQueue backgroundTaskQueue)
{
_backgroundTaskQueue = backgroundTaskQueue ?? throw new ArgumentNullException(nameof(backgroundTaskQueue));
}
public IActionResult Index()
{
_backgroundTaskQueue.EnqueueTask(async (serviceScopeFactory, cancellationToken) =>
{
// Get services
using var scope = serviceScopeFactory.CreateScope();
var myService = scope.ServiceProvider.GetRequiredService<IMyService>();
var logger = scope.ServiceProvider.GetRequiredService<ILogger<ExampleController>>();
try
{
// Do something expensive
await myService.DoSomethingAsync(cancellationToken);
}
catch(Exception ex)
{
logger.LogError(ex, "Could not do something expensive");
}
});
return Ok();
}
}
为什么要使用IServiceScopeFactory？
理论上，我们可以直接使用我们注入到控制器中的服务对象。这可能适用于单例服务以及大多数范围服务。
但是，对于实现IDisposable（例如DbContext）的作用域服务，这可能会中断：将任务入队后，控制器方法返回并且请求完成。然后框架清理注入的服务。如果我们的后台任务足够慢或延迟，它可能会尝试调用已释放服务的方法，然后会遇到错误。
为避免这种情况，我们的排队任务应始终创建自己的服务范围，并且不应使用来自周围控制器的服务实例。