从 ASP.NET Core 中的控制器操作运行后台任务

Posted 2023-03-31

【中文标题】从 ASP.NET Core 中的控制器操作运行后台任务

【英文标题】：Run a background task from a controller action in ASP.NET Core

【发布时间】：2018-04-13 09:27:58

【问题描述】：

我正在使用 C# 和 ASP.NET Core 2.0 开发一个带有 REST API 的 Web 应用程序。

我想要实现的是当客户端向端点发送请求时，我将运行与客户端请求上下文分离的后台任务，如果任务成功启动，该任务将结束。

我知道有HostedService，但问题是HostedService 在服务器启动时启动，据我所知，无法从控制器手动启动HostedService。

这是一个演示问题的简单代码。

[Authorize(AuthenticationSchemes = "UsersScheme")]
public class UsersController : Controller

    [HttpPost]
    public async Task<JsonResult> StartJob([FromForm] string UserId, [FromServices] IBackgroundJobService backgroundService)
    
        // check user account
        (bool isStarted, string data) result = backgroundService.Start();

        return JsonResult(result);
    

【问题讨论】：

使用像 Hangifre 这样的第三方工具，但这里肯定有成千上万个类似的问题。

感谢您的评论，我最终使用了 Hangfire，它非常强大。考虑写一个答案，以便我接受。

【参考方案1】：

您仍然可以将IHostedService 与BlockingCollection 结合使用作为后台任务的基础。

为BlockingCollection 创建包装器，以便您可以将其作为单例注入。

public class TasksToRun

    private readonly BlockingCollection<TaskSettings> _tasks;

    public TasksToRun() => _tasks = new BlockingCollection<TaskSettings>();

    public void Enqueue(TaskSettings settings) => _tasks.Add(settings);

    public TaskSettings Dequeue(CancellationToken token) => _tasks.Take(token);

然后在IHostedService 的实现中“监听”任务并在任务“到达”时执行它。如果集合为空，BlockingCollection 将停止执行 - 因此您的while 循环不会消耗处理器时间。 .Take 方法接受 cancellationToken 作为参数。使用令牌，您可以在应用程序停止时取消“等待”下一个任务。

public class BackgroundService : IHostedService

    private readonly TasksToRun _tasks;

    private CancellationTokenSource _tokenSource;

    private Task _currentTask;

    public BackgroundService(TasksToRun tasks) => _tasks = tasks;

    public async Task StartAsync(CancellationToken cancellationToken)
    
        _tokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
        while (cancellationToken.IsCancellationRequested == false)
        
            try
            
                var taskToRun = _tasks.Dequeue(_tokenSource.Token);

                // We need to save executable task, 
                // so we can gratefully wait for it's completion in Stop method
                _currentTask = ExecuteTask(taskToRun);               
                await _currentTask;
            
            catch (OperationCanceledException)
            
                // execution cancelled
            
        
    

    public async Task StopAsync(CancellationToken cancellationToken)
    
        _tokenSource.Cancel(); // cancel "waiting" for task in blocking collection

        if (_currentTask == null) return;

        // wait when _currentTask is complete
        await Task.WhenAny(_currentTask, Task.Delay(-1, cancellationToken));
    

在控制器中，您只需将要运行的任务添加到我们的集合中

public class JobController : Controller

    private readonly TasksToRun _tasks;

    public JobController(TasksToRun tasks) => _tasks = tasks;

    public IActionResult PostJob()
    
        var settings = CreateTaskSettings();

        _tasks.Enqueue(settings);

        return Ok();
    

阻塞收集的包装器应该注册为单例依赖注入

services.AddSingleton<TasksToRun, TasksToRun>();

注册后台服务

services.AddHostedService<BackgroundService>();

【讨论】：

@TylerDurden，我想你可以问一个关于你的问题的问题，并且很确定你会得到正确的答案

@Fabio 问题在于 Deque 永远阻塞。将取消令牌从 _tokenSource 传递到 TasksToRun.Deque 可以解决此问题。现在它关闭了。

这真的有效吗？我尝试实现并很快注意到托管服务的StartAsync 方法中的BlockingCollection 的Take 方法阻止了服务启动。这也会阻止 ASP.NET 运行时启动，因为服务从未完成注册。

@Justin，不，它不起作用。我们需要在StartAsync 中返回Task.CompletedTask 才能启动Web 应用程序。实际工作可以包含在Task 中。但随后可以使用没有IHostedService 的任务。我真的很想知道为什么这个答案会获得如此多的选票。

@Fabio 您的回答并不表明它是伪代码。此外，它看起来像编写的完全有效的运行时代码，因此显然具有误导性。我建议您编辑您的答案以澄清。

【参考方案2】：

Microsoft 在https://docs.microsoft.com/en-us/aspnet/core/fundamentals/host/hosted-services?view=aspnetcore-2.1 上记录了相同的内容

它使用BackgroundTaskQueue 来完成，它从Controller 分配工作，工作由派生自BackgroundService 的QueueHostedService 执行。

【讨论】：

显示的链接与OP的问题不同。问题询问如何从控制器启动任务。引用的链接显示了如何在用户按下键时启动键盘轮询器并启动任务。完全不一样。

【参考方案3】：

这在很大程度上受到了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）的作用域服务，这可能会中断：将任务入队后，控制器方法返回并且请求完成。然后框架清理注入的服务。如果我们的后台任务足够慢或延迟，它可能会尝试调用已释放服务的方法，然后会遇到错误。

为避免这种情况，我们的排队任务应始终创建自己的服务范围，并且不应使用来自周围控制器的服务实例。

【讨论】：

如 net5 中所宣传的那样工作。谢谢。

@Chris 哦，这行有点误导。注释意味着后台作业一直等到下一个任务可用。但是，它并没有阻塞，而是使用await，因此线程被释放，可以在其他地方使用。只有当某些东西入队时，_taskQueue 才会触发ExecuteAsync 中代码的延续。参见这里：Does await completely block the thread?

@ChsharpNewbie 我不认为我完全理解你的问题。如果您的意思是可能的竞争条件问题：这是没有问题的，因为ConcurrentQueue 负责必要的锁定和序列化，而SemaphoreSlim 允许线程安全等待直到有新项目可用。所以当前任务完成了，然后后台服务从队列中取出下一个任务，如果有的话。

@ChsharpNewbie 好吧，这取决于各自的实现，但应该也可以。如果队列中没有项目，TryDequeue 将返回 false，因此需要进行相应处理。使用SemaphoreSlim 的优点是执行会等待直到有新项目可用，然后立即 使用它，而无需进一步的开销/轮询延迟等。线程安全已经由ConcurrentQueue，所以不应该有并发问题。

@ChsharpNewbie 不，这应该可以正常工作。

【参考方案4】：

你可以在ThreadPool中使用另一个线程：

排队执行的方法。该方法在线程池时执行 线程可用。

public class ToDoController : Controller

    private readonly IServiceScopeFactory _serviceScopeFactory;
    public ToDoController(IServiceScopeFactory serviceScopeFactory)
    
        _serviceScopeFactory = serviceScopeFactory;
    
    public string Index(Func<IToDoDependency,Task> DoHeavyWork)
    
        ThreadPool.QueueUserWorkItem(delegate 
            // Get services
            using var scope = _serviceScopeFactory.CreateScope();
            var dependency= scope.ServiceProvider.GetRequiredService<IToDoDependency>();
            DoHeavyWork(dependency);

            // OR 
            // Get the heavy work from ServiceProvider
            var heavyWorkSvc= scope.ServiceProvider.GetRequiredService<IHeavyWorkService>();
            heavyWorkSvc.Do(dependency);
        );
        return "Immediate Response";
    

【讨论】：

什么是DoHeavyWork？

后台操作占用大量时间，无法在请求范围内执行。或者，如果您想立即向用户发送响应然后执行操作。比如数据库/缓存更新...

我编辑了代码以使其更清晰