elastic-job之简单job

本文涉及的产品
任务调度 XXL-JOB 版免费试用,400 元额度,开发版规格
云原生网关 MSE Higress,422元/月
MSE Nacos/ZooKeeper 企业版试用,1600元额度,限量50份
简介: 简介 elastic-job是当当网开源的基于zookeeper和quartz实现的分布式作业调度框架。github地址是https://github.com/dangdangdotcom/elastic-job,官方网站是http://elasticjob.io/。

简介

elastic-job是当当网开源的基于zookeeper和quartz实现的分布式作业调度框架。github地址是https://github.com/dangdangdotcom/elastic-job,官方网站是http://elasticjob.io/。elastic-job分elastic-job-lite和elastic-job-cloud,elastic-job-lite定位为轻量级的无中心化解决方案,本文要介绍的用法也是基于elastic-job-lite的。官方的文档其实挺齐全的,本文旨在对elastic-job的应用做一个简单的介绍,也算是完善自己的知识库,详细的信息请参考官方网站

核心概念

  • job:即要执行的任务
  • 分片:即把任务拆分成多个片段,分别调度。这些片段可以落在不同的节点上,所以在实现任务调度的接口时需要根据当前的片段来进行操作,否则就失去了意义。片段数是从0开始的,比如总的分片数是6,一共有两台机器,则第一台机器上分配的片段数将是0,1,2,而第二台机器上分配的片段数将是3,4,5。
  • 重新分片:当有新的机器加入或者有机器宕机的时候都将触发重新分片。因为分片本来就是把总的片段数平均分配给不同的节点,节点数变了,每台机器能够分配的片段必将发生变化。

简单任务

简单任务对应于com.dangdang.ddframe.job.api.simple.SimpleJob接口,该接口的定义如下:

public interface SimpleJob extends ElasticJob {
    
    /**
     * 执行作业.
     *
     * @param shardingContext 分片上下文
     */
    void execute(ShardingContext shardingContext);
}

该接口只定义了一个方法,用于执行需要的任务,你可以把你的定时作业需要执行的逻辑在此方法中实现。elastic-job定时调度时就会调度该execute方法。该方法只接收一个ShardingContext类型的参数。该参数中包含了任务调度一些比较核心的信息,比如分片总数、当前的分片等。任务的实现需要根据当前的片段数来进行,否则可能达不到你的预期效果。以下是一个简单的示例。

/**
 * 普通作业,与Quartz的定时作业类似,只是会多了分片等功能
 * @author Elim
 * 2016年10月29日
 */
public class MyElasticJob implements SimpleJob {
	
	private static final Logger LOGGER = Logger.getLogger(MyElasticJob.class);

	@Override
	public void execute(ShardingContext context) {
		//当你的作业是分片的时候,你需要在你的Job的execute方法中根据当前的分片shardingItem的不同取值实现不同的逻辑,
		//要把所有的shardingItem都覆盖到,因为在分布式环境,每台机器都不能确保它当前的分片是哪一个,并且我们需要保持程序
		//的一致性,程序编写好了对部署是不会有影响的。
		int shardingItem = context.getShardingItem();
		switch (shardingItem) {
			case 0:
				LOGGER.info("处理第一个分片");
				break;
			case 1: 
				LOGGER.info("处理第二个分片");
				break;
			case 2:
				LOGGER.info("处理第三个分片");
				break;
			case 3:
				LOGGER.info("处理第四个分片");
				break;
			case 4:
				LOGGER.info("处理第五个分片");
				break;
			case 5:
				LOGGER.info("处理第六个分片");
				break;
		}
		LOGGER.info(context);
	}

}

配置任务

定义好了作业任务的实现类后为了使作业任务生效,我们需要对其进行配置。配置有两种方式,基于API的配置和基于Spring命名空间的配置。以下介绍的都是基于Spring命名空间的配置。

引入命名空间

首先需要引入reg和job命名空间,示例如下:

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 
	xmlns:context="http://www.springframework.org/schema/context"
	xmlns:reg="http://www.dangdang.com/schema/ddframe/reg"
	xmlns:job="http://www.dangdang.com/schema/ddframe/job"
	xsi:schemaLocation="http://www.springframework.org/schema/beans
                        http://www.springframework.org/schema/beans/spring-beans.xsd
                        http://www.springframework.org/schema/context
					    http://www.springframework.org/schema/context/spring-context.xsd
                        http://www.dangdang.com/schema/ddframe/reg
                        http://www.dangdang.com/schema/ddframe/reg/reg.xsd
                        http://www.dangdang.com/schema/ddframe/job
                        http://www.dangdang.com/schema/ddframe/job/job.xsd
                        ">
</beans>

配置注册中心

reg用于配置作用注册中心,即配置zookeeper。

