Elastic Job 入門

阿新 • • 發佈：2017-10-15

job elastic 實現入門

Elastic job是當當網架構師張亮,曹昊和江樹建基於Zookepper、Quartz開發並開源的一個Java分布式定時任務，解決了Quartz不支持分布式的弊端。Elastic job主要的功能有支持彈性擴容，通過Zookepper集中管理和監控job，支持失效轉移等，這些都是Quartz等其他定時任務無法比擬的。

目前Elastic job的最新版本已經由原來的elastic-job-core分離除了兩個項目，分別為Elastic-Job-Lite和Elastic-Job-Cloud。Elastic-Job是一個分布式調度解決方案，由兩個相互獨立的子項目Elastic-Job-Lite和Elastic-Job-Cloud組成，Elastic-Job-Lite定位為輕量級無中心化解決方案，使用jar包的形式提供分布式任務的協調服務。 Elastic-Job-Cloud使用Mesos + Docker(TBD)的解決方案，額外提供資源治理、應用分發以及進程隔離等服務，Elastic-Job-Lite和Elastic-Job-Cloud提供同一套API開發作業，開發者僅需一次開發，即可根據需要以Lite或Cloud的方式部署

開頭copy 網上一段術語，講解的很清晰。總結就是：elastic job 是用來實現分布式的定時任務，比如說定時任務項目，部署在2臺，但是只需要執行一次，而又想保持2臺服務器代碼都一致，這時候elasitc job 可以很完美的解決問題，當然使用的是zookeeper的相關特性。

1.實戰

項目目錄如下

$($N1Y~4P2YP_L_UT~{A)S95.png$

1.pom.xml

新建maven項目，引入springboot的相關jar包

再引入elastic job 的相關核心包 elastic-job-core，elastic-job-spring

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>

  <groupId>com.xj</groupId>
  <artifactId>el</artifactId>
  <version>0.0.1-SNAPSHOT</version>
  <packaging>jar</packaging>

  <name>el</name>
  <url>http://maven.apache.org</url>

  <properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  </properties>

 <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>1.4.0.RELEASE</version>
    </parent>



  <dependencies>
    <dependency>
      <groupId>junit</groupId>
      <artifactId>junit</artifactId>
      <version>3.8.1</version>
      <scope>test</scope>
    </dependency>
    
       <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    
    
    <!-- 引入elastic-job-core核心模塊 -->  
<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>


  </dependencies>
  
  
  <build>
    <plugins>
      <plugin>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-maven-plugin</artifactId>
      </plugin>
    </plugins>
  </build>
  
  
  
</project>

2.reg.properties

存放zookeeper 配置中心的相關配置信息

serverLists=10.3.142.107:2181,10.3.142.107:2182,106.3.14.107:2183
namespace=elastic-job-example
baseSleepTimeMilliseconds=1000
maxSleepTimeMilliseconds=3000
maxRetries=3

3.withNamespace.xml

存放作業的相關詳情，這裏測試存放了 simple 與dataFlow 2中情況。

<?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 
                        ">
    
    <context:property-placeholder location="classpath:conf/*.properties" /> 
     
    <!--配置作業註冊中心 -->  
    <reg:zookeeper id="regCenter" server-lists="${serverLists}" namespace="${namespace}"  
                   base-sleep-time-milliseconds="${baseSleepTimeMilliseconds}" max-sleep-time-milliseconds="${baseSleepTimeMilliseconds}" max-retries="${maxRetries}" />  
  
    <!-- 配置作業-->  
    <job:simple id="mySimpleJob" class="com.xj.el.MySimpleJob" registry-center-ref="regCenter"  
                sharding-total-count="2" cron="0/3 * * * * ?" overwrite="true" />  
  
  
   
    <job:dataflow id="throughputDataFlow"  class="com.xj.el.MyThroughputDataFlowElasticJob" registry-center-ref="regCenter"
     cron="0/10 * * * * ?"  sharding-total-count="3"   sharding-item-parameters="0=A,1=B,2=C"
     process-count-interval-seconds= "10"  concurrent-data-process-thread-count="10"/>
  
</beans>

elastic-job提供了三種類型的作業：Simple類型作業、Dataflow類型作業、Script類型作業。這裏主要講解前兩者。Script類型作業意為腳本類型作業，支持shell，python，perl等所有類型腳本，使用不多，可以參見github文檔。

SimpleJob需要實現SimpleJob接口，意為簡單實現，未經過任何封裝，與quartz原生接口相似，比如示例代碼中所使用的job。

Dataflow類型用於處理數據流，需實現DataflowJob接口。該接口提供2個方法可供覆蓋，分別用於抓取(fetchData)和處理(processData)數據。
可通過DataflowJobConfiguration配置是否流式處理。
流式處理數據只有fetchData方法的返回值為null或集合長度為空時，作業才停止抓取，否則作業將一直運行下去；非流式處理數據則只會在每次作業執行過程中執行一次fetchData方法和processData方法，隨即完成本次作業。
實際開發中，Dataflow類型的job還是很有好用的。

4.MySimpleJob.java

package com.xj.el;

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

public class MySimpleJob extends AbstractSimpleElasticJob{

	@Override
	public void process(JobExecutionMultipleShardingContext shardingContext) {
		  System.out.println(String.format("------Thread ID: %s, 任務總片數: %s, 當前分片項: %s",  
		 Thread.currentThread().getId(), shardingContext.getShardingTotalCount(), shardingContext.getShardingItems()));  
	}	

}

5.MyThroughputDataFlowElasticJob.java

package com.xj.el;
import java.sql.Time;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

import com.dangdang.ddframe.job.api.JobExecutionMultipleShardingContext;
import com.dangdang.ddframe.job.internal.job.AbstractJobExecutionShardingContext;
import com.dangdang.ddframe.job.plugin.job.type.dataflow.AbstractIndividualThroughputDataFlowElasticJob;


public class MyThroughputDataFlowElasticJob extends AbstractIndividualThroughputDataFlowElasticJob<TestBean>{
		
	public List<TestBean> fetchData(JobExecutionMultipleShardingContext shardingContext) {
		List<TestBean> testBeanList = new ArrayList();
	     //獲取testBean的相關數據
		return testBeanList;
	}

	
	public boolean processData(JobExecutionMultipleShardingContext shardingContext, TestBean data) {
		//處理上文testBean的相關數據
		return false;
	}
	
}

6.springboot 測試主類 SpringBootSampleApplication.java

package com.xj.el;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.context.annotation.ImportResource;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * Hello world!
 *
 */
@RestController
@EnableAutoConfiguration
@ImportResource(locations = {"classpath:withNamespace.xml"})
public class SpringBootSampleApplication 
{
	 public static void main(String[] args) {
	        SpringApplication.run(SpringBootSampleApplication.class, args);
	    }
	 
	 
	@RequestMapping("/")
    String home() {
        return "Hello World!";
    }
}

本文出自 “布拉君君” 博客，請務必保留此出處http://5148737.blog.51cto.com/5138737/1972498

Elastic Job 入門

job elastic 實現入門 Elastic job是當當網架構師張亮,曹昊和江樹建基於Zookepper、Quartz開發並開源的一個Java分布式定時任務，解決了Quartz不支持分布式的弊端。Elastic job主要的功能有支持彈性擴容，通過Zookepper集中管理和監控job，

Elastic Job 入門

1.實戰

Elastic Job 入門

Elastic Job入門（1） - 簡介

Elastic Job入門（3） - 整合Springboot

Elastic Job 入門教程（二）— Spring Boot框架下是實現Elastic Job 指令碼作業（Script Job）

Elastic Job 入門教程（四）— 事件追蹤

elastic-job入門例項

Elastic-job入門介紹

Elastic Job入門示例-實現DataflowJob介面

Elastic Job入門

快速入門Elastic-Job

Elastic-Job定時入門

elastic-job詳解（一）：數據分片

elastic-job詳解（三）：Job的手動觸發功能

elastic-job詳解（四）：失效轉移

分布式任務elastic-job

elastic job簡單用法

Elastic-Job-Lite 源碼分析 —— 運維平臺

Elastic Job學習

Elastic-Job-Lite 源碼分析 —— 作業分片策略

Elastic-Job開發指南

Elastic Job 入門

1.實戰

相關推薦