当当开源任务调度（作业框架）elastic-job-1.1.1试用

 

当当开源任务调度（作业框架）elastic-job-1.1.1试用

 

安装zookeeper

http://zookeeper.apache.org/doc/r3.4.6/zookeeperStarted.html

windows下的zookeeper启动脚本（我用的是UBUNTU，未测试这个脚本）

来源：http://www.ibm.com/developerworks/cn/opensource/os-cn-zookeeper/index.html

 setlocal 

 set ZOOCFGDIR=%~dp0%..\conf 

 set ZOO_LOG_DIR=%~dp0%.. 

 set ZOO_LOG4J_PROP=INFO,CONSOLE 

 set CLASSPATH=%ZOOCFGDIR% 

 

 set CLASSPATH=%~dp0..\*;%~dp0..\lib\*;%CLASSPATH% 

 set CLASSPATH=%~dp0..\build\classes;%~dp0..\build\lib\*;%CLASSPATH% 

 set ZOOCFG=%ZOOCFGDIR%\zoo.cfg 

 set ZOOMAIN=org.apache.zookeeper.server.ZooKeeperServerMain 

 java "-Dzookeeper.log.dir=%ZOO_LOG_DIR%" "-Dzookeeper.root.logger=%ZOO_LOG4J_PROP%" 

 -cp "%CLASSPATH%" %ZOOMAIN% "%ZOOCFG%" %* 

 endlocal

 

 

添加Maven依赖（spring依赖什么的就不说了）：

  <!-- 引入elastic-job核心模块 -->

<dependency>

   <groupId>com.dangdang</groupId>

   <artifactId>elastic-job-core</artifactId>

   <version>1.1.1</version>

</dependency>

 

<!-- 使用springframework自定义命名空间时引入 -->

<dependency>

   <groupId>com.dangdang</groupId>

   <artifactId>elastic-job-spring</artifactId>

   <version>1.1.1</version>

</dependency>

 

可能会出现构件找不到。如果你使用的是nexus本地maven仓库，可以登录web上去（例如http://192.168.1.250:8081/nexus/#view-repositories;central~browsestorage），删除对应的文件夹然后再尝试。

如果还是不行可以试试maven的-U参数 强制更新。

 

 

Spring配置文件方式注册作业：zookeeper的ip需要修改，命名空间dd-job会字典在zookeeper里创建。

 

<beans xmlns="http://www.springframework.org/schema/beans" 

    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.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

    ">

 

 <!--配置作业注册中心 -->

    <reg:zookeeper id="regCenter" server-lists="192.168.1.251:2181" namespace="dd-job" base-sleep-time-milliseconds="1000" max-sleep-time-milliseconds="3000" max-retries="3" />

 

    <!-- 配置作业-->

    <job:simple id="myElasticJob" class="test.MyElasticJob" registry-center-ref="regCenter" cron="0/10 * * * * ?"   sharding-total-count="3" sharding-item-parameters="0=A,1=B,2=C" />

 

</beans>

 

作业类test.MyElasticJob：

package test;

 

import java.util.List;

import java.util.Map;

 

import com.dangdang.ddframe.job.api.JobExecutionMultipleShardingContext;

import com.dangdang.ddframe.job.plugin.job.type.simple.AbstractSimpleElasticJob;

 

public class MyElasticJob extends AbstractSimpleElasticJob {

public MyElasticJob() {

System.out.println("MyElasticJob");

}

 

    @Override

    public void process(JobExecutionMultipleShardingContext context) {

        System.out.println("context:"+context);

//        String param=context.getJobParameter();

//        Map<Integer, String> map=context.getShardingItemParameters();

//        List<Integer> list=context.getShardingItems();

//        String name=context.getJobName();

//        Map<Integer, String> offset=context.getOffsets();

//        System.out.println(System.currentTimeMillis()/1000+":"+param+",map:"+map+",list:"+list+",name:"+name+",offset:"+offset);

    }

}

 

sping加载类：

package test;

 

import org.springframework.context.ApplicationContext;

import org.springframework.context.support.ClassPathXmlApplicationContext;

 

public class MainBusinessProcess {

public static void main(String[] args) {

ApplicationContext ctx = new ClassPathXmlApplicationContext("classpath:spring-bean.xml");

System.out.println(ctx);

}

}

 

 

 

java方法方式注册作业：zookeeper的ip需要修改，命名空间elastic-job-example会字典在zookeeper里创建。

 

package test;

 

import com.dangdang.ddframe.job.api.JobScheduler;

import com.dangdang.ddframe.job.api.config.impl.DataFlowJobConfiguration;

import com.dangdang.ddframe.job.api.config.impl.DataFlowJobConfiguration.DataFlowJobConfigurationBuilder;

import com.dangdang.ddframe.job.api.config.impl.SimpleJobConfiguration;