<reg:zookeeper id="regCenter" server-lists="localhost:2181"
	namespace="dd-job" base-sleep-time-milliseconds="1000"
	max-sleep-time-milliseconds="3000" max-retries="3" />
  • id用于给该注册中心命名。
  • server-lists用于指定使用的zookeeper的地址,多个地址之间用英文的逗号分隔。
  • namespace用于指定注册中心在zookeeper中的命名空间,属于zookeeper的概念。
  • base-sleep-time-milliseconds用于指定等待重试的间隔时间的初始值,单位是毫秒。
  • max-sleep-time-milliseconds用于指定等待重试的间隔时间的最大值,单位是毫秒。
  • max-retries用于指定最大的重试次数。

配置作业

作业通过job命名空间配置,简单任务通过<job:simple/>指定。

<job:simple id="myElasticJob" class="com.elim.learn.elastic.job.MyElasticJob"
	registry-center-ref="regCenter" cron="0/30 * * * * ?"
	sharding-total-count="6" sharding-item-parameters="0=A,1=B,2=C,3=D,4=E,5=F"
	failover="true" overwrite="true" />
  • id用于给该任务命名。
  • class用于指定需要应用的SimpleJob实现类。
  • registry-center-ref用于指定需要使用的注册中心。
  • cron用于指定定时调度的规则,应用cron表达式的语法。
  • sharding-total-count用于指定总的分片数。
  • sharding-item-parameters用于指定每片对应的参数,该参数可以通过ShardingContext的getShardingParameter()获取。
  • failover用于指定是否需要开启失效转移。只有在monitorExecution为true的情况下才有效,可以通过<job:simple monitor-execution="true"/>来指定,不过该属性值默认也是true。
  • overwrite用于指定该配置是否需要用来覆盖注册中心的配置。修改了配置后一定要记得指定该属性值为true,否则还是使用的注册中心的旧的配置。

使用上面的配置后,在Spring容器启动后,我们的作业就会每30秒调度一次了。如果只有一台机器,那么上面的6片都会落到同一台机器上,一共会发起6次调度。如果有两台机器就是每台会得到三个分片,以此类推。当机器数量超出了分片数后,有的机器就会得不到分片,就没有调度的机会,除非有机器宕机了,触发了重新分片。

需要注意的是节点必须是在不同的机器上运行才行,一台机器上启动多个JVM是不会被认为是多个节点的,因为elastic-job是以客户端的IP地址来识别一个节点的。

完整配置如下:

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 
	xmlns:context="http://www.springframework.org/schema/context"
	xmlns:reg="http://www.dangdang.com/schema/ddframe/reg"
	xmlns:job="http://www.dangdang.com/schema/ddframe/job"
	xsi:schemaLocation="http://www.springframework.org/schema/beans
                        http://www.springframework.org/schema/beans/spring-beans.xsd
                        http://www.springframework.org/schema/context
					    http://www.springframework.org/schema/context/spring-context.xsd
                        http://www.dangdang.com/schema/ddframe/reg
                        http://www.dangdang.com/schema/ddframe/reg/reg.xsd
                        http://www.dangdang.com/schema/ddframe/job
                        http://www.dangdang.com/schema/ddframe/job/job.xsd
                        ">

	<description>
		官方文档:http://elasticjob.io/index.html
	</description>
	
	<!-- 如果需要做分布式作业调度,则对应的实例必须是在多台机器上跑的,因为elastic-job是以IP来区分一个节点的;另外namespace和使用的 
		zookeeper也必须是一样的 -->
	<!--配置作业注册中心,server-lists用于指定zookeeper的地址,多个zookeeper之间用逗号分隔;
		namespace用于指定zookeeper命名空间;
		max-retries用于指定最大重试次数 -->
	<reg:zookeeper id="regCenter" server-lists="localhost:2181"
		namespace="dd-job" base-sleep-time-milliseconds="1000"
		max-sleep-time-milliseconds="3000" max-retries="3" />

	<!-- 可通过在http://repo1.maven.org/maven2/com/dangdang/elastic-job-console/1.1.1/下载对应的war包监控elastic-job的运行状态 -->

	<!-- 配置作业 -->
	<!-- 参数overwrite为true即允许客户端的作业配置覆盖注册中心的配置,每次启动服务都会将客户端的覆盖注册中心的, 默认为false。参数failover表示是否开启失效转移,默认为false,其它参数配置请参考官方文档 -->

	<!-- sharding-total-count参数用于指定分片数,当分片数大于机器数量的时候,每台机器分配到的片数会是平均的, 第一片是从0开始的,比如总共分6片,有两台机器,则第一台机器会分得0,1,2三片,而第二台机器会分得3,4,5三片;当有 
		机器宕机了或者有新机器加入的时候都会触发重新分片。如果有多台机器,而分片总数是1的时候即相当于1主多从的配置。 sharding-item-parameters用于指定与分片对应的别名。 
		job-sharding-strategy-class:可以通过它来指定作业分片策略,可选策略可参考官方文档https://github.com/dangdangdotcom/elastic-job/blob/master/elastic-job-doc/content/post/user_guide/lite/other/lite_job_strategy.md。 -->
	<!-- SimpleJob的执行可以参考源码com.dangdang.ddframe.job.executor.type.SimpleJobExecutor的处理逻辑 -->
	<job:simple id="myElasticJob" class="com.elim.learn.elastic.job.MyElasticJob"
		registry-center-ref="regCenter" cron="0/30 * * * * ?"
		sharding-total-count="6" sharding-item-parameters="0=A,1=B,2=C,3=D,4=E,5=F"
		failover="true" overwrite="true" />

