开发者社区> elim1> 正文

elastic-job之简单job

简介: 简介 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日)

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
如何设置阿里云服务器安全组?阿里云安全组规则详细解说
阿里云安全组设置详细图文教程(收藏起来) 阿里云服务器安全组设置规则分享,阿里云服务器安全组如何放行端口设置教程。阿里云会要求客户设置安全组,如果不设置,阿里云会指定默认的安全组。那么,这个安全组是什么呢?顾名思义,就是为了服务器安全设置的。安全组其实就是一个虚拟的防火墙,可以让用户从端口、IP的维度来筛选对应服务器的访问者,从而形成一个云上的安全域。
18579 0
阿里云服务器如何登录?阿里云服务器的三种登录方法
购买阿里云ECS云服务器后如何登录?场景不同,阿里云优惠总结大概有三种登录方式: 登录到ECS云服务器控制台 在ECS云服务器控制台用户可以更改密码、更换系.
27718 0
阿里云服务器安全组设置内网互通的方法
虽然0.0.0.0/0使用非常方便,但是发现很多同学使用它来做内网互通,这是有安全风险的,实例有可能会在经典网络被内网IP访问到。下面介绍一下四种安全的内网互联设置方法。 购买前请先:领取阿里云幸运券,有很多优惠,可到下文中领取。
21933 0
阿里云服务器端口号设置
阿里云服务器初级使用者可能面临的问题之一. 使用tomcat或者其他服务器软件设置端口号后,比如 一些不是默认的, mysql的 3306, mssql的1433,有时候打不开网页, 原因是没有在ecs安全组去设置这个端口号. 解决: 点击ecs下网络和安全下的安全组 在弹出的安全组中,如果没有就新建安全组,然后点击配置规则 最后如上图点击添加...或快速创建.   have fun!  将编程看作是一门艺术,而不单单是个技术。
19979 0
阿里云服务器ECS登录用户名是什么?系统不同默认账号也不同
阿里云服务器Windows系统默认用户名administrator,Linux镜像服务器用户名root
15287 0
腾讯云服务器 设置ngxin + fastdfs +tomcat 开机自启动
在tomcat中新建一个可以启动的 .sh 脚本文件 /usr/local/tomcat7/bin/ export JAVA_HOME=/usr/local/java/jdk7 export PATH=$JAVA_HOME/bin/:$PATH export CLASSPATH=.
14852 0
使用OpenApi弹性释放和设置云服务器ECS释放
云服务器ECS的一个重要特性就是按需创建资源。您可以在业务高峰期按需弹性的自定义规则进行资源创建,在完成业务计算的时候释放资源。本篇将提供几个Tips帮助您更加容易和自动化的完成云服务器的释放和弹性设置。
20878 0
+关注
elim1
喜欢研究技术的码农
300
文章
0
问答
文章排行榜
最热
最新
相关电子书
更多
JS零基础入门教程(上册)
立即下载
性能优化方法论
立即下载
手把手学习日志服务SLS,云启实验室实战指南
立即下载