1.什么是ScheduleMaster

ScheduleMaster分布式任务调度系统。简称:集中任务调度系统,最简单的理解ScheduleMaster,就是对不同的系统里面的调度任务做统一管理的框架

例如我们现在有多个系统,每个系统针对自己处理不同的业务场景。衍生出自己的调度任务,想象一下,如果每个系统人为去维护,那随着调度任务越来越多,人是崩溃的吧,可见维护技术成本是巨大的,这时我们需要选择分布式任务系统框架做统一的管理

当然有目前有很多相对优秀分布式任务系统框架,我们主要学习 ScheduleMaster

2.使用ScheduleMaster

1.首先我们需要使用NET Core web Api创建几个模拟的微服务,分别为 考勤算薪邮件短信

2.下载开源的ScheduleMaster,并且使用ScheduleMaster调度我们的微服务接口

- sqlserver:"Persist Security Info = False; User ID =sa; Password =123456; Initial Catalog =schedule_master; Server =."
- postgresql:"Server=localhost;Port=5432;Database=schedule_master;User Id=postgres;Password=123456;Pooling=true;MaxPoolSize=20;"
- mysql:"Data Source=localhost;Database=schedule_master;User ID=root;Password=123456;pooling=true;CharSet=utf8mb4;port=3306;sslmode=none;TreatTinyAsBoolean=true"

修改Host的配置文件和支持的数据库,框架默认使用Mysql

修改Web的配置文件和支持的数据库,框架默认使用Mysql

3.进入Hos.ScheduleMaster.Web项目的发布目录,dotnet Hos.ScheduleMaster.Web.dll启动项目 ,此时会生成数据库

4.进入Hos.ScheduleMaster.QuartzHost项目的发布目录,执行命令,启动项目

dotnet Hos.ScheduleMaster.QuartzHost.dll --urls http://*:30003

1.配置Http调度任务

5.准备就绪后,使用后台查看节点管理,可以看到web主节点30000任务调度的接口30002已经在运行

6.使用任务列表菜单,创建定时调度任务,配置基础信息元数据配置,然后点击保存就开始执行任务

2.配置程序集调度任务

1.创建一个类库,安装ScheduleMaster库, 创建一个类继承TaskBase,实现抽象方法,然后编译程序集

namespace TaskExcuteService
{
    class AssemblyTask : TaskBase
    {
        public override void Run(TaskContext context)
        {
            context.WriteLog("程序集任务");
        }
    }
}

2.找到debug目录,去掉Base程序集,然后添加为压缩文件

3.在web界面找到任务配置,选择程序集进行基本配置,配置元数据,输入程序集名称,然后输入类,并将程序集上传,保存就可以运行了

3.使用Api接入任务

为了方便业务系统更好的接入调度系统,ScheduleMaster创建任务不仅可以在控制台中实现,系统也提供了WebAPI供业务系统使用代码接入,这种方式对延时任务来说尤其重要。

1.API Server 对接流程

  • 在控制台中创建好专用的API对接用户账号。

  • 使用对接账号的用户名设置为http header中的ms_auth_user值。

  • 使用经过哈希运算过的秘钥设置为http header中的ms_auth_secret值,计算规则:按{用户名}{hash(密码)}{用户名}的格式拼接得到字符串str,然后再对str做一次hash运算即得到最终秘钥,hash函数是小写的32位MD5算法。

  • 使用form格式发起http调用,如果非法用户会返回401-Unauthorized。

HttpClient client = new HttpClient();
client.DefaultRequestHeaders.Add("ms_auth_user", "admin");
client.DefaultRequestHeaders.Add("ms_auth_secret", SecurityHelper.MD5($"admin{SecurityHelper.MD5("111111")}}admin"));

所有接口采用统一的返回格式,字段如下:

参数名称参数类型说明
Successbool是否成功
Statusint结果状态,0-请求失败 1-请求成功 2-登录失败 3-参数异常 4-数据异常
Messagestring返回的消息
Dataobject返回的数据

2.创建程序集任务

使用API创建任务的方式不支持上传程序包,所以在任务需要启动时要确保程序包已通过其他方式上传,否则会启动失败。

  • 接口地址:http://yourip:30000/api/task/create

  • 请求类型:POST

  • 参数格式:application/x-www-form-urlencoded

  • 返回结果:创建成功返回任务id

  • 参数列表:

参数名称参数类型是否必填说明
MetaTypeint任务类型,这里固定是1
Titlestring任务名称
RunLoopbool是否按周期执行
CronExpressionstringcron表达式,如果RunLoop为true则必填
AssemblyNamestring程序集名称
ClassNamestring执行类名称,包含完整命名空间
StartDateDateTime任务开始时间
EndDateDateTime任务停止时间,为空表示不限停止时间
Remarkstring任务描述说明
KeepersList<int>监护人id
NextsList<guid>子级任务id
ExecutorsList<string>执行节点名称
RunNowbool创建成功是否立即启动
ParamsList<ScheduleParam>自定义参数列表,也可以通过CustomParamsJson字段直接传json格式字符串