</beans>

如果你的job已经定义为了Spring的一个bean,那么在定义<job:simple/>时也可以不指定class,而是指定job-ref属性关联job对应的bean,如:

<bean id="simpleJob" class="com.elim.learn.elastic.job.MyElasticJob"/>
<job:simple id="myElasticJob" job-ref="simpleJob"
	registry-center-ref="regCenter" cron="0/30 * * * * ?"
	sharding-total-count="6" sharding-item-parameters="0=A,1=B,2=C,3=D,4=E,5=F"
	failover="true" overwrite="true" />

这里的bean需要定义在<job:simple/>的前面,否则会提示找不到对应的bean定义。

(本文由Elim写于2017年10月1日)

目录
相关文章
|
应用服务中间件 nginx
nginx优化:URI过长或request header过大导致400或414报错
当出现URI过长或请求头过大导致400或414报错时,可以通过以下方式对Nginx进行优化: 1. 调整client_max_body_size参数:该参数用于限制请求体的大小。默认情况下,Nginx的client_max_body_size参数设置为1M。如果请求体超过这个大小,Nginx会返回400错误。您可以根据实际需求适当增加这个值,例如设置为10M或更大。 ``` http { client_max_body_size 10M; } ``` 2. 调整large_client_header_buffers参数:该参数用于调整请求头缓冲区的大
7563 0
|
XML 域名解析 JSON
【RESTful】RESTful API 接口设计规范 | 示例
【RESTful】RESTful API 接口设计规范 | 示例
13106 0
【RESTful】RESTful API 接口设计规范 | 示例
|
算法 Dubbo NoSQL
Java中5种List的去重方法及它们的效率对比,你用对了吗?
01、使用两个for循环实现List去重(有序) /**使用两个for循环实现List去重(有序) * * @param list * */ public static List removeDuplicationBy2For(List<Integer> list) { for (int i=0;i<list.size();i++) { for (int j=i+1;j<list.size();j++) { if(list.get(i).equa
25265 2
Java中5种List的去重方法及它们的效率对比,你用对了吗?
|
调度
【Cron表达式】cron表达式详细介绍以及常用的例子
【Cron表达式】cron表达式详细介绍以及常用的例子
4918 2
|
消息中间件 监控 前端开发
面试官:核心线程数为0时,线程池如何执行?
线程池是 Java 中用于提升程序执行效率的主要手段,也是并发编程中的核心实现技术,并且它也被广泛的应用在日常项目的开发之中。那问题来了,如果把线程池中的核心线程数设置为 0 时,线程池是如何执行的? 要回答这个问题,我们首先要了解在正常情况下,线程池的执行流程,也就是说当有一个任务来了之后,线程池是如何运行的? ## 1.线程池的执行流程 正常情况下(核心线程数不为 0 的情况下)线程池的执行流程如下: 1. **判断核心线程数**:先判断当前工作线程数是否大于核心线程数,如果结果为 false,则新建线程并执行任务。 2. **判断任务队列**:如果大于核心线程数,则判断任务队列是否
376 1
面试官:核心线程数为0时,线程池如何执行?
|
SQL druid Java
解决 ‘The last packet successfully received from the server was xxx milliseconds ago‘ 问题
解决 ‘The last packet successfully received from the server was xxx milliseconds ago‘ 问题
7399 0
|
消息中间件 小程序 Java
【规范】看看人家Git提交描述,那叫一个规矩
本文通过IDEA中的Git描述规范插件【git commit message helper】,介绍了Git提交描述的规范流程,强调了团队开发中统一标准的重要性,并通过实例展示了规范的提交记录如何提高代码管理和维护效率。最后,文章提供了几个实用的Git提交描述案例,帮助读者更好地理解和应用这些规范。
3530 0
【规范】看看人家Git提交描述,那叫一个规矩
|
JSON 负载均衡 Java
SpringCloud Feign 远程调用(史上最详细讲解)
SpringCloud Feign 远程调用(史上最详细讲解)
14785 0
SpringCloud Feign 远程调用(史上最详细讲解)
|
运维 算法 Java
熟练使用 Elastic Job系列之作业分片策略(五)
框架默认提供了三种分片策略
868 0
熟练使用 Elastic Job系列之作业分片策略(五)