刘玟
.net core实践系列之短信服务-Sikiro.SMS.Job服务的实现
来源:范淑婷     发布时间: 2019-05-01      浏览次数:11

字号:

前言

本篇会继续讲解Sikiro.SMS.Job服务的实现,在我写第一篇的时候,我就发现我当时设计的架构里Sikiro.SMS.Job这个可以选择不需要,而使用MQ代替。但是为了说明调度任务使用实现也坚持写了下。后面会一篇针对架构、实现优化的讲解。

源码地址:https://github.com/SkyChenSky/Sikiro.SMS

Quartz的简介

Quartz.NET是一款功能齐全的开源作业调度框架,小至的应用程序,大到企业系统都可以适用。Quartz是作者James House用JAVA语言编写的,而Quartz.NET是从Quartz移植过来的C#版本。

Quartz.Net的作用

Quartz.Net是多线程的,允许多个JOB同时执行。Quartz.Net可以进行持久化,结合管理后台可以进行可视化的监控Quartz.Net提供API进行远程操控,结合管理后台可以进行运维管理

在一般企业,可以利用Quartz.Net框架做各种的定时任务,例如,数据迁移、跑报表等等。

Cron表达式

字段名是否必填值范围特殊字符
SecondsYES0-59, - * /
MinutesYES0-59, - * /
HoursYES0-23, - * /
Day of monthYES1-31, - * ? / L W
MonthYES1-12 or JAN-DEC, - * /
Day of weekYES1-7 or SUN-SAT, - * ? / L #
YearNOempty, 1970-2099, - * /

缺点

Quartz.Net的缺点很明显,没有自带的管理后台,而同款的Hangfir调度任务框架则会有更加良好的易用性。但是在Github上有不少人开源了Quartz.Net的管理后台,对此作为了弥补。

其他

其他Quartz.Net的信息可以看我之前记录的一篇文章《Quartz.NET的使用(附源码)》

Quartz.Net DEMO:https://github.com/SkyChenSky/QuartzDotNetDemo.git

业务流程

从MongoDB持久化的数据,查询出状态为待处理并且定时时间小于当前时间的数据。通过Mongo驱动提供的FindOneAndUpdate对文档进行原子性操作(更新中间状态并查询出刚更新的文档)。如果有数据则发送到MQ,由Sikiro.SMS.Bus进行订阅发送,因为本次有数据,我认为可能还会有其他需要发送的数据,因此立刻调用JOB自身方法,进行下一条需要处理的数据进行发送。如果此次JOB的执行并没有数据,那么认为接下来一段时间没有需要处理的数据,这次调度结束。

TimeSendSms示例

public class TimeSendSms : BaseJob { private readonly SmsService _smsService; private readonly IBus _bus; public TimeSendSms(SmsService smsService, IBus bus) { _smsService = smsService; _bus = bus; } protected override void ExecuteBusiness() { _smsService.GetToBeSend(); if (_smsService.Sms != null) _bus.Publish(_smsService.Sms.MapTo<SmsModel, SmsQueueModel>()); _smsService.ContinueDo(ExecuteBusiness); } protected override void OnException() { _smsService.RollBack(); } }

模板模式

Job的轮询处理流程基本相似,查询出需要执行数据-遍历业务处理-如果有异常则特殊处理,因此针对类似流程相同,但是实现有差异的程序,我们可以使用模板模式。

public abstract class BaseJob : IJob { private void OnException(Action action) { try { action(); } catch (Exception e) { e.WriteToFile(); OnException(); } } public Task Execute(IJobExecutionContext context) { OnException(ExecuteBusiness); return null; } protected virtual void OnException() { } protected abstract void ExecuteBusiness(); }

 

Mongo的原子性

原子性

原子是物理概念,指的是指化学反应不可再分的基本微粒。而计算机领域的原子性强调的对象是操作(指令、事务)。我们所说的指令组是原子操作,意思要么一起成功,要么一起失败。不允许2个指令里,一个成功一个失败的情况存在。

MongoDB 原子操作

MongoDB的原子操作就是要么这个文档完整的保存到Mongodb,要么没有保存到Mongodb,不会出现查询到的文档没有保存完整的情况。

MongoDB的文档的保存,修改,删除等操作都是原子性,除此之外还提供了FindOneAndDelete、FindOneAndUpdate、FindOneAndReplace等原子操作。

以FindOneAndUpdate为例,对某文档FindOneAndUpdate,可以文档B进行Update操作完成后返回出文档B的结果,根据参数返回结果是更新前还是更新后(一般我们需要更新后)。

而这FindOneAndUpdate的操作对于我们更新到中间状态的非常实用:

避免进行Update后无法良好的查询到刚Update的文档避免应用集群部署时批量更新后,无法良好分配任务批量更新多个文档需要isolated标识隔离,全局锁在大并发情况下性能并不乐观

虽然以上可以通过更新时标识版本号进行解决,这无疑增加实现难度。

MongoDB锁机制

Mongodb并发操作又读写锁来进行控制。

简单来说

当进行读操作的时候会加读锁,这个时候其他读操作可以也获得读锁,但是不能加写锁,也就是说不能进行写操作。

当进行写操作的时候会加写锁,这个时候其他操作无法加任何锁,也就是说不能进行其他的读操作和写操作。

多个JOB的并发性

综上所述,落实到我们应用场景,在部署多个调度任务服务,或者JOB多个线程去跑时,我们可以使用FindOneAndUpdate,每个调度任务每次只处理一个文档,Update操作的时候会进行写锁阻塞其他进程(进程)的写操作。那么就可以保证每个调度任务都可以只处理唯一一个有效的文档,避免重复处理。

下面是我的Sikiro.Nosql.Mongo的FindOneAndUpdate封装示例,因为Update字段的不友好,所以我封装了一下Lambda表达式,ReturnDocument = ReturnDocument.After标识响应数据是更新前还是更新后的文档。

public T GetAndUpdate<T>(string database, string collection, Expression<Func<T, bool>> predicate, Expression<Func<T, T>> updateExpression) { var db = _mongoClient.GetDatabase(database); var col = db.GetCollection<T>(collection); var updateDefinitionList = MongoExpression<T>.GetUpdateDefinition(updateExpression); var updateDefinitionBuilder = new UpdateDefinitionBuilder<T>().Combine(updateDefinitionList); return col.FindOneAndUpdate(predicate, updateDefinitionBuilder, new FindOneAndUpdateOptions<T, T> { ReturnDocument = ReturnDocument.After });

SQL Server的UpdateSelect

SQL Server的操作也具有上述FindOneAndUpdate的功能,我们公司成他为UpdateSelect,下面是示例代码:

UPDATE TOP ( 100 )        SYS_USER WITH ( UPDLOCK, READPAST )SET     USER_STATUS = 1OUTPUT  INSERTED.[USER_NAME] ,        INSERTED.SYS_USERID ,        INSERTED.EMAILFROM    SYS_USERWHERE   CREATE_DATETIME < "2018-09-13"        AND USER_STATUS = 2;

结尾

本篇介绍了调度任务结合MongoDB原子操作的使用,使得调度任务服务可以具有良好的伸缩性。如果有任何建议与问题可以在下方评论反馈给我。

, 1, 0, 9);

  • 相关内容: