Quartz.Net的集群部署詳解
標簽（空格分隔）： Quartz.Net Job
最近工作上要用Job，公司的job有些不滿足個人的使用，於是就想自己搞一個Job站練練手，網上看了一下，發現Quartz，於是就了解了一下。
第一版
目前個人使用的是Asp.net Core,在core2.0下面進行的開發。
第一版自己簡單的寫了一個調度器。
public static class SchedulerManage
{
private static IScheduler _scheduler = null;
private static object obj = new object();
public static IScheduler Scheduler
{
get
{
var scheduler = _scheduler;
if (scheduler == null)
{
//在這之前有可能_scheduler被改變了scheduler用的還是原來的值
lock (obj)
{
//這里讀取最新的內存里面的值賦值給scheduler，保證讀取到的是最新的_scheduler
scheduler = Volatile.Read(ref _scheduler);
if (scheduler == null)
{
scheduler = GetScheduler().Result;
Volatile.Write(ref _scheduler, scheduler);
}
}
}
return scheduler;
}
}
public static async Task<BaseResponse> RunJob(IJobDetail job, ITrigger trigger)
{
var response = new BaseResponse();
try
{
var isExist = await Scheduler.CheckExists(job.Key);
var time = DateTimeOffset.Now;
if (isExist)
{
//恢復已經存在任務
await Scheduler.ResumeJob(job.Key);
}
else
{
time = await Scheduler.ScheduleJob(job, trigger);
}
response.IsSuccess = true;
response.Msg = time.ToString("yyyy-MM-dd HH:mm:ss");
}
catch (Exception ex)
{
response.Msg = ex.Message;
}
return response;
}
public static async Task<BaseResponse> StopJob(JobKey jobKey)
{
var response = new BaseResponse();
try
{
var isExist = await Scheduler.CheckExists(jobKey);
if (isExist)
{
await Scheduler.PauseJob(jobKey);
}
response.IsSuccess = true;
response.Msg = "暫停成功！！";
}
catch (Exception ex)
{
response.Msg = ex.Message;
}
return response;
}
public static async Task<BaseResponse> DelJob(JobKey jobKey)
{
var response = new BaseResponse();
try
{
var isExist = await Scheduler.CheckExists(jobKey);
if (isExist)
{
response.IsSuccess = await Scheduler.DeleteJob(jobKey);
}
}
catch (Exception ex)
{
response.IsSuccess = false;
response.Msg = ex.Message;
}
return response;
}
private static async Task<IScheduler> GetScheduler()
{
NameValueCollection props = new NameValueCollection() {
{"quartz.serializer.type", "binary" }
};
StdSchedulerFactory factory = new StdSchedulerFactory(props);
var scheduler = await factory.GetScheduler();
await scheduler.Start();
return scheduler;
}
}
簡單的實現了，動態的運行job，暫停Job，添加job。弄完以后，發現貌似沒啥問題，只要自己把運行的job信息找張表存儲一下，好像都ok了。
輪到發布的時候，突然發現現實機器不止一台，是通過Nigix進行反向代理。突然發現以下幾個問題：
1，多台機器很有可能一個Job在多台機器上運行。
2，當進行部署的時候，必須得停掉機器，如何在機器停掉以后重新部署的時候自動恢復正在運行的Job。
3，如何均衡的運行所有job。
個人當時的想法
1，第一個問題：由於是經過Nigix的反向代理，添加Job和運行job只能落到一台服務器上，基本沒啥問題。個人控制好RunJob的接口，運行了一次，把JobDetail的那張表的運行狀態改成已運行，也就不存在多個機器同時運行的情況。
2，在第一個問題解決的情況下，由於我們公司的Nigix反向代理的邏輯是：均衡策略。所以均衡運行所有job都沒啥問題。
3，重點來了！！！！
如何在部署的時候恢復正在運行的Job？
由於我們已經有了一張JobDetail表。里面可以獲取到哪些正在運行的Job。wome我們把他找出來直接在程序啟動的時候運行一下不就好了嗎嘛。
下面是個人實現的：
//HostedService，在主機運行的時候運行的一個服務
public class HostedService : IHostedService
{
public HostedService(ISchedulerJob schedulerCenter)
{
_schedulerJob = schedulerCenter;
}
private ISchedulerJob _schedulerJob = null;
public async Task StartAsync(CancellationToken cancellationToken)
{
LogHelper.WriteLog("開啟Hosted+Env:"+env);
var reids= new RedisOperation();
if (reids.SetNx("RedisJobLock", "1"))
{
await _schedulerJob.StartAllRuningJob();
}
reids.Expire("RedisJobLock", 300);
}
public async Task StopAsync(CancellationToken cancellationToken)
{
LogHelper.WriteLog("結束Hosted");
var redis = new RedisOperation();
if (redis.RedisExists("RedisJobLock"))
{
var count=redis.DelKey("RedisJobLock");
LogHelper.WriteLog("刪除Reidskey-RedisJobLock結果：" + count);
}
}
}
//注入用的特性
[ServiceDescriptor(typeof(ISchedulerJob), ServiceLifetime.Transient)]
public class SchedulerCenter : ISchedulerJob
{
public SchedulerCenter(ISchedulerJobFacade schedulerJobFacade)
{
_schedulerJobFacade = schedulerJobFacade;
}
private ISchedulerJobFacade _schedulerJobFacade = null;
public async Task<BaseResponse> DelJob(SchedulerJobModel jobModel)
{
var response = new BaseResponse();
if (jobModel != null && jobModel.JobId != 0 && jobModel.JobName != null)
{
response = await _schedulerJobFacade.Modify(new SchedulerJobModifyRequest() { JobId = jobModel.JobId, DataFlag = 0 });
if (response.IsSuccess)
{
response = await SchedulerManage.DelJob(GetJobKey(jobModel));
if (!response.IsSuccess)
{
response = await _schedulerJobFacade.Modify(new SchedulerJobModifyRequest() { JobId = jobModel.JobId, DataFlag = 1 });
}
}
}
else
{
response.Msg = "請求參數有誤";
}
return response;
}
public async Task<BaseResponse> RunJob(SchedulerJobModel jobModel)
{
if (jobModel != null)
{
var jobKey = GetJobKey(jobModel);
var triggleBuilder = TriggerBuilder.Create().WithIdentity(jobModel.JobName + "Trigger", jobModel.JobGroup).WithCronSchedule(jobModel.JobCron).StartAt(jobModel.JobStartTime);
if (jobModel.JobEndTime != null && jobModel.JobEndTime != new DateTime(1900, 1, 1) && jobModel.JobEndTime == new DateTime(1, 1, 1))
{
triggleBuilder.EndAt(jobModel.JobEndTime);
}
triggleBuilder.ForJob(jobKey);
var triggle = triggleBuilder.Build();
var data = new JobDataMap();
data.Add("***", "***");
data.Add("***", "***");
data.Add("***", "***");
var job = JobBuilder.Create<SchedulerJob>().WithIdentity(jobKey).SetJobData(data).Build();
var result = await SchedulerManage.RunJob(job, triggle);
if (result.IsSuccess)
{
var response = await _schedulerJobFacade.Modify(new SchedulerJobModifyRequest() { JobId = jobModel.JobId, JobState = 1 });
if (!response.IsSuccess)
{
await SchedulerManage.StopJob(jobKey);
}
return response;
}
else
{
return result;
}
}
else
{
return new BaseResponse() { Msg = "Job名稱為空！！" };
}
}
public async Task<BaseResponse> StopJob(SchedulerJobModel jobModel)
{
var response = new BaseResponse();
if (jobModel != null && jobModel.JobId != 0 && jobModel.JobName != null)
{
response = await _schedulerJobFacade.Modify(new SchedulerJobModifyRequest() { JobId = jobModel.JobId, JobState = 2 });
if (response.IsSuccess)
{
response = await SchedulerManage.StopJob(GetJobKey(jobModel));
if (!response.IsSuccess)
{
response = await _schedulerJobFacade.Modify(new SchedulerJobModifyRequest() { JobId = jobModel.JobId, JobState = 2 });
}
}
}
else
{
response.Msg = "請求參數有誤";
}
return response;
}
private JobKey GetJobKey(SchedulerJobModel jobModel)
{
return new JobKey($"{jobModel.JobId}_{jobModel.JobName}", jobModel.JobGroup);
}
public async Task<BaseResponse> StartAllRuningJob()
{
try
{
var jobListResponse = await _schedulerJobFacade.QueryList(new SchedulerJobListRequest() { DataFlag = 1, JobState = 1, Environment=Kernel.Environment.ToLower() });
if (!jobListResponse.IsSuccess)
{
return jobListResponse;
}
var jobList = jobListResponse.Models;
foreach (var job in jobList)
{
await RunJob(job);
}
return new BaseResponse() { IsSuccess = true, Msg = "程序啟動時，啟動所有運行中的job成功！！" };
}
catch (Exception ex)
{
LogHelper.WriteExceptionLog(ex);
return new BaseResponse() { IsSuccess = false, Msg = "程序啟動時，啟動所有運行中的job失敗！！" };
}
}
}
在程序啟動的時候，把所有的Job去運行一遍，當中對於多次運行的用到了Redis的分布式鎖，現在啟動的時候鎖住，不讓別人運行，在程序卸載的時候去把鎖釋放掉！！感覺沒啥問題，主要是可能負載均衡有問題，全打到一台服務器上去了，勉強能夠快速的打到效果。當然高可用什么的就先犧牲掉了。
坑點又來了
大家知道，在稍微大點的公司，運維和開發是分開的，公司用的daoker進行部署，在程序停止的時候，不會調用
HostedService的StopAsync方法！！
當時心里真是一萬個和諧和諧奔騰而過!!
個人也就懶得和運維去扯這些東西了。最后的最后就是：設置一個redis的分布式鎖的過期時間，大概預估一個部署的時間，只要在部署直接，鎖能夠在就行了，然后每次部署的間隔要大於鎖過期時間。好麻煩，說多了都是淚！！
Quartz.Net的分布式集群運用
Schedule配置
public async Task<IScheduler> GetScheduler()
{
var properties = new NameValueCollection();
properties["quartz.serializer.type"] = "binary";
//存儲類型
properties["quartz.jobStore.type"] = "Quartz.Impl.AdoJobStore.JobStoreTX, Quartz";
//表明前綴
properties["quartz.jobStore.tablePrefix"] = "QRTZ_";
//驅動類型
properties["quartz.jobStore.driverDelegateType"] = "Quartz.Impl.AdoJobStore.SqlServerDelegate, Quartz";
//數據庫名稱
properties["quartz.jobStore.dataSource"] = "SchedulJob";
//連接字符串Data Source = myServerAddress;Initial Catalog = myDataBase;User Id = myUsername;Password = myPassword;
properties["quartz.dataSource.SchedulJob.connectionString"] = "Data Source =.; Initial Catalog = SchedulJob;User ID = sa; Password = *****;";
//sqlserver版本(Core下面已經沒有什么20，21版本了)
properties["quartz.dataSource.SchedulJob.provider"] = "SqlServer";
//是否集群，集群模式下要設置為true
properties["quartz.jobStore.clustered"] = "true";
properties["quartz.scheduler.instanceName"] = "TestScheduler";
//集群模式下設置為auto，自動獲取實例的Id，集群下一定要id不一樣，不然不會自動恢復
properties["quartz.scheduler.instanceId"] = "AUTO";
properties["quartz.threadPool.type"] = "Quartz.Simpl.SimpleThreadPool, Quartz";
properties["quartz.threadPool.threadCount"] = "25";
properties["quartz.threadPool.threadPriority"] = "Normal";
properties["quartz.jobStore.misfireThreshold"] = "60000";
properties["quartz.jobStore.useProperties"] = "false";
ISchedulerFactory factory = new StdSchedulerFactory(properties);
return await factory.GetScheduler();
}
然后是測試代碼：
public async Task TestJob()
{
var sched = await GetScheduler();
//Console.WriteLine("***** Deleting existing jobs/triggers *****");
//sched.Clear();
Console.WriteLine("------- Initialization Complete -----------");
Console.WriteLine("------- Scheduling Jobs ------------------");
string schedId = sched.SchedulerName; //sched.SchedulerInstanceId;
int count = 1;
IJobDetail job = JobBuilder.Create<SimpleRecoveryJob>()
.WithIdentity("job_" + count, schedId) // put triggers in group named after the cluster node instance just to distinguish (in logging) what was scheduled from where
.RequestRecovery() // ask scheduler to re-execute this job if it was in progress when the scheduler went down...
.Build();
ISimpleTrigger trigger = (ISimpleTrigger)TriggerBuilder.Create()
.WithIdentity("triger_" + count, schedId)
.StartAt(DateBuilder.FutureDate(1, IntervalUnit.Second))
.WithSimpleSchedule(x => x.WithRepeatCount(1000).WithInterval(TimeSpan.FromSeconds(5)))
.Build();
Console.WriteLine("{0} will run at: {1} and repeat: {2} times, every {3} seconds", job.Key, trigger.GetNextFireTimeUtc(), trigger.RepeatCount, trigger.RepeatInterval.TotalSeconds);
sched.ScheduleJob(job, trigger);
count++;
job = JobBuilder.Create<SimpleRecoveryJob>()
.WithIdentity("job_" + count, schedId) // put triggers in group named after the cluster node instance just to distinguish (in logging) what was scheduled from where
.RequestRecovery() // ask scheduler to re-execute this job if it was in progress when the scheduler went down...
.Build();
trigger = (ISimpleTrigger)TriggerBuilder.Create()
.WithIdentity("triger_" + count, schedId)
.StartAt(DateBuilder.FutureDate(2, IntervalUnit.Second))
.WithSimpleSchedule(x => x.WithRepeatCount(1000).WithInterval(TimeSpan.FromSeconds(5)))
.Build();
Console.WriteLine(string.Format("{0} will run at: {1} and repeat: {2} times, every {3} seconds", job.Key, trigger.GetNextFireTimeUtc(), trigger.RepeatCount, trigger.RepeatInterval.TotalSeconds));
sched.ScheduleJob(job, trigger);
// jobs don't start firing until start() has been called...
Console.WriteLine("------- Starting Scheduler ---------------");
sched.Start();
Console.WriteLine("------- Started Scheduler ----------------");
Console.WriteLine("------- Waiting for one hour... ----------");
Thread.Sleep(TimeSpan.FromHours(1));
Console.WriteLine("------- Shutting Down --------------------");
sched.Shutdown();
Console.WriteLine("------- Shutdown Complete ----------------");
}
測試添加兩個job,每隔5s執行一次。
在圖中可以看到：job1和job2不會重復執行，當我停了Job2時，job2也在job1當中運行。
這樣就可以實現分布式部署時的問題了，Quzrtz.net的數據庫結構隨便網上找一下，運行一些就好了。
截取幾個數據庫的數據圖：基本上就存儲了一些這樣的信息
JobDetail
觸發器的數據
這個是調度器的
這個是鎖的
下一期：
1.Job的介紹：有狀態Job，無狀態Job。
2.MisFire
3.Trigger,Cron介紹
4.第一部分的改造，自己實現一個基於在HostedService能夠進行分布式調度的Job類，其實只要實現了這個，其他的上面講的都沒有問題。棄用Quartz的表的行級鎖。因為這並發高了比較慢！！
個人問題
個人還是沒有測試出來這個RequestRecovery。怎么用過的！！