dotnet core系列之Background tasks with hosted services (后台任务)-蒲公英云

这篇简单讲asp.net core 中的后台任务

用到的包：

Microsoft.AspNetCore.App metapackage

或者加入

Microsoft.Extensions.Hosting

一. Timed background tasks（定时后台任务）

使用到System.Threading.Timer类。定时器触发任务的DoWork方法。定时器在StopAsync上停止，并且释放是在Dispose上

internal class TimedHostedService : IHostedService, IDisposable
{
    private readonly ILogger _logger;
    private Timer _timer;
    public TimedHostedService(ILogger<TimedHostedService> logger)
    {
        _logger = logger;
    }
    public Task StartAsync(CancellationToken cancellationToken)
    {
        _logger.LogInformation("Timed Background Service is starting.");
        _timer = new Timer(DoWork, null, TimeSpan.Zero, 
            TimeSpan.FromSeconds(5));
        return Task.CompletedTask;
    }
    private void DoWork(object state)
    {
        _logger.LogInformation("Timed Background Service is working.");
    }
    public Task StopAsync(CancellationToken cancellationToken)
    {
        _logger.LogInformation("Timed Background Service is stopping.");
        _timer?.Change(Timeout.Infinite, 0);
        return Task.CompletedTask;
    }
    public void Dispose()
    {
        _timer?.Dispose();
    }
}

服务是在Startup.ConfigureServices上使用AddHostedService扩展方法注册：

services.AddHostedService<TimedHostedService>();

二. Consuming a scoped service in a background task 在后台任务中运行scoped service

使用IHostService中的scoped services, 创建一个scope. 对于一个hosted service默认没有scope被创建。

这个scoped 后台任务服务包含后台任务逻辑。下面的例子中，一个ILogger被注入到了service中：

internal interface IScopedProcessingService
{
    void DoWork();
}
internal class ScopedProcessingService : IScopedProcessingService
{
    private readonly ILogger _logger;
    public ScopedProcessingService(ILogger<ScopedProcessingService> logger)
    {
        _logger = logger;
    }
    public void DoWork()
    {
        _logger.LogInformation("Scoped Processing Service is working.");
    }
}

这个hosted service 创建了一个scope解析了scoped后台任务服务来调用它的DoWork方法：

internal class ConsumeScopedServiceHostedService : IHostedService
{
    private readonly ILogger _logger;
    public ConsumeScopedServiceHostedService(IServiceProvider services, 
        ILogger<ConsumeScopedServiceHostedService> logger)
    {
        Services = services;
        _logger = logger;
    }
    public IServiceProvider Services { get; }
    public Task StartAsync(CancellationToken cancellationToken)
    {
        _logger.LogInformation(
            "Consume Scoped Service Hosted Service is starting.");
        DoWork();
        return Task.CompletedTask;
    }
    private void DoWork()
    {
        _logger.LogInformation(
            "Consume Scoped Service Hosted Service is working.");
        using (var scope = Services.CreateScope())
        {
            var scopedProcessingService = 
                scope.ServiceProvider
                    .GetRequiredService<IScopedProcessingService>();
            scopedProcessingService.DoWork();
        }
    }
    public Task StopAsync(CancellationToken cancellationToken)
    {
        _logger.LogInformation(
            "Consume Scoped Service Hosted Service is stopping.");
        return Task.CompletedTask;
    }
}

服务注册在Startup.ConfigureServices中。IHostedService的实现用AddHostedService扩展方法注册：

services.AddHostedService<ConsumeScopedServiceHostedService>();
services.AddScoped<IScopedProcessingService, ScopedProcessingService>();

三. Queued background tasks 排队的后台任务

public interface IBackgroundTaskQueue
{
    void QueueBackgroundWorkItem(Func<CancellationToken, Task> workItem);
    Task<Func<CancellationToken, Task>> DequeueAsync(
        CancellationToken cancellationToken);
}
public class BackgroundTaskQueue : IBackgroundTaskQueue
{
    private ConcurrentQueue<Func<CancellationToken, Task>> _workItems = 
        new ConcurrentQueue<Func<CancellationToken, Task>>();
    private SemaphoreSlim _signal = new SemaphoreSlim(0);
    public void QueueBackgroundWorkItem(
        Func<CancellationToken, Task> workItem)
    {
        if (workItem == null)
        {
            throw new ArgumentNullException(nameof(workItem));
        }
        _workItems.Enqueue(workItem);
        _signal.Release();
    }
    public async Task<Func<CancellationToken, Task>> DequeueAsync(
        CancellationToken cancellationToken)
    {
        await _signal.WaitAsync(cancellationToken);
        _workItems.TryDequeue(out var workItem);
        return workItem;
    }
}

在 QueueHostedService中，队列中的后台任务出队列并且作为BackroundService执行。BackgroundService是一个实现了IHostedService接口的类。

public class QueuedHostedService : BackgroundService
{
    private readonly ILogger _logger;
    public QueuedHostedService(IBackgroundTaskQueue taskQueue, 
        ILoggerFactory loggerFactory)
    {
        TaskQueue = taskQueue;
        _logger = loggerFactory.CreateLogger<QueuedHostedService>();
    }
    public IBackgroundTaskQueue TaskQueue { get; }
    protected async override Task ExecuteAsync(
        CancellationToken cancellationToken)
    {
        _logger.LogInformation("Queued Hosted Service is starting.");
        while (!cancellationToken.IsCancellationRequested)
        {
            var workItem = await TaskQueue.DequeueAsync(cancellationToken);
            try
            {
                await workItem(cancellationToken);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, 
                   $"Error occurred executing {nameof(workItem)}.");
            }
        }
        _logger.LogInformation("Queued Hosted Service is stopping.");
    }
}

服务注册在Startup.ConfigureService方法中。IHostedService的实现用AddHostedService扩展方法注册：

services.AddHostedService<QueuedHostedService>();
services.AddSingleton<IBackgroundTaskQueue, BackgroundTaskQueue>();

在Index page model类中：

IBackgroundTaskQueue被注入到构造函数并且指定给Queue

一个 IServiceScopeFactory被注入并且指定给_serviceScopeFactory. 这个工厂用来创建IServiceScope实例， IServiceScope实例是用来在scope内创建 services的。一个scope被创建时为了用应用的AppDbContext（a scoped service）来写数据库记录在 IBackgroundTaskQueue 中（a singleton service）.

public class IndexModel : PageModel
{

private readonly AppDbContext _db;
private readonly ILogger _logger;
private readonly IServiceScopeFactory _serviceScopeFactory;
public IndexModel(AppDbContext db, IBackgroundTaskQueue queue, 
    ILogger<IndexModel> logger, IServiceScopeFactory serviceScopeFactory)
{
    _db = db;
    _logger = logger;
    Queue = queue;
    _serviceScopeFactory = serviceScopeFactory;
}
public IBackgroundTaskQueue Queue { get; }

当 Index page 上的Add Task按钮被选中时，OnPostAddTask方法被执行。QueueBackgroundWorkItem被调用来使work item入队。

public IActionResult OnPostAddTaskAsync()
{
    Queue.QueueBackgroundWorkItem(async token =>
    {
        var guid = Guid.NewGuid().ToString();
        using (var scope = _serviceScopeFactory.CreateScope())
        {
            var scopedServices = scope.ServiceProvider;
            var db = scopedServices.GetRequiredService<AppDbContext>();
            for (int delayLoop = 1; delayLoop < 4; delayLoop++)
            {
                try
                {
                    db.Messages.Add(
                        new Message() 
                        { 
                            Text = $"Queued Background Task {guid} has " +
                                $"written a step. {delayLoop}/3"
                        });
                    await db.SaveChangesAsync();
                }
                catch (Exception ex)
                {
                    _logger.LogError(ex, 
                        "An error occurred writing to the " +
                        $"database. Error: {ex.Message}");
                }
                await Task.Delay(TimeSpan.FromSeconds(5), token);
            }
        }
        _logger.LogInformation(
            $"Queued Background Task {guid} is complete. 3/3");
    });
    return RedirectToPage();
}