Worker Service介绍

一、概述

1、概念

2、应用场景

• 处理来自队列、服务总线或事件流的消息、事件
• 响应对象、文件存储中的文件更改
• 聚合数据存储中的数据
• 丰富数据提取管道中的数据
• 可以支持定期的批处理工作负载

二、创建Worker Service

2.1 创建新项目->选择 Worker Service

2.2 项目创建成功之后，会自动创建两个类：Program和Worker

2.2.1 Program.cs

 复制代码
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
namespace WorkerService1
{
    public class Program
    {
        public static void Main(string[] args)
        {
            CreateHostBuilder(args).Build().Run();
        }
        public static IHostBuilder CreateHostBuilder(string[] args) =>
            Host.CreateDefaultBuilder(args)
                .ConfigureServices((hostContext, services) =>
                {
                    services.AddHostedService<Worker>();
                });
    }
}

2.2.2 Worker.cs

 复制代码
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
namespace WorkerService1
{
    public class Worker : BackgroundService
    {
        private readonly ILogger<Worker> _logger;
        public Worker(ILogger<Worker> logger)
        {
            _logger = logger;
        }
        /// <summary>
        /// 重写BackgroundService.ExecuteAsync方法，封装windows服务或linux守护程序中的处理逻辑
        /// </summary>
        /// <param name="stoppingToken"></param>
        /// <returns></returns>
        protected override async Task ExecuteAsync(CancellationToken stoppingToken)
        {
            //如果服务被停止，那么下面的IsCancellationRequested会返回true，我们就应该结束循环
            while (!stoppingToken.IsCancellationRequested)
            {
                //模拟服务中的处理逻辑，这里我们仅输出一条日志，并且等待1秒钟时间
                _logger.LogInformation("Worker running at: {time}", DateTimeOffset.Now);
                await Task.Delay(1000, stoppingToken);
            }
        }
    }
}

2.2.3 重写BackgroundService类的StartAsync、ExecuteAsync、StopAsync方法

 复制代码
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
namespace WorkerService1
{
    public class Worker : BackgroundService
    {
        private readonly ILogger<Worker> _logger;
        public Worker(ILogger<Worker> logger)
        {
            _logger = logger;
        }
        //重写BackgroundService.StartAsync方法，在开始服务的时候，执行一些处理逻辑，这里我们仅输出一条日志
        public override async Task StartAsync(CancellationToken cancellationToken)
        {
            _logger.LogInformation("Worker starting at: {time}", DateTimeOffset.Now);
            await base.StartAsync(cancellationToken);
        }
        /// <summary>
        /// 重写BackgroundService.ExecuteAsync方法，封装windows服务或linux守护程序中的处理逻辑
        /// </summary>
        /// <param name="stoppingToken"></param>
        /// <returns></returns>
        protected override async Task ExecuteAsync(CancellationToken stoppingToken)
        {
            //如果服务被停止，那么下面的IsCancellationRequested会返回true，我们就应该结束循环
            while (!stoppingToken.IsCancellationRequested)
            {
                //模拟服务中的处理逻辑，这里我们仅输出一条日志，并且等待1秒钟时间
                _logger.LogInformation("Worker running at: {time}", DateTimeOffset.Now);
                await Task.Delay(1000, stoppingToken);
            }
        }
        //重写BackgroundService.StopAsync方法，在结束服务的时候，执行一些处理逻辑，这里我们仅输出一条日志
        public override async Task StopAsync(CancellationToken cancellationToken)
        {
            _logger.LogInformation("Worker stopping at: {time}", DateTimeOffset.Now);
            await base.StopAsync(cancellationToken);
        }
    }
}

