分布式之任务调度学习二

2024-01-03 12:56:30

4 Quartz 集成到 Spring

Spring-quartz 工程
Spring 在 spring-context-support.jar 中直接提供了对 Quartz 的支持
在这里插入图片描述
可以在配置文件中把 JobDetail、Trigger、Scheduler 定义成 Bean。

4.1 定义 Job

<bean name="myJob1" class="org.springframework.scheduling.quartz.JobDetailFactoryBean">
	<property name="name" value="my_job_1"/>
	<property name="group" value="my_group"/>
	<property name="jobClass" value="com.gupaoedu.quartz.MyJob1"/>
	<property name="durability" value="true"/>
</bean>

4.2 定义 Trigger

<bean name="simpleTrigger" class="org.springframework.scheduling.quartz.SimpleTriggerFactoryBean">
	<property name="name" value="my_trigger_1"/>
	<property name="group" value="my_group"/>
	<property name="jobDetail" ref="myJob1"/>
	<property name="startDelay" value="1000"/>
	<property name="repeatInterval" value="5000"/>
	<property name="repeatCount" value="2"/>
</bean>

4.3 定义 Scheduler

<bean name="scheduler" class="org.springframework.scheduling.quartz.SchedulerFactoryBean">
	<property name="triggers">
		<list>
			<ref bean="simpleTrigger"/>
			<ref bean="cronTrigger"/>
		</list>
	</property>
</bean>

既然可以在配置文件配置,当然也可以用@Bean 注解配置。在配置类上加上@Configuration 让 Spring 读取到。

public class QuartzConfig {
	@Bean
	public JobDetail printTimeJobDetail(){
		return JobBuilder.newJob(MyJob1.class)
		.withIdentity("gupaoJob")
		.usingJobData("gupao", "职位更好的你")
		.storeDurably()
		.build();
	}
	@Bean
	public Trigger printTimeJobTrigger() {
		CronScheduleBuilder cronScheduleBuilder = CronScheduleBuilder.cronSchedule("0/5 * * * * ?");
		return TriggerBuilder.newTrigger()
		.forJob(printTimeJobDetail())
		.withIdentity("quartzTaskService")
		.withSchedule(cronScheduleBuilder)
		.build();
	}
}

5 动态调度的实现

springboot-quartz 工程
传统的 Spring 方式集成,由于任务信息全部配置在 xml 文件中,如果需要操作任务或者修改任务运行频率,只能重新编译、打包、部署、重启,如果有紧急问题需要处理,会浪费很多的时间。
有没有可以动态调度任务的方法?比如停止一个 Job?启动一个 Job?修改 Job 的触发频率?
读取配置文件、写入配置文件、重启 Scheduler 或重启应用明显是不可取的。
对于这种频繁变更并且需要实时生效的配置信息,我们可以放到哪里?
ZK、Redis、DB tables。
并且,我们可以提供一个界面,实现对数据表的轻松操作。

5.1 配置管理

这里我们用最简单的数据库的实现。
问题 1:建一张什么样的表?参考 JobDetail 的属性

