.net core 任务调度

标签：core task taskModel System using net 任务调度 public

    任务调度在项目开发中，已经变得越来越常见，每个项目一般都会出现几个需要定时程序来完成的，有的选择直接与web网站一起运行，有的选择将任务调度作为服务单独运行。还有的直接做一个任务调度中心，所有任务都放在任务调度中心中统一管理。

    关于任务调度，以前也介绍过，Quartz任务调度：（https://www.cnblogs.com/zpy1993-09/p/15548372.html）这次打算做一个通用点的任务调度，也就是任务调度中心来统一管理任务调度。Quartz任务调度用起来还是非常不错的。但是缺点还是有的，当然也是大多任务调度插件都出现的。

    1，每增加一个任务调度都要自己一个执行逻辑。无论是在web平台上还是单独作为服务运行，只要添加新的任务调度，都要重新修改发布。

    2，不能够很好的兼容任何项目，比如A平台是.net 开发的，B平台是Java开发的，那么如果任务调度要统一管理，那么必然很麻烦。

    3，如果项目越来越多，可能出现C平台，D平台　那么所有的执行逻辑都要写在任务调度中心中，也会显得很混乱。

 大家对服务注册和发现，都应该很熟悉，例如Consul ,ZooKeeper等，就拿Consul来说吧，不管什么平台，你只需注册服务，Consul会管理好我们每个服务包括服务下的多个实例。所以我想做任务调度的时候能不能不在管执行逻辑，把执行逻辑抛给需要添加任务的一方，让他们自己实现逻辑，而任务调度中心只负责管理什么时候执行。

   有了思路，我们可以跟着思路慢慢走，设想一下，如果A平台要实现一个任务调度，而任务调度中心又不想自己管理执行逻辑，要怎么做？

  

 

上图的意思就是：

 1，任务调度中心向外暴漏API接口，API接口包含注册任务，停止任务，启动任务，修改任务等接口

 2，通过web调用任务调度中心的添加任务向调度中心注册任务。

3，注册方在注册任务的时候提供一个api接口，定时任务的执行逻辑都写在接口里面。然后把接口作为任务信息注册到任务调度中心。