2.2.4 避免线程阻塞

 复制代码
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
    while (!stoppingToken.IsCancellationRequested)
    {
        _logger.LogInformation("Worker running at: {time}", DateTimeOffset.Now);
        Thread.Sleep(1000);//使用Thread.Sleep进行同步等待，调用ExecuteAsync方法的线程会一直执行这里的循环，被不停地被阻塞
    }
    await Task.CompletedTask;
}

 复制代码
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
namespace WorkerService1
{
    public class Worker : BackgroundService
    {
        private readonly ILogger<Worker> _logger;
        public Worker(ILogger<Worker> logger)
        {
            _logger = logger;
        }
        //重写BackgroundService.StartAsync方法，在开始服务的时候，执行一些处理逻辑，这里我们仅输出一条日志
        public override async Task StartAsync(CancellationToken cancellationToken)
        {
            _logger.LogInformation("Worker starting at: {time}", DateTimeOffset.Now);
            await base.StartAsync(cancellationToken);
        }
        //第一个 windows服务或linux守护程序 的处理逻辑，由RunTaskOne方法内部启动的Task任务线程进行处理，同样可以从参数CancellationToken stoppingToken中的IsCancellationRequested属性，得知Worker Service服务是否已经被停止
        protected Task RunTaskOne(CancellationToken stoppingToken)
        {
            return Task.Run(() =>
            {
                //如果服务被停止，那么下面的IsCancellationRequested会返回true，我们就应该结束循环
                while (!stoppingToken.IsCancellationRequested)
                {
                    _logger.LogInformation("RunTaskOne running at: {time}", DateTimeOffset.Now);
                    Thread.Sleep(1000);
                }
            }, stoppingToken);
        }
        //第二个 windows服务或linux守护程序 的处理逻辑，由RunTaskTwo方法内部启动的Task任务线程进行处理，同样可以从参数CancellationToken stoppingToken中的IsCancellationRequested属性，得知Worker Service服务是否已经被停止
        protected Task RunTaskTwo(CancellationToken stoppingToken)
        {
            return Task.Run(() =>
            {
                //如果服务被停止，那么下面的IsCancellationRequested会返回true，我们就应该结束循环
                while (!stoppingToken.IsCancellationRequested)
                {
                    _logger.LogInformation("RunTaskTwo running at: {time}", DateTimeOffset.Now);
                    Thread.Sleep(1000);
                }
            }, stoppingToken);
        }
        //第三个 windows服务或linux守护程序 的处理逻辑，由RunTaskThree方法内部启动的Task任务线程进行处理，同样可以从参数CancellationToken stoppingToken中的IsCancellationRequested属性，得知Worker Service服务是否已经被停止
        protected Task RunTaskThree(CancellationToken stoppingToken)
        {
            return Task.Run(() =>
            {
                //如果服务被停止，那么下面的IsCancellationRequested会返回true，我们就应该结束循环
                while (!stoppingToken.IsCancellationRequested)
                {
                    _logger.LogInformation("RunTaskThree running at: {time}", DateTimeOffset.Now);
                    Thread.Sleep(1000);
                }
            }, stoppingToken);
        }
        //重写BackgroundService.ExecuteAsync方法，封装windows服务或linux守护程序中的处理逻辑
        protected override async Task ExecuteAsync(CancellationToken stoppingToken)
        {
            try
            {
                Task taskOne = RunTaskOne(stoppingToken);
                Task taskTwo = RunTaskTwo(stoppingToken);
                Task taskThree = RunTaskThree(stoppingToken);
                await Task.WhenAll(taskOne, taskTwo, taskThree);//使用await关键字，异步等待RunTaskOne、RunTaskTwo、RunTaskThree方法返回的三个Task对象完成，这样调用ExecuteAsync方法的线程会立即返回，不会卡在这里被阻塞
            }
            catch (Exception ex)
            {
                //RunTaskOne、RunTaskTwo、RunTaskThree方法中，异常捕获后的处理逻辑，这里我们仅输出一条日志
                _logger.LogError(ex.Message);
            }
            finally
            {
                //Worker Service服务停止后，如果有需要收尾的逻辑，可以写在这里
            }
        }
        //重写BackgroundService.StopAsync方法，在结束服务的时候，执行一些处理逻辑，这里我们仅输出一条日志
        public override async Task StopAsync(CancellationToken cancellationToken)
        {
            _logger.LogInformation("Worker stopping at: {time}", DateTimeOffset.Now);
            await base.StopAsync(cancellationToken);
        }
    }
}

2.2.5 在Worker Service中运行多个Worker类

 复制代码
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using System;
using System.Threading;
using System.Threading.Tasks;
namespace WorkerService1
{
    public class WorkerOne:BackgroundService
    {
        private readonly ILogger<WorkerOne> _logger;
        public WorkerOne(ILogger<WorkerOne> logger)
        {
            _logger = logger;
        }
        //重写BackgroundService.StartAsync方法，在开始服务的时候，执行一些处理逻辑，这里我们仅输出一条日志
        public override async Task StartAsync(CancellationToken cancellationToken)
        {
            _logger.LogInformation("WorkerOne starting at: {time}", DateTimeOffset.Now);
            await base.StartAsync(cancellationToken);
        }
        //重写BackgroundService.ExecuteAsync方法，封装windows服务或linux守护程序中的处理逻辑
        protected override async Task ExecuteAsync(CancellationToken stoppingToken)
        {
            //如果服务被停止，那么下面的IsCancellationRequested会返回true，我们就应该结束循环
            while (!stoppingToken.IsCancellationRequested)
            {
                //模拟服务中的处理逻辑，这里我们仅输出一条日志，并且等待1秒钟时间
                _logger.LogInformation("WorkerOne running at: {time}", DateTimeOffset.Now);
                await Task.Delay(1000, stoppingToken);
            }
        }
        //重写BackgroundService.StopAsync方法，在结束服务的时候，执行一些处理逻辑，这里我们仅输出一条日志
        public override async Task StopAsync(CancellationToken cancellationToken)
        {
            _logger.LogInformation("WorkerOne stopping at: {time}", DateTimeOffset.Now);
            await base.StopAsync(cancellationToken);
        }
    }
}

 复制代码
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using System;
using System.Threading;
using System.Threading.Tasks;
namespace WorkerService1
{
    public class WorkerTwo : BackgroundService
    {
        private readonly ILogger<WorkerTwo> _logger;
        public WorkerTwo(ILogger<WorkerTwo> logger)
        {
            _logger = logger;
        }
        //重写BackgroundService.StartAsync方法，在开始服务的时候，执行一些处理逻辑，这里我们仅输出一条日志
        public override async Task StartAsync(CancellationToken cancellationToken)
        {
            _logger.LogInformation("WorkerTwo starting at: {time}", DateTimeOffset.Now);
            await base.StartAsync(cancellationToken);
        }
        //重写BackgroundService.ExecuteAsync方法，封装windows服务或linux守护程序中的处理逻辑
        protected override async Task ExecuteAsync(CancellationToken stoppingToken)
        {
            //如果服务被停止，那么下面的IsCancellationRequested会返回true，我们就应该结束循环
            while (!stoppingToken.IsCancellationRequested)
            {
                //模拟服务中的处理逻辑，这里我们仅输出一条日志，并且等待1秒钟时间
                _logger.LogInformation("WorkerTwo running at: {time}", DateTimeOffset.Now);
                await Task.Delay(1000, stoppingToken);
            }
        }
        //重写BackgroundService.StopAsync方法，在结束服务的时候，执行一些处理逻辑，这里我们仅输出一条日志
        public override async Task StopAsync(CancellationToken cancellationToken)
        {
            _logger.LogInformation("WorkerTwo stopping at: {time}", DateTimeOffset.Now);
            await base.StopAsync(cancellationToken);
        }
    }
}

 复制代码
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
namespace WorkerService1
{
    public class Program
    {
        public static void Main(string[] args)
        {
            CreateHostBuilder(args).Build().Run();
        }
        public static IHostBuilder CreateHostBuilder(string[] args) =>
            Host.CreateDefaultBuilder(args)
                .ConfigureServices((hostContext, services) =>
                {
                    services.AddHostedService<WorkerOne>();
                    services.AddHostedService<WorkerTwo>();
                    //services.AddHostedService<Worker>();
                });
    }
}

三、部署为Windows服务运行

• 在项目中添加nuget包：Microsoft.Extensions.Hosting.WindowsServices

• 在program.cs内部，将UseWindowsService()添加到CreateHostBuilder

 复制代码
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
namespace WorkerService1
{
    public class Program
    {
        public static void Main(string[] args)
        {
            CreateHostBuilder(args).Build().Run();
        }
        public static IHostBuilder CreateHostBuilder(string[] args) =>
            Host.CreateDefaultBuilder(args)
                .UseWindowsService()
                .ConfigureServices((hostContext, services) =>
                {
                    services.AddHostedService<Worker>();
                });
    }
}

• 执行一下命令发布项目

 复制代码
dotnet publish -c Release -o D:\PersonalInfo\windowsservices\WindowsService1\WorkerService1\Release

• 默认情况下Worker Service项目会被发布为一个exe文件：

• 使用sc.exe工具来管理服务，输入命令创建为windows服务（Run as administrator）启动Powershell：

 复制代码
sc.exe create NETCoreWorkerService1 binPath=D:\PersonalInfo\windowsservices\WindowsService1\WorkerService1\bin\Release\net5.0\publish\WorkerService1.exe

• 查看服务状态，在powershell中执行（Run as administrator）：

 复制代码
sc.exe query NETCoreWorkerService1

• 启动命令，在powershell中执行（Run as administrator）：

• 在windows服务列表查看，NETCoreWorkerService1已安装成功：

• 停用 、删除命令：

 复制代码
sc.exe stop NETCoreWorkerService1 
sc.exe delete NETCoreWorkerService1

四、部署作为Linux守护程序运行

• 添加Microsoft.Extensions.Hosting.Systemd NuGet包到项目中
• 将UseSystemd()添加到主机构建器中

 复制代码
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
namespace WorkerService1
{
    public class Program
    {
        public static void Main(string[] args)
        {
            CreateHostBuilder(args).Build().Run();
        }
        public static IHostBuilder CreateHostBuilder(string[] args) =>           
            Host.CreateDefaultBuilder(args)
                //.UseWindowsService()
                .UseSystemd()
                .ConfigureServices((hostContext, services) =>
                {
                    services.AddHostedService<Worker>();
                });
    }
}

在 Windows 平台上调用 UseSystemd 方法也是不会报错的，Windows 平台会忽略此调用。具体如何添加守护进程可以参考https://www.cnblogs.com/qtiger/p/13853828.html

五、参考资料

• https://www.cnblogs.com/OpenCoder/p/12191164.html
• https://docs.microsoft.com/en-us/dotnet/architecture/microservices/multi-container-microservice-net-applications/background-tasks-with-ihostedservice#implementing-ihostedservice-with-a-custom-hosted-service-class-deriving-from-the-backgroundservice-base-class