简介
过去,如果在业务中需要处理任务调度的时候,大家都会使用第三方的任务调度组件,而第三方组件有一套自己的规则,在微服务的中显得那么格格不入,这样就会造成代码臃肿,耦合性高,如果有分布式还需要搭建新的分布式环境,如果把任务调度做成组件服务,这个就完全满足了微服务的模块化,组件化,而下面谈的是在surging 中如何支持规则引擎自定义脚本。
调度频率设置
首先在开始之前,先看看如何通过脚本分配多种调度计划,先看下表:
方法描述EveryMinute()每分钟执行一次任务EveryFiveMinutes();每五分钟执行一次任务EveryTenMinutes(); 每十分钟执行一次任务EveryThirtyMinutes()每半小时执行一次任务Hourly();每小时执行一次任务HourlyAt(10)每一个小时的第 10 分钟运行一次Daily()每到午夜执行一次任务DailyAt("3:00")在 3:00 执行一次任务TwiceDaily(1, 3)在 1:00 和 3:00 分别执行一次任务Weekly()每周执行一次任务Monthly()每月执行一次任务MonthlyOn(4, "3:00")在每个月的第四天的 3:00 执行一次任务Quarterly()每季度执行一次任务Yearly()每年执行一次任务Timezone("utc")设置utc时区
举个例子,在工作日每三秒在时间8:00-23:30内执行任务。脚本如下:- parser.TimeZone(""utc"")
- .Weekdays()<br> .SecondAt(3)<br> .Between(""8:00"", ""23:30"")
复制代码 额外的限制条件列表如下:
方法描述Weekdays()限制任务在工作日Sundays()限制任务在星期日Mondays()限制任务在星期一Tuesdays()限制任务在星期二Wednesdays()限制任务在星期三Thursdays()限制任务在星期四Fridays()限制任务在星期五Saturdays()限制任务在星期六When( function(lastExecTime))限制任务基于一个script脚本返回为真的验证Skip( function(lastExecTime))限制任务基于一个script脚本返回为假的验证
举个例子,在工作日每三秒在时间8:00-23:30内执行任务。如果设置When返回为true,skip返回false 就会执行,脚本如下:- parser.TimeZone(""utc"")
- .When(function(lastExecTime){
- return true;
- })
- .Skip(
- function(lastExecTime){
- return false;
- })
- .Weekdays()
- .SecondAt(3)
- .Between(""8:00"", ""23:30"")
复制代码 然后在function 脚本中支持DateUtils对象,可以针对lastExecTime进行逻辑判断,比如是否是周末:DateUtils.IsWeekend(lastExecTime), 是否是今天DateUtils.IsToday(lastExecTime),代码如下:
- parser.TimeZone(""utc"")
- .When(function(lastExecTime){
- return DateUtils.IsToday(lastExecTime);
- })
- .Skip(
- function(lastExecTime){
- return DateUtils.IsWeekend(lastExecTime);
- })
- .Weekdays()
- .SecondAt(3)
- .Between(""8:00"", ""23:30"")
复制代码
编写调度服务
surging微服务引擎是支持后台管理托管服务的,如果要基于BackgroundService编写任务调度,那服务就要继承BackgroundServiceBehavior,还要继承ISingleInstance以设置注入单例模式,
首先,创建接口服务,这样就可以远程添加任务,开启关闭服务了,代码如下:- [ServiceBundle("Background/{Service}")]
- public interface IWorkService : IServiceKey
- {
- Task<bool> AddWork(Message message);
- Task StartAsync();
- Task StopAsync();
- }
复制代码 然后创建业务领域服务,以下代码是通过规则引擎自定义脚本设置执行频率,并且可以设置execsize 以标识同时执行任务的大小,通过以下业务逻辑代码大家可以扩展支持持久化。- public class WorkService : BackgroundServiceBehavior, IWorkService, ISingleInstance
- {
- private readonly ILogger<WorkService> _logger;
- private readonly Queue<Tuple<Message, RulesEngine.RulesEngine, SchedulerRuleWorkflow>> _queue = new Queue<Tuple<Message, RulesEngine.RulesEngine, SchedulerRuleWorkflow>>();
- private readonly ConcurrentDictionary<string, DateTime> _keyValuePairs = new ConcurrentDictionary<string, DateTime>();
- private readonly IServiceProxyProvider _serviceProxyProvider;
- private AtomicLong _atomic=new AtomicLong(1);
- private const int EXECSIZE = 1;
- private CancellationToken _token;
- public WorkService(ILogger<WorkService> logger, IServiceProxyProvider serviceProxyProvider)
- {
- _logger = logger;
- _serviceProxyProvider = serviceProxyProvider;
- /* var script = @"parser
- .Weekdays().SecondAt(3).Between(""8:00"", ""22:00"")";*/
- var script = @"parser
- .TimeZone(""utc"")
- .When(
- function(lastExecTime){
- return DateUtils.IsToday(lastExecTime);
- }).Skip(
- function(lastExecTime){
- return DateUtils.IsWeekend(lastExecTime);
- }).Weekdays().SecondAt(3).Between(""8:00"", ""23:30"")";
- var ruleWorkflow = GetSchedulerRuleWorkflow(script);
- var messageId = Guid.NewGuid().ToString();
- _keyValuePairs.AddOrUpdate(messageId, DateTime.Now, (key, value) => DateTime.Now);
- _queue.Enqueue(new Tuple<Message, RulesEngine.RulesEngine, SchedulerRuleWorkflow>(new Message() { MessageId= messageId,Config=new SchedulerConfig() { IsPersistence=true} }, GetRuleEngine(ruleWorkflow), ruleWorkflow));
- }
- public Task<bool> AddWork(Message message)
- {
- var ruleWorkflow = GetSchedulerRuleWorkflow(message.Config.Script);
- _keyValuePairs.AddOrUpdate(message.MessageId, DateTime.Now, (key, value) => DateTime.Now);
- _queue.Enqueue(new Tuple<Message, RulesEngine.RulesEngine, SchedulerRuleWorkflow>(message, GetRuleEngine(ruleWorkflow), ruleWorkflow));
- return Task.FromResult(true);
- }
- protected override async Task ExecuteAsync(CancellationToken stoppingToken)
- {
- try
- {
- _token = stoppingToken;
- _logger.LogInformation("Worker running at: {time}", DateTimeOffset.Now);
- _queue.TryDequeue(out Tuple<Message, RulesEngine.RulesEngine, SchedulerRuleWorkflow>? message);
- if (message != null)
- {
- var parser = await GetParser(message.Item3, message.Item2);
- await PayloadSubscribe(parser, message.Item1, message.Item2, message.Item3);
- _keyValuePairs.TryGetValue(message.Item1.MessageId, out DateTime dateTime);
- parser.Build(dateTime == DateTime.MinValue ? DateTime.Now : dateTime);
- }
- if (!_token.IsCancellationRequested && (message == null || _atomic.GetAndAdd(1) == EXECSIZE))
- {
- _atomic = new AtomicLong(1);
- await Task.Delay(1000, stoppingToken);
- }
- }
- catch (Exception ex){
- _logger.LogError("WorkService execute error, message:{message} ,trace info:{trace} ", ex.Message, ex.StackTrace);
- }
- }
- public async Task StartAsync()
- {
- if (_token.IsCancellationRequested)
- {
- await base.StartAsync(_token);
- }
- }
- public async Task StopAsync()
- {
- if (!_token.IsCancellationRequested)
- {
- await base.StopAsync(_token);
- }
- }
- private async Task PayloadSubscribe(RulePipePayloadParser parser, Message message, RulesEngine.RulesEngine rulesEngine, SchedulerRuleWorkflow ruleWorkflow)
- {
- parser.HandlePayload().Subscribe(async (temperature) =>
- {
- try
- {
- if (temperature)
- {
- await ExecuteByPlanAsyn(message);
- _logger.LogInformation("Worker exec at: {time}", DateTimeOffset.Now);
- }
- }
- catch (Exception ex) { }
- finally
- {
- if (message.Config.IsPersistence || (!temperature && !message.Config.IsPersistence))
- _queue.Enqueue(new Tuple<Message, RulesEngine.RulesEngine, SchedulerRuleWorkflow>(message, rulesEngine, ruleWorkflow));
- }
- });
- }
- private async Task<bool> ExecuteByPlanAsyn(Message message)
- {
- var result = false;
- var isExec = true;
- try
- {
- if (!string.IsNullOrEmpty(message.RoutePath))
- {
- var serviceResult = await _serviceProxyProvider.Invoke<object>(message.Parameters, message.RoutePath, message.ServiceKey);
- bool.TryParse(serviceResult?.ToString(), out result);
- isExec = true;
- }
- }
- catch { }
- finally
- {
- if (isExec && message.Config.IsPersistence)
- _keyValuePairs.AddOrUpdate(message.MessageId, DateTime.Now, (key, value) => DateTime.Now);
- else if (!message.Config.IsPersistence)
- _keyValuePairs.TryRemove(message.MessageId, out DateTime dateTime);
- }
- return result;
- }
- private async Task<RulePipePayloadParser> GetParser(SchedulerRuleWorkflow ruleWorkflow, RulesEngine.RulesEngine engine)
- {
- var payloadParser = new RulePipePayloadParser();
- var ruleResult = await engine.ExecuteActionWorkflowAsync(ruleWorkflow.WorkflowName, ruleWorkflow.RuleName, new RuleParameter[] { new RuleParameter("parser", payloadParser) });
- if (ruleResult.Exception != null && _logger.IsEnabled(LogLevel.Error))
- _logger.LogError(ruleResult.Exception, ruleResult.Exception.Message);
- return payloadParser;
- }
- private RulesEngine.RulesEngine GetRuleEngine(SchedulerRuleWorkflow ruleWorkFlow)
- {
- var reSettingsWithCustomTypes = new ReSettings { CustomTypes = new Type[] { typeof(RulePipePayloadParser) } };
- var result = new RulesEngine.RulesEngine(new Workflow[] { ruleWorkFlow.GetWorkflow() }, null, reSettingsWithCustomTypes);
- return result;
- }
- private SchedulerRuleWorkflow GetSchedulerRuleWorkflow(string script)
- {
- var result = new SchedulerRuleWorkflow("1==1");
- if (!string.IsNullOrEmpty(script))
- {
- result = new SchedulerRuleWorkflow(script);
- }
- return result;
- }
- }
复制代码 总结
因为工作繁忙,微服务平台暂时搁置,等公司基于surging 的物联网项目上线后,再投入时间研发,surging 一直开发中未曾放弃,也许你没看到的版本才是最强大的。之前的QQ群被封了,如果感兴趣可以加:744677125
开源地址:https://github.com/fanliang11/surging
免责声明:如果侵犯了您的权益,请联系站长,我们会及时删除侵权内容,谢谢合作! |