4，任务调度中心会按照任务信息给的执行频率去定时调用注册方注册的接口。(任务信息中会包含一个唯一的Id 用来识别当前任务）

按照上面的设想，我们基本可以处理到任务调度的有些弊端，做到不同项目真正的任务统一管理，并且不需要对任务调度中i心尽心代码更改，频繁发布。

说干就干，这里我用.net core 控制台应用程序来实现这个过程。目前已经初步实现了，但是由于东西也不少，我做了服务端的任务调度中心和客户端如何调用后续作为dll文件引入项目，所以这里就把实现的难点说一下。

第一个要面临的问题就是在.net core控制条中内嵌一个APi接口类用来外部调用。

 

 

第二个问题就是 任务调度插件一般执行逻辑都要放在一个执行类或者执行方法中，比如Quartz任务调度，每执行一个任务都要建一个执行类还要继承Quartz插件的IJob。然后在注册，才能跑起来。

第一个问题稍微容易，只要对.net core熟悉，基本都能做：

首先仿照webApi的形式进行配置：首先新建一个startup类：

using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Http;
using Microsoft.AspNetCore.Mvc;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using System;
using System.Collections.Generic;
using System.Text;


namespace HK.Jobs.WebApi
{
  public class Startup
  {
   

    public void ConfigureServices(IServiceCollection services)
    {
           
            //配置Mvc + json 序列化
            services.AddMvc();

    }

    public void Configure(IApplicationBuilder app)
    {
           
            app.UseMvc(routes =>
           {
             routes.MapRoute(
                name: "default",
                template: "{controller}/{action}/{id?}");
           });


        }
  }
}

　　然后在程序入口中注册：

using HK.Jobs.Common;
using HK.Jobs.Models.Tasks;
using HK.Jobs.Scheduler.SchedulerCenter;
using HK.Jobs.WebApi;
using Microsoft.AspNetCore;
using Microsoft.AspNetCore.Hosting;
using Microsoft.Extensions.Configuration;
using Quartz;
using System;
using System.Collections.Generic;
using System.IO;
using System.Reflection;
using System.Reflection.Emit;

namespace HK.Jobs
{
    class Program
    {
        static void Main(string[] args)
        {
//在webapi未注册之前开启任务调度（这里在程序启动的时候开启任务调度）
             var res=  SchedulerHelper.StartScheduleAsync();
            ///下面是仿照webapi形式
            var host = WebHost.CreateDefaultBuilder(args)
                  .UseKestrel()
                  .UseContentRoot(Directory.GetCurrentDirectory())
                   .ConfigureAppConfiguration((hostContext, configApp) =>
                   {

                       configApp.AddJsonFile("appsettings.json", true);///该文件自己建

                   })
                        .UseUrls("http://localhost::8080")//自己定义ip和端口
                        .UseStartup<Startup>()
                        .Build();

            host.Run();
           


        }
    }
}

 新建一个Controlle类：

namespace HK.Task.WebApi.Controllers
{

    public class TaskSetController:Controller
  {
    
        public string index()
        {
            return "Hello!你好。";
        }

   
    }
}

那么跑起来看一下：

 

 执行api：

 

 显然，没有任何问题。后续只需要写对外交互的任务接口就行了。

那么难点就在第二个了，这个任务调度的执行类是一个难点，因为如何定时任务的执行逻辑抛给注册者的话，那么这个执行类就不确定了，可能是一个，可能是10个。如果每添加一个任务我们还要自己在建一个执行类，那和以前有什么区别，不还是要修改发布，那执行逻辑抛给注册者还有什么用。以前我的思路就是一起建个20个执行类task_1~Task_20，如果有添加注册的，可以从这个20个实例中随机拿出一个没有被只用的执行类的类名，然后通过反射注册到quartz中，但是这有个限制，就是任务调度的个数，如果你建20个执行类，那么任务调度的上限就是20个。这样有点太苦逼了。

 

 那么如何解决这个问题呢，那就是动态创建一个任务类，动态继承IJob接口，动态向quartz注册。那么问题解决了，外部添加一个一个任务，我就创建一个随机类，然后注册到quartz中执行。这样就没有限制了。我们先看一下这个类的结构：首先是继承IJob，然后实现IJob的Excute类。那么如何动态去创建这个执行类呢。

   1，手写IL（这个有点困难，没有接触过的实在难搞）

   2，基于Roslyn实现代码动态编译。(微软自带的，可以直接引入包。比起IL容易太多）

  3，使用 Natasha 这个是基于Roslyn别人封装好的，相对于Roslyn可能又简单了一些。

  4，CodeAnalysis 基于Roslyn的代码分析器。

目前我使用的是CodeAnalysis来实现动态代码。不多说了，搞简单一点，把整个流程跑起来，直接上代码：

  首先是任务详情的model类：这里面有邮件信息和健康检查的可以忽略，因为只是简单测试，如果把真个工程说一下，太多了。

using System;
using System.Collections.Generic;
using System.Text;

namespace HK.Jobs.Models.Tasks
{
    public class QZTaskModel
    {
        /// <summary>
        /// 执行传参
        /// </summary>
        public string JobParams { get; set; }

        public string HttpRequest { get; set; }
        public TaskModel taskModel { get; set; } = new TaskModel();
    }
}

　　

using System;
using System.Collections.Generic;
using System.Text;

namespace HK.Jobs.Models.Tasks
{
    public class TaskMessage
    {
        /// <summary>
        /// 是否对任务执行接口进行健康检查
        /// </summary>
        public bool isHealth { get; set; } = false;
        /// <summary>
        /// 是否使用邮箱发送
        /// </summary>
        public bool isEmail { get; set; } = false;
        /// <summary>
        ///接口信息
        /// </summary>
        public object http { get; set; }
        /// <summary>
        /// 任务信息
        /// </summary>
        public object task { get; set; }
        /// <summary>
        /// 邮件信息
        /// </summary>
        public object eamil { get; set; }
    }
}

using System;
using System.Collections.Generic;
using System.Text;

namespace HK.Jobs.Models.Tasks
{
    /// <summary>
    /// 任务执行实体
    /// </summary>
    public class TaskModel
    {
        /// <summary>
        /// 本次执行任务的Id，后续其它客户端与该服务交互
        /// 来识别某个客户端开启的是哪个任务
        /// </summary>
        public string TaskId { get; set; }
        /// <summary>
        /// 任务名称：例如你要建一个任务邮件发送，那你就可以起一个邮件发送的任务
        /// </summary>
        public string TaskName { get; set; }
        /// <summary>
        /// 类名字符串
        /// </summary>
        public string ClassName { get; set; }
        /// <summary>
        /// 组名
        /// </summary>
        public string TaskGroup { get; set; }
        /// <summary>
        /// 任务的描述
        /// </summary>
        public string TaskInfo { get; set; }
        /// <summary>
        /// 任务运行时间表达式
        /// </summary>
        public string Cron { get; set; }
        /// <summary>
        /// 执行次数
        /// </summary>
        public int RunTimes { get; set; }
        /// <summary>
        /// 开始时间
        /// </summary>
        public DateTime? BeginTime { get; set; }
        /// <summary>
        /// 结束时间
        /// </summary>
        public DateTime? EndTime { get; set; }
        /// <summary>
        /// 触发器类型（0、simple 1、cron）
        /// </summary>
        public int TriggerType { get; set; }
        /// <summary>
        /// 执行间隔时间, 秒为单位
        /// </summary>
        public int IntervalSecond { get; set; }

       



    }
}

添加任务调度的方法：

        /// <summary>
        /// 添加一个计划任务（映射程序集指定IJob实现类）
        /// </summary>
        /// <param name="task"></param>
        /// <returns></returns>
        public static async Task<MessageModel<string>> AddScheduleJobAsync(QZTaskModel task)
        {
            var result = new MessageModel<string>();
            try
            {
                if (task != null)
                {
                    JobKey jobKey = new JobKey(task.taskModel.TaskId, task.taskModel.TaskGroup);
                    if (await _scheduler.Result.CheckExists(jobKey))
                    {
                        result.success = false;
                        result.msg = $"该任务计划已经在执行:【{task.taskModel.TaskName}】,请勿重复启动！";
                        return result;
                    }
                    #region 设置开始时间和结束时间

                    if (task.taskModel.BeginTime == null)
                    {
                        task.taskModel.BeginTime = DateTime.Now;
                    }
                    DateTimeOffset starRunTime = DateBuilder.NextGivenSecondDate(task.taskModel.BeginTime, 1);//设置开始时间
                    if (task.taskModel.EndTime == null)
                    {
                        task.taskModel.EndTime = DateTime.MaxValue.AddDays(-1);
                    }
                    DateTimeOffset endRunTime = DateBuilder.NextGivenSecondDate(task.taskModel.EndTime, 1);//设置暂停时间

                    #endregion

                    #region 通过反射获取程序集类型和类   

                    // Assembly assembly = Assembly.Load(new AssemblyName("HK.Jobs"));
                    //Type jobType = assembly.GetType("HK.Jobs.Scheduler.Jobs." + task.taskModel.ClassName);

                    #endregion
                    //动态构建任务调度类的类型
                    Type jobType = CodeHelper.BuildType(task.taskModel.ClassName);
                    //判断任务调度是否开启
                    if (!_scheduler.Result.IsStarted)
                    {
                        await StartScheduleAsync();
                    }

                    //传入反射出来的执行程序集
                    IJobDetail job = new JobDetailImpl(task.taskModel.TaskId, task.taskModel.TaskGroup, jobType);
                    job.JobDataMap.Add(task.JobParams, task.HttpRequest);
                    ITrigger trigger;

                    #region 泛型传递
                    //IJobDetail job = JobBuilder.Create<T>()
                    //    .WithIdentity(sysSchedule.Name, sysSchedule.JobGroup)
                    //    .Build();
                    #endregion

                    if (task.taskModel.Cron != null && CronExpression.IsValidExpression(task.taskModel.Cron) && task.taskModel.TriggerType > 0)
                    {
                        trigger = CreateCronTrigger(task);
                    }
                    else
                    {
                        trigger = CreateSimpleTrigger(task);
                    }
                    // 告诉Quartz使用我们的触发器来安排作业
                    await _scheduler.Result.ScheduleJob(job, trigger);
             
                    //await Task.Delay(TimeSpan.FromSeconds(120));
                    //await Console.Out.WriteLineAsync("关闭了调度器！");
                    //await _scheduler.Result.Shutdown();
                    result.success = true;
                    result.msg = $"启动任务:【{task.taskModel.TaskName}】成功";
                    return result;
                }
                else
                {
                    result.success = false;
                    result.msg = $"任务计划不存在:【{task.taskModel.TaskName}】";
                    return result;
                }
            }
            catch (Exception ex)
            {
                throw ex;
            }
        }
  #region 创建触发器帮助方法

        /// <summary>
        /// 创建SimpleTrigger触发器（简单触发器）
        /// </summary>
        /// <param name="sysSchedule"></param>
        /// <param name="starRunTime"></param>
        /// <param name="endRunTime"></param>
        /// <returns></returns>
        private static ITrigger CreateSimpleTrigger(QZTaskModel task)
        {
            if (task.taskModel.RunTimes > 0)
            {
                ITrigger trigger = TriggerBuilder.Create()
                .WithIdentity(task.taskModel.TaskId, task.taskModel.TaskGroup)
                .StartAt(task.taskModel.BeginTime.Value)
                .EndAt(task.taskModel.EndTime.Value)
                .WithSimpleSchedule(x =>
                x.WithIntervalInSeconds(task.taskModel.IntervalSecond)
                .WithRepeatCount(task.taskModel.RunTimes)).ForJob(task.taskModel.TaskId, task.taskModel.TaskGroup).Build();
                return trigger;
            }
            else
            {
                ITrigger trigger = TriggerBuilder.Create()
                .WithIdentity(task.taskModel.TaskId, task.taskModel.TaskGroup)
                .StartAt(task.taskModel.BeginTime.Value)
                .EndAt(task.taskModel.EndTime.Value)
                .WithSimpleSchedule(x =>
                x.WithIntervalInSeconds(task.taskModel.IntervalSecond)
                .RepeatForever()).ForJob(task.taskModel.TaskId, task.taskModel.TaskGroup).Build();
                return trigger;
            }
            // 触发作业立即运行，然后每10秒重复一次，无限循环

        }
        /// <summary>
        /// 创建类型Cron的触发器
        /// </summary>
        /// <param name="m"></param>
        /// <returns></returns>
        private static ITrigger CreateCronTrigger(QZTaskModel task)
        {
            // 作业触发器
            return TriggerBuilder.Create()
                   .WithIdentity(task.taskModel.TaskId, task.taskModel.TaskGroup)
                   .StartAt(task.taskModel.BeginTime.Value)//开始时间
                   .EndAt(task.taskModel.EndTime.Value)//结束数据
                   .WithCronSchedule(task.taskModel.Cron)//指定cron表达式
                   .ForJob(task.taskModel.TaskId, task.taskModel.TaskGroup)//作业名称
                   .Build();
        }


        #endregion

　　动态构建任务调度执行类：

using Microsoft.CodeAnalysis;
using Microsoft.CodeAnalysis.CSharp;
using Quartz;
using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Reflection;
using System.Text;

namespace HK.Jobs.Common
{
    /// <summary>
    /// 动态创建类，动态继承任务调度IJob接口，动态编译执行。
    /// </summary>
   public  class CodeHelper
    {
        /// <summary>
        /// 动态构建任务调度执行类的执行脚本
        /// </summary>
        /// <param name="classname">类名：注意类名自己在类名符合范围内随机定义，保证唯一</param>
        /// <returns></returns>
        public static string stringclass(string classname)
        {
            string strcalssname = "";
            //strcalssname += "using Quartz;";
            strcalssname += "using System;";
            strcalssname += "namespace HK.Jobs.Scheduler.Jobs";
            strcalssname += "{";
            strcalssname += " public class "+classname+ " : Quartz.IJob";
            strcalssname += "{";
            strcalssname += " public System.Threading.Tasks.Task Execute(Quartz.IJobExecutionContext context)";
            strcalssname += "{";
            strcalssname += " string name = context. JobDetail.JobDataMap[\"" + classname + "\"].ToString();";
            strcalssname += "System.Console.WriteLine(@name);";
            strcalssname += "using (var client = new System.Net.Http.HttpClient())";
            strcalssname += " {";
           // strcalssname += "client.DefaultRequestHeaders.Add(\"Method\", \"Get\");";
            strcalssname += "var result=client.GetAsync(name).Result.Content.ReadAsStringAsync();";
            strcalssname += "System.Console.WriteLine(@result);";
            strcalssname += "}";
            strcalssname += " return System.Threading.Tasks.Task.CompletedTask;";
            strcalssname += "}";
            strcalssname += "}";
            strcalssname += "}";
            return strcalssname;
        }
        /// <summary>
        /// 构建Type类型
        /// </summary>
        /// <param name="classname"></param>
        /// <returns></returns>
        public static Type BuildType(string classname)
        {
            var text = stringclass(classname);
            // 分析语法树
            var syntaxTree = CSharpSyntaxTree.ParseText(text, new CSharpParseOptions(LanguageVersion.Latest));
       
            // 配置引用
            var references = new[]
            {

            typeof(object).Assembly,
               Assembly.Load("Quartz"),
               Assembly.Load("System"),
               Assembly.Load("System.Threading"),
               Assembly.Load("netstandard"),
               Assembly.Load("System.Runtime"),
               Assembly.Load("System.Console"),
               Assembly.Load("System.Net.Http"),
               Assembly.Load("System.Private.Uri")
              }

            .Select(assembly => assembly.Location)
                .Distinct()
                .Select(l => MetadataReference.CreateFromFile(l))
                .Cast<MetadataReference>()
                .ToArray();

            var assemblyName = $"{Path.GetRandomFileName()}";
            // 获取编译
            var compilation = CSharpCompilation.Create(assemblyName)
                .WithOptions(new CSharpCompilationOptions(OutputKind.DynamicallyLinkedLibrary))
                .AddReferences(references)
                .AddSyntaxTrees(syntaxTree);

            using var ms = new MemoryStream();
            // 生成编译结果并导出程序集信息到 stream 中
            var compilationResult = compilation.Emit(ms);
            if (compilationResult.Success)
            {
                var assemblyBytes = ms.ToArray();
                // 加载程序集
                return Assembly.Load(assemblyBytes).GetTypes().FirstOrDefault(c => c.Name == classname); ;
            }

            var error = new StringBuilder();
            foreach (var t in compilationResult.Diagnostics)
            {
                error.AppendLine($"{t.GetMessage()}");
            }
            throw new ArgumentException($"Compile error:{Environment.NewLine}{error}");
        }
      
    }

api接口类：

  /// <summary>
       /// /向任务调度中心注册一个任务
       /// </summary>
       /// <param name="task">任务详情</param>
       /// <returns></returns>
        public async Task<JsonResult> AddJob([FromBody] TaskMessage task)
        {
            Console.WriteLine("进入接口");
            if (task.task == null || task.http == null)
            {
                var result = new MessageModel<string>();
                result.success = false;
                result.msg = $"task任务信息和tpp信息不能为空！";
                result.status = 404;
                return Json(result);

            }
            //获取任务详情
            TaskModel taskModel = task.task.ToJson().ToObject<TaskModel>();
            //给任务执行类命名
            string classname = "Task_"+ taskModel.TaskId;
            //赋值
            taskModel.ClassName= classname;
            //获取外方传入的定时任务执行类的接口
            HttpModel httpModel = task.http.ToJson().ToObject<HttpModel>();
            //定义任务具体详情，这个详情是把外方传入的http接口保函在内
            QZTaskModel qZTaskModel = new QZTaskModel();
            qZTaskModel.taskModel = taskModel;
            qZTaskModel.JobParams = classname;
            qZTaskModel.HttpRequest = httpModel.HttpTask;

            if (task.isEmail == true)
            {
                if (task.eamil==null)
                {
                    var result = new MessageModel<string>();
                    result.success = false;
                    result.msg = $"如果要使用邮件发送，请传入文件发送者的信息和接收的信息！";
                    result.status = 404;
                    return Json(result);
                }
                EmailModel email = task.eamil.ToJson().ToObject<EmailModel>();
            }
            if (!CacheHelper.GetKeys(taskModel.TaskId))
            {
                CacheHelper.Set_NotExpire(taskModel.TaskId, task);
            }
            else
            {
                var result = new MessageModel<string>();
                result.success = false;
                result.msg = $"该TaskId已经存在，请重新定义TaskId！";
                result.status =404;
                return Json(result);
            }
           
             var res = await SchedulerHelper.AddScheduleJobAsync(qZTaskModel);
            return Json(res);
        }

　　那么整个过程已经完毕，测试一下， 

1,启动任务调度服务：

 

 在启动一个外部的注册的api接口，也就是外部要执行任务调度的逻辑代码接口：

 

 客户端模拟向任务调度注册一个一个任务：

 

 启动客户端注册：

 

 这样整个流程就跑通了，后续就是看你如何慢慢完善了。就比如可以让对方注册邮箱，和仿照Consul一样让外来方注册进去一个空接口让它是对接口的健康检查，既可以监控外来方的API接口是否正常，任务调度也可以根据接口是否正常而主动的剔除某一个任务。

 

 

标签：core,task,taskModel,System,using,net,任务调度,public
From： https://www.cnblogs.com/zpy1993-09/p/16972073.html