Elastic-Job之简单Job

我就是我 2022-06-04 06:29 355阅读 0赞

## 简介 ##

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

## 核心概念 ##

*  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命名空间，示例如下：

### 配置注册中心 ###

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>
    
        
        
        <reg:zookeeper id="regCenter" server-lists="localhost:2181" namespace="dd-job" base-sleep-time-milliseconds="1000" max-sleep-time-milliseconds="3000" max-retries="3" />
    
        
    
        
        
    
        
        
        <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需要定义在`<job:simple/>`的前面，否则会提示找不到对应的bean定义。

（本文由Elim写于2017年10月1日）

[https_github.com_dangdangdotcom_elastic-job]: https://github.com/dangdangdotcom/elastic-job
[http_elasticjob.io]: http://elasticjob.io/