quarts集群 运维_Quartz.Net分布式运用
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 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 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 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 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 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 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().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 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 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 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()
.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()
.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。怎么用过的!!
quarts集群 运维_Quartz.Net分布式运用相关推荐
- quarts集群 运维_分布式定时任务调度系统技术解决方案(xxl-job、Elastic-job、Saturn)...
1.业务场景 保险人管系统每月工资结算,平安有150万代理人,如何快速的进行工资结算(数据运算型) 保险短信开门红/电商双十一 1000w+短信发送(短时汇聚型) 工作中业务场景非常多,所涉及到的场景 ...
- quarts集群 运维_精讲Elastic-job + Quartz实现企业级定时任务
掌握分布式集群方式的定时任务框架,可以弥补企业中常用的单点任务的缺点,以更高的性能更好的稳定性处理分布式定时任务服务:本课程带你掌握分布式框架Elastic-Job和Quartz,在以多种方式开发定时 ...
- quarts集群 运维_知识拆解 精讲Elastic-job + Quartz实现企业级定时任务 完整版
本站资源全部免费,回复即可查看下载地址! 您需要 登录 才可以下载或查看,没有帐号?立即注册 x 161112tyn9ymz155ohuy1h.jpg (7.39 KB, 下载次数: 15) 2019 ...
- quarts集群 运维_定时任务之Quartz和Elastic-Job
一.背景 常用的定时任务一般有:JDK Timer.Spring Task.Quartz.xxl-job.Elastic-job. JDK Timer:JDK自带的定时任务,1.5之前不支持多线程,1 ...
- 阿里云注册集群+Prometheus 解决多云容器集群运维痛点
作者:左知 容器集群可观测现状 随着 Kubernetes(K8s)容器编排工具已经成为事实上行业通用技术底座,容器集群监控经历多种方案实践后,Prometheus 最终成为容器集群监控的事实标准. ...
- etcd 集群运维实践
[编者的话]etcd 是 Kubernetes 集群的数据核心,最严重的情况是,当 etcd 出问题彻底无法恢复的时候,解决问题的办法可能只有重新搭建一个环境.因此围绕 etcd 相关的运维知识就比较 ...
- Kafka的灵魂伴侣Logi-KafkaManger(4)之运维管控–集群运维(数据迁移和集群在线升级)
推荐一款非常好用的kafka管理平台,kafka的灵魂伴侣 滴滴开源Logi-KafkaManager 一站式Kafka监控与管控平台 技术交流 有想进滴滴LogI开源用户群的加我个人微信: jjdl ...
- 阿里巴巴大规模神龙裸金属 Kubernetes 集群运维实践
戳蓝字"CSDN云计算"关注我们哦! 导读:值得阿里巴巴技术人骄傲的是 2019 年阿里巴巴 双11 核心系统 100% 以云原生的方式上云,完美支撑了 54.4w 峰值流量以及 ...
- 管理大规模容器集群能力包括_阿里巴巴大规模神龙裸金属 Kubernetes 集群运维实践...
导读:值得阿里巴巴技术人骄傲的是 2019 年阿里巴巴 双11 核心系统 100% 以云原生的方式上云,完美支撑了 54.4w 峰值流量以及 2684 亿的成交量.背后承载海量交易的计算力就是来源于容 ...
最新文章
- http响应Last-Modified和ETag以及Apache和Nginx中的配置
- TZOJ 4865 统计单词数(模拟字符串)
- 回溯算法-01遍历所有排列方式问题
- linux检查磁盘空间使用情况df 命令
- SAP Spartacus list item点击之后的detail页面跳转
- 安装php7的mysql扩展,php7安装mysql扩展的方法是什么
- Mac如何设置Vamare Fusion虚拟集的vmnet-8网卡
- minist _On_[GoogleNet]
- Java面向对象之继承、super关键字、方法重写
- 前端基础:vue.js跟node.js分别是什么
- 判断拐点_一文教你“如何寻找拐点”——拐点判断,简单易懂,建议收藏
- css布局模型(摘抄自慕课)
- Android获取Java类名/文件名/方法名/行号
- 【分享】5G+北斗RTK高精度人员定位解决方案
- VC++ 6.0 中如何使用 CRT 调试功能来检测内存泄漏 调试方法
- 如何获取Windows 10的旧默认桌面背景
- linux授读写权限,Linux系统中,设定资料读写权限
- 通过ip地址连接局域网内的打印机(win7、win10)
- 如何低成本,快速构建企业 Wiki 和团队知识分享平台
- gaussian窗口函数_几种常见窗函数及其matlab应用
热门文章
- 打造最快的Hash表
- MATLAB编写自己的BP神经网络程序
- WebStorm 2018.3.4破解方式
- vue获取当前选中行的数据_Vue编程的团队代码规范
- tomcat7的安装与maven安装
- 如何更改webstrom的默认端口63342
- 计算器如何输出log以2为底的对数(利用对数log换底公式)
- Android10加入APEX目的
- sshd启动报错解决:sshd re-exec requires execution with an absolute path
- MacBook2016在SSD上安装Win To Go(成功经验分享)