ScheduleParam:

参数名称参数类型是否必填说明
ParamKeystring参数名称
ParamValuestring参数值
ParamRemarkstring参数说明
HttpClient client = new HttpClient();
List<KeyValuePair<string, string>> args = new List<KeyValuePair<string, string>>();
args.Add(new KeyValuePair<string, string>("MetaType", "1"));
args.Add(new KeyValuePair<string, string>("RunLoop", "true"));
args.Add(new KeyValuePair<string, string>("CronExpression", "33 0/8 * * * ?"));
args.Add(new KeyValuePair<string, string>("Remark", "By Xunit Tester Created"));
args.Add(new KeyValuePair<string, string>("StartDate", DateTime.Today.ToString("yyyy-MM-dd HH:mm:ss")));
args.Add(new KeyValuePair<string, string>("Title", "程序集接口测试任务"));
args.Add(new KeyValuePair<string, string>("AssemblyName", "Hos.ScheduleMaster.Demo"));
args.Add(new KeyValuePair<string, string>("ClassName", "Hos.ScheduleMaster.Demo.Simple"));
args.Add(new KeyValuePair<string, string>("CustomParamsJson", "[{\"ParamKey\":\"k1\",\"ParamValue\":\"1111\",\"ParamRemark\":\"r1\"},{\"ParamKey\":\"k2\",\"ParamValue\":\"2222\",\"ParamRemark\":\"r2\"}]"));
args.Add(new KeyValuePair<string, string>("Keepers", "1"));
args.Add(new KeyValuePair<string, string>("Keepers", "2"));
//args.Add(new KeyValuePair<string, string>("Nexts", ""));
//args.Add(new KeyValuePair<string, string>("Executors", ""));
HttpContent reqContent = new FormUrlEncodedContent(args);
var response = await client.PostAsync("http://localhost:30000/api/Task/Create", reqContent);
var content = await response.Content.ReadAsStringAsync();
Debug.WriteLine(content);

3.创建HTTP任务

  • 接口地址:http://yourip:30000/api/task/create

  • 请求类型:POST

  • 参数格式:application/x-www-form-urlencoded

  • 返回结果:创建成功返回任务id

  • 参数列表:

参数名称参数类型是否必填说明
MetaTypeint任务类型,这里固定是2
Titlestring任务名称
RunLoopbool是否按周期执行
CronExpressionstringcron表达式,如果RunLoop为true则必填
StartDateDateTime任务开始时间
EndDateDateTime任务停止时间,为空表示不限停止时间
Remarkstring任务描述说明
HttpRequestUrlstring请求地址
HttpMethodstring请求方式,仅支持GET\POST\PUT\DELETE
HttpContentTypestring参数格式,仅支持application/json和application/x-www-form-urlencoded
HttpHeadersstring自定义请求头,ScheduleParam列表的json字符串
HttpBodystring如果是json格式参数,则是对应参数的json字符串;如果是form格式参数,则是对应ScheduleParam列表的json字符串。
KeepersList<int>监护人id
NextsList<guid>子级任务id
ExecutorsList<string>执行节点名称
RunNowbool创建成功是否立即启动
HttpClient client = new HttpClient();
List<KeyValuePair<string, string>> args = new List<KeyValuePair<string, string>>();
args.Add(new KeyValuePair<string, string>("MetaType", "2"));
args.Add(new KeyValuePair<string, string>("RunLoop", "true"));
args.Add(new KeyValuePair<string, string>("CronExpression", "22 0/8 * * * ?"));
args.Add(new KeyValuePair<string, string>("Remark", "By Xunit Tester Created"));
args.Add(new KeyValuePair<string, string>("StartDate", DateTime.Today.ToString("yyyy-MM-dd HH:mm:ss")));
args.Add(new KeyValuePair<string, string>("Title", "Http接口测试任务"));
args.Add(new KeyValuePair<string, string>("HttpRequestUrl", "http://localhost:56655/api/1.0/value/jsonpost"));
args.Add(new KeyValuePair<string, string>("HttpMethod", "POST"));
args.Add(new KeyValuePair<string, string>("HttpContentType", "application/json"));
args.Add(new KeyValuePair<string, string>("HttpHeaders", "[]"));
args.Add(new KeyValuePair<string, string>("HttpBody", "{ \"Posts\": [{ \"PostId\": 666, \"Title\": \"tester\", \"Content\":\"testtesttest\" }], \"BlogId\": 111, \"Url\":\"qweqrrttryrtyrtrtrt\" }"));
HttpContent reqContent = new FormUrlEncodedContent(args);
var response = await client.PostAsync("http://localhost:30000/api/Task/Create", reqContent);
var content = await response.Content.ReadAsStringAsync();
Debug.WriteLine(content);