CREATE TABLE `sys_job`(
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'ID', `job_name` varchar(512) NOT NULL COMMENT '任务名称', `job_group` varchar(512) NOT NULL COMMENT '任务组名', `job_cron` varchar(512) NOT NULL COMMENT '时间表达式', `job_class_path` varchar(1024) NOT NULL COMMENT '类路径,全类型', `job_data_map` varchar(1024) DEFAULT NULL COMMENT '传递 map 参数', `job_status` int(2) NOT NULL COMMENT '状态:1 启用 0 停用', `job_describe` varchar(1024) DEFAULT NULL COMMENT '任务功能描述', PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=25 DEFAULT CHARSET=utf8;

5.2 数据操作与任务调度

操作数据表非常简单,SSM 增删改查。
但是在修改了表的数据之后,怎么让调度器知道呢?
调度器的接口:Scheduler
在我们的需求中,我们需要做的事情:
1、 新增一个任务
2、 删除一个任务
3、 启动、停止一个任务
4、 修改任务的信息(包括调度规律)

5.3 容器启动与 Service 注入

5.3.1 容器启动

因为任务没有定义在 ApplicationContext.xml 中,而是放到了数据库中,Spring Boot 启动时,怎么读取任务信息?
或者,怎么在 Spring 启动完成的时候做一些事情?
创建一个类,实现 CommandLineRunner 接口,实现 run 方法。
从表中查出状态是 1 的任务,然后构建。

5.3.2 Service 类注入到 Job 中

Spring Bean 如何注入到实现了 Job 接口的类中?
例如在 TestTask3 中,需要注入 ISysJobService,查询数据库发送邮件。
如果没有任何配置,注入会报空指针异常。
原因:
因为定时任务 Job 对象的实例化过程是在 Quartz 中进行的,而 Service Bean 是由Spring 容器管理的,Quartz 察觉不到 Service Bean 的存在,所以无法将 Service Bean装配到 Job 对象中。
分析:
Quartz 集成到 Spring 中,用到 SchedulerFactoryBean,其实现了 InitializingBean方法,在唯一的方法 afterPropertiesSet()在 Bean 的属性初始化后调用。调度器用 AdaptableJobFactory 对 Job 对象进行实例化。所以,如果我们可以把这个 JobFactory 指定为我们自定义的工厂的话,就可以在 Job 实例化完成之后,把 Job纳入到 Spring 容器中管理。
解决这个问题的步骤:
1、定义一个 AdaptableJobFactory,实现 JobFactory 接口,实现接口定义的newJob 方法,在这里面返回 Job 实例
在这里插入图片描述
2、定义一个 MyJobFactory,继承 AdaptableJobFactory。
使用 Spring 的 AutowireCapableBeanFactory,把 Job 实例注入到容器中。

@Component
public class MyJobFactory extends AdaptableJobFactory {
@Autowired
	private AutowireCapableBeanFactory capableBeanFactory;
		protected Object createJobInstance(TriggerFiredBundle bundle) throws Exception {
		Object jobInstance = super.createJobInstance(bundle);
		capableBeanFactory.autowireBean(jobInstance);
		return jobInstance;
	}
}

3、指定 Scheduler 的 JobFactory 为自定义的 JobFactory。

scheduler.setJobFactory(myJobFactory);

6 Quartz 集群部署

springboot-quartz 工程

6.1 为什么需要集群?

1、防止单点故障,减少对业务的影响
2、减少节点的压力,例如在 10 点要触发 1000 个任务,如果有 10 个节点,则每个节点之需要执行 100 个任务

6.2 集群需要解决的问题?

1、任务重跑,因为节点部署的内容是一样的,到 10 点的时候,每个节点都会执行相同的操作,引起数据混乱。比如跑批,绝对不能执行多次。
2、任务漏跑,假如任务是平均分配的,本来应该在某个节点上执行的任务,因为节点故障,一直没有得到执行。
3、水平集群需要注意时间同步问题
4、Quartz 使用的是随机的负载均衡算法,不能指定节点执行
所以必须要有一种共享数据或者通信的机制。在分布式系统的不同节点中,我们可以采用什么样的方式,实现数据共享?
两两通信,或者基于分布式的服务,实现数据共享。
例如:ZK、Redis、DB。
在 Quartz 中,提供了一种简单的方式,基于数据库共享任务执行信息。也就是说,一个节点执行任务的时候,会操作数据库,其他的节点查询数据库,便可以感知到了。
同样的问题:建什么表?哪些字段?依旧使用系统自带的 11 张表

6.3 集群配置与验证

quartz.properties 配置。
四个配置:集群实例 ID、集群开关、数据库持久化、数据源信息
注意先清空 quartz 所有表、改端口、两个任务频率改成一样
验证 1:先后启动 2 个节点,任务是否重跑
验证 2:停掉一个节点,任务是否漏

7 Quartz 调度原理

问题:
1、Job 没有继承 Thread 和实现 Runnable,是怎么被调用的?通过反射还是什么?
2、任务是什么时候被调度的?是谁在监视任务还是监视 Trigger?
3、任务是怎么被调用的?谁执行了任务?
4、任务本身有状态吗?还是触发器有状态?
看源码的入口

Scheduler scheduler = factory.getScheduler();
scheduler.scheduleJob(jobDetail, trigger);
scheduler.start();

7.1 获取调度器实例

7.1.1 读取配置文件
public Scheduler getScheduler() throws SchedulerException {
	if (cfg == null) {
		// 读取 quartz.properties 配置文件
		initialize();
	}
	// 这个类是一个 HashMap,用来基于调度器的名称保证调度器的唯一性
	SchedulerRepository schedRep = SchedulerRepository.getInstance();
	Scheduler sched = schedRep.lookup(getSchedulerName());
	// 如果调度器已经存在了
	if (sched != null) {
		// 调度器关闭了,移除
		if (sched.isShutdown()) {
			schedRep.remove(getSchedulerName());
		} else {
			// 返回调度器
			return sched;
		}
	}
	// 调度器不存在,初始化
	sched = instantiate();
	return sched;
}

instantiate()方法中做了初始化的所有工作:

// 存储任务信息的 JobStore
JobStore js = null;
// 创建线程池,默认是 SimpleThreadPool
ThreadPool tp = null;
// 创建调度器
QuartzScheduler qs = null;
// 连接数据库的连接管理器
DBConnectionManager dbMgr = null;
// 自动生成 ID
// 创建线程执行器,默认为 DefaultThreadExecutor
ThreadExecutor threadExecutor;

文章来源:https://blog.csdn.net/lssffy/article/details/135358567
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。