Worker Service

本文转载自查看原文 2021-05-26 18:04 177 Worker Service/ 定时服务

一、概述

1、概念

Worker Service 是使用模板构建的 .NET 项目，在VS2019中可以找到。Worker Service可以用来编写长时间运行的后台服务，并且能部署成windows服务或linux守护程序。Worker Service 没有用户界面，也不支持直接的用户交互，它们特别适用于设计微服务架构。在微服务体系结构中，职责通常被划分为不同的、可单独部署的、可伸缩的服务。随着微服务架构的成长和发展，拥有大量的 Worker Service 会变得越来越常见。

2、应用场景

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

二、创建Worker Service

创建新项目->选择 Worker Service

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

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>();
                });
    }
}

Program类跟ASP.NET Core Web应用程序非常类似，不同之处没有了startup类，并且把worker服务添加到DI container中。

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);
            }
        }
    }
}

worker继承自BackgroundService ，而后者又实现IHostedService接口。worker类的构造函数中，使用了.NET Core自带的日志组件接口对象ILogger，它是通过DI（依赖注入）注入到worker类的构造函数中的。ExecuteAsync 方法用来完成相应的逻辑，该方法实际上属于BackgroundService类，可以在worker类中重写（override）它。通过ExecuteAsync方法传入的CancellationToken参数对象，来判断是否应该结束循环，例如如果windows服务被停止，那么参数中CancellationToken类的IsCancellationRequested属性会返回true，ExecuteAsync方法就会停止循环，来结束整个windows服务。

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

　　我们也可以在worker类中重写BackgroundService.StartAsync方法和BackgroundService.StopAsync方法，注意重写时，不要忘记在worker类中调用base.StartAsync和base.StopAsync，因为BackgroundService类的StartAsync和StopAsync会执行一些Worker Service的核心代码，在开始和结束Worker Service服务（例如开始和停止windows服务）的时候，来执行一些处理逻辑，本例中我们分别输出了一条日志：

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);
        }
    }
}

由于BackgroundService类的StartAsync、ExecuteAsync、StopAsync方法返回的都是Task类型，我们可以使用async和await关键字将它们重写为异步函数，来提高程序的性能。运行结果如下所示，每隔1秒循环打印运行的时间（可以在启动的控制台中使用快捷键"Ctrl+C"来停止Worker Service的运行，相当于停止windows服务或linux守护程序）：

从下图可以看到Worker Service项目从本质上来说就是一个控制台项目，只不过当它被部署为windows服务或linux守护程序后，不会显示控制台窗口。所以实际上在Visual Studio中进行调试的时候，完全可以用Console.WriteLine等控制台方法来替代ILogger接口的日志输出方法，不过由于ILogger接口的日志输出方法也可以记录到文件等媒介上，还是更推荐使用ILogger接口来输出调试信息。

避免线程阻塞

　　不要让线程阻塞worker类中重写的StartAsync、ExecuteAsync、StopAsync方法，因为StartAsync方法负责启动Worker Service，如果调用StartAsync方法的线程被一直阻塞了，那么Worker Service的启动就一直完成不了。同理StopAsync方法负责结束Worker Service，如果调用StopAsync方法的线程被一直阻塞了，那么Worker Service的结束就一直完成不了。这里主要说明下为什么ExecuteAsync方法不能被阻塞，我们尝试把本例中的ExecuteAsync方法改为如下代码：

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;
}

我们将ExecuteAsync方法中的异步等待方法Task.Delay，改为了同步等待方法Thread.Sleep。由于Thread.Sleep方法是将执行线程通过阻塞的方式来进行等待，所以现在调用ExecuteAsync方法的线程会一直执行ExecuteAsync方法中的循环，被不停地被阻塞，除非ExecuteAsync方法中的循环结束，那么调用ExecuteAsync方法的线程会被一直卡在ExecuteAsync方法中。现在我们在Visual Studio中运行Worker Service，执行结果如下：

我们可以看到当我们在控制台中使用快捷键"Ctrl+C"试图停止Worker Service后（上图红色框中输出的日志），ExecuteAsync方法中的循环还是在不停地运行来输出日志，这说明ExecuteAsync方法的CancellationToken参数的IsCancellationRequested属性还是返回的false，所以这就是问题所在，如果我们直接用调用ExecuteAsync方法的线程去做循环，来执行windows服务或linux守护程序的处理逻辑，会导致Worker Service无法被正常停止，因为ExecuteAsync方法的CancellationToken参数没有被更新。所以，那些很耗时并且要循环处理的 windows服务或linux守护程序的处理逻辑，应该要放到另外的线程中去执行，而不是由调用ExecuteAsync方法的线程去执行。所以假设我们现在有三个windows服务或linux守护程序的逻辑现在要被处理，我们可以将它们放到三个新的线程中去执行，如下代码所示：

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);
        }
    }
}

所以现在调用ExecuteAsync方法的线程就不会被阻塞了，执行结果如下：

可以看到这次，当我们在控制台中使用快捷键"Ctrl+C"试图停止Worker Service后，ExecuteAsync方法就立即停止运行了，所以这里再次强调千万不要去阻塞调用ExecuteAsync方法的线程！另外上面代码中，我们在worker类重写的ExecuteAsync方法中放了一个finally代码块，这个代码块可以用来执行一些Worker Service服务停止后的一些收尾代码逻辑（例如关闭数据库连接、释放资源等），我更倾向于使用ExecuteAsync方法中的finally代码块来做Worker Service的收尾工作，而不是在worker类重写的StopAsync方法中来做收尾工作，从BackgroundService的源代码，我们可以看出worker类的StopAsync方法是有可能比ExecuteAsync方法先完成的，所以Worker Service的收尾工作应该放到ExecuteAsync方法中的finally代码块，因为ExecuteAsync方法中的finally代码块，肯定是在RunTaskOne、RunTaskTwo、RunTaskThree方法返回的三个Task对象执行完毕后才执行的。

在Worker Service中运行多个Worker类

　　在前面的例子中，可以看到我们在一个Worker类中定义了三个方法RunTaskOne、RunTaskTwo、RunTaskThree，来执行三个 windows服务或linux守护程序的逻辑。其实我们还可以在一个Worker Service项目中，定义和执行多个Worker类，而不是把所有的代码逻辑都放在一个Worker类中。首先我们定义第一个Worker类WorkerOne：

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);
        }
    }
}

接着我们定义第二个Worker类WorkerTwo：

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);
        }
    }
}

然后我们在Program类中，将WorkerOne和WorkerTwo服务添加到DI container中：

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>();
                });
    }
}

然后在Visual Studio中运行Worker Service，执行结果如下：

可以看到WorkerOne和WorkerTwo类都被执行了，并且都输出了日志信息

三、部署为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>();
                });
    }
}

注意在非 Windows 平台上调用 UseWindowsService 方法也是不会报错的，非 Windows 平台会忽略此调用。

执行一下命令发布项目

dotnet publish -c Release -o D:\PersonalInfo\windowsservices\WindowsService1\WorkerService1\Release

在Powershell中执行：

也可以在Visual Studio中用项目自身的发布向导来将Worker Service项目发布到文件夹"bin\Release\net5.0\publish\"中：

默认情况下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）：

sc.exe start NETCoreWorkerService1

在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

五、参考资料

免责声明！

本站转载的文章为个人学习借鉴使用，本站对版权不负任何法律责任。如果侵犯了您的隐私权益，请联系本站邮箱yoyou2525@163.com删除。

猜您在找 Service Worker,Web Worker,WebSocket的对比 Worker Service in ASP .NET Core .NET Core中的Worker Service 基于.Net Core 5.0 Worker Service 的 Quart 服务 .Net core Worker Service 扩展库 .NET 中的 Worker Service 入门介绍 .NET 6学习笔记(2)——通过Worker Service创建Windows Service 在 .NET Core 中使用 Worker Service（辅助角色服务）渐进式web应用开发---Service Worker 与页面通信(七) 渐进式web应用开发---service worker 原理及介绍(一)