4.创建延时任务

  • 接口地址:http://yourip:30000/api/delaytask/create

  • 请求类型:POST

  • 参数格式:application/x-www-form-urlencoded

  • 返回结果:创建成功返回任务id

  • 参数列表:

参数名称参数类型是否必填说明
SourceAppstring来源
Topicstring主题
ContentKeystring业务关键字
DelayTimeSpanint延迟相对时间
DelayAbsoluteTimeDateTime延迟绝对时间
NotifyUrlstring回调地址
NotifyDataTypestring回调参数格式,仅支持application/json和application/x-www-form-urlencoded
NotifyBodystring回调参数,json格式字符串
for (int i = 0; i < 5; i++)
{
    int rndNum = new Random().Next(20, 500);
    List<KeyValuePair<string, string>> args = new List<KeyValuePair<string, string>>();
    args.Add(new KeyValuePair<string, string>("SourceApp", "TestApp"));
    args.Add(new KeyValuePair<string, string>("Topic", "TestApp.Trade.TimeoutCancel"));
    args.Add(new KeyValuePair<string, string>("ContentKey", i.ToString()));
    args.Add(new KeyValuePair<string, string>("DelayTimeSpan", rndNum.ToString()));
    args.Add(new KeyValuePair<string, string>("DelayAbsoluteTime", DateTime.Now.AddSeconds(rndNum).ToString("yyyy-MM-dd HH:mm:ss")));
    args.Add(new KeyValuePair<string, string>("NotifyUrl", "http://localhost:56655/api/1.0/value/delaypost"));
    args.Add(new KeyValuePair<string, string>("NotifyDataType", "application/json"));
    args.Add(new KeyValuePair<string, string>("NotifyBody", "{ \"Posts\": [{ \"PostId\": 666, \"Title\": \"tester\", \"Content\":\"testtesttest\" }], \"BlogId\": 111, \"Url\":\"qweqrrttryrtyrtrtrt\" }"));
    HttpContent reqContent = new FormUrlEncodedContent(args);
    var response = await client.PostAsync("http://localhost:30000/api/DelayTask/Create", reqContent);
    var content = await response.Content.ReadAsStringAsync();
    Debug.WriteLine(content);
}

4.框架简单分析

1.全局设计

根据官网的设计图,以及操作的流程,简单总结一下任务全局流程为客户端—–>master——>work—–>执行任务。从大的架构方向的思想就是,将原有单个服务业务和任务调度混合的方式做一些改变,使业务调度分离,专门把任务调度部分剥离出来,作为一个独立的进程,来统一调用管理任务,而原有服务只做业务处理

####### 2.Master和Work分析

我们主要从主节点从节点数据表这几个方面来简单分析,整个业务的时序流程,从本质来看并不复杂masterWork之间的关系是相互配合的,可能乍一看下面整个图有点混乱,下面来解释一下,其实Master节点将任务添加数据中,然后work节点,去从对应的数据表中取出任务数据,然后根据任务对应的配置生成配置信息,调用Quartz执行任务,这就是我们整个框架中最核心的业务。

当然master和work除了主要承载了整个管理系统的UI可视化、后台业务操作任务执行之外,如果从细节以及实现方式来说主要做了以下事情:

Master

  • 1.分配任务执行选择节点
  • 2.对work节点进行健康检查,对任务进行故障转移

Work

  • 1.取出任务配置信息
  • 2.使用Quartz根据配置运行任务
  • 3.使用反射调用程序集
  • 4.使用httpclient调用http 接口
作者:|有梦想的鱼i|,原文链接: https://www.cnblogs.com/yuxl01/p/16311534.html

文章推荐

SpringCloud 服务治理

linux-ext4格式文件误删除,该如何恢复?

手把手教你使用 Spring Boot 3 开发上线一个前后端分离的生...

如何简单、高效的进行留存和漏斗分析

Go到底能不能实现安全的双检锁?

企业应用架构研究系列二十六:信号量SemaphoreSlim与Semapho...

一次搞定 vue3的 路由 + 菜单 + tabs

基于.NetCore开发博客项目 StarBlog - (8) 分类层级结构展示

Jwt隐藏大坑,通过源码帮你揭秘

Python数据分析--工具安装及Numpy介绍(1)

kubeadm安装k8s 1.23.5

一文带你了解 「图数据库」Nebula 的存储设计和思考