zoukankan      html  css  js  c++  java

  • 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服务运行

    • 在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:PersonalInfowindowsservicesWindowsService1WorkerService1Release

    在Powershell中执行：

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

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

    • 使用sc.exe工具来管理服务，输入命令创建为windows服务（Run as administrator）启动Powershell：
    sc.exe create NETCoreWorkerService1 binPath=D:PersonalInfowindowsservicesWindowsService1WorkerService1inRelease
et5.0publishWorkerService1.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守护程序运行

    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

    五、参考资料
  • 相关阅读:
    Nginx实现动静分离
    Nginx实现负载均衡
    Linux安装Kafka
    Linux安装RocketMQ
    转://ORA-00603,ORA-27501,ORA-27300,ORA-27301,ORA-27302故障案例一则
    CPU, PSU, SPU
    转://Oracle打补丁方法论
    转://Oracle数据库补丁分析实践
    转://Linux Multipath多路径配置与使用案例
    转://Oracle中定义者权限和调用者权限案例分析
  • 原文地址：https://www.cnblogs.com/qtiger/p/14814556.html
Copyright © 2011-2022 走看看