import com.dangdang.ddframe.job.api.config.impl.SimpleJobConfiguration.SimpleJobConfigurationBuilder;

import com.dangdang.ddframe.reg.base.CoordinatorRegistryCenter;

import com.dangdang.ddframe.reg.zookeeper.ZookeeperConfiguration;

import com.dangdang.ddframe.reg.zookeeper.ZookeeperRegistryCenter;

 

public class JobRegDemo {

 

    // 定义Zookeeper注册中心配置对象

    private ZookeeperConfiguration zkConfig = new ZookeeperConfiguration("192.168.1.251:2181", "elastic-job-example", 1000, 3000, 3);

     

    // 定义Zookeeper注册中心

    private CoordinatorRegistryCenter regCenter = new ZookeeperRegistryCenter(zkConfig);

     

    // 定义作业1配置对象

    private SimpleJobConfigurationBuilder jobConfig1build = new SimpleJobConfiguration.SimpleJobConfigurationBuilder("simpleJobDemo", SimpleJobDemo.class, 10, "0/5 * * * * ?");

     

    // 定义作业2配置对象

    private DataFlowJobConfigurationBuilder jobConfig2build = new DataFlowJobConfiguration.DataFlowJobConfigurationBuilder("dataFlowElasticJobDemo", DataFlowElasticJobDemo.class, 10, "0/5 * * * * ?");

     

    // 定义作业3配置对象

    //private JobConfiguration jobConfig3build = new JobConfiguration("sequencePerpetualElasticDemoJob", SequencePerpetualElasticDemoJob.class, 10, "0/5 * * * * ?");

     

    public static void main(final String[] args) {

        new JobRegDemo().init();

    }

     

    private void init() {

        // 连接注册中心

        regCenter.init();

        // 启动作业1

        new JobScheduler(regCenter, jobConfig1build.build()).init();

        // 启动作业2

        new JobScheduler(regCenter, jobConfig2build.build()).init();

        // 启动作业3

        //new JobScheduler(regCenter, jobConfig3build.build()).init();

    }

}

 

 

 

作业类test.DataFlowElasticJobDemo：

package test;

 

import java.util.List;

import java.util.concurrent.ExecutorService;

 

import com.dangdang.ddframe.job.api.DataFlowElasticJob;

import com.dangdang.ddframe.job.api.JobExecutionMultipleShardingContext;

import com.dangdang.ddframe.job.exception.JobException;

import com.dangdang.ddframe.job.internal.schedule.JobFacade;

 

public class DataFlowElasticJobDemo implements DataFlowElasticJob<String, JobExecutionMultipleShardingContext> {

 

@Override

public void execute() {

System.out.println("DataFlowElasticJobDemo");

}

 

@Override

public void handleJobExecutionException(JobException jobException) {

// TODO Auto-generated method stub

 

}

 

@Override

public JobFacade getJobFacade() {

// TODO Auto-generated method stub

return null;

}

 

@Override

public void setJobFacade(JobFacade jobFacade) {

// TODO Auto-generated method stub

 

}

 

@Override

public List<String> fetchData(JobExecutionMultipleShardingContext shardingContext) {

// TODO Auto-generated method stub

return null;

}

 

@Override

public void updateOffset(int item, String offset) {

// TODO Auto-generated method stub

 

}

 

@Override

public ExecutorService getExecutorService() {

// TODO Auto-generated method stub

return null;

}

 

}

 

 

 

作业类test.SimpleJobDemo：

package test;

 

import java.util.List;

import java.util.Map;

 

import com.dangdang.ddframe.job.api.JobExecutionMultipleShardingContext;

import com.dangdang.ddframe.job.plugin.job.type.simple.AbstractSimpleElasticJob;

 

public class SimpleJobDemo extends AbstractSimpleElasticJob {

public SimpleJobDemo() {

System.out.println("SimpleJobDemo");

}

 

    @Override

    public void process(JobExecutionMultipleShardingContext context) {

        System.out.println("context:"+context);

    }

}

 

 

 

两个main函数都可以运行，运行后作业会定时执行。

 

 

源码：https://github.com/dangdangdotcom/elastic-job

其中elastic-job-console是监控台，web界面的，可以用来监控和操作当前正在运行的作业。

maven打包后可以生成elastic-job-console-1.1.1.war。

放到tomcat可以直接访问http://localhost:8080/elastic-job-console-1.1.1

用户密码都是root

进入后，注册中心里连接zookeeper（需要输入ip端口和命名空间），即可查看之前两种方法运行的三个job。

注意有两个命名空间dd-job（1个作业）和elastic-job-example（2个作业）。

 

一台电脑看不出分片与否，因为都分到一个电脑上了。

一个以上就可以从context的shardingItems看出区别。一台电脑挂了分片会重新进行。这个开源的作业框架还是很好用的。