elastic-job入門例項

阿新 • • 發佈：2018-12-30

說明

Elastic-Job是一個分散式排程解決方案，由兩個相互獨立的子專案Elastic-Job-Lite和Elastic-Job-Cloud組成。

Elastic-Job-Lite定位為輕量級無中心化解決方案，使用jar包的形式提供分散式任務的協調服務；Elastic-Job-Cloud採用自研Mesos Framework的解決方案，額外提供資源治理、應用分發以及程序隔離等功能。

功能列表

1. 任務分片

將整體任務拆解為多個子任務
可通過伺服器的增減彈性伸縮任務處理能力
分散式協調，任務伺服器上下線的全自動發現與處理

2. 多工型別

基於時間驅動的任務
基於資料驅動的任務（TBD）

同時支援常駐任務和瞬時任務
多語言任務支援

3. 雲原生

完美結合Mesos或Kubernetes等排程平臺
任務不依賴於IP、磁碟、資料等有狀態元件
合理的資源排程，基於Netflix的Fenzo進行資源分配

4. 容錯性

支援定時自我故障檢測與自動修復
分散式任務分片唯一性保證
支援失效轉移和錯過任務重觸發

5. 任務聚合

相同任務聚合至相同的執行器統一處理
節省系統資源與初始化開銷
動態調配追加資源至新分配的任務

6. 易用性

完善的運維平臺
提供任務執行歷史資料追蹤能力
註冊中心資料一鍵dump用於備份與除錯問題

接下來我們就開始實現一個小例子

構建工具

gradle

專案結構如下

這裡寫圖片描述

引入依賴

在build.gradle檔案中

//elastic-job
            [group: 'com.dangdang', name: 'elastic-job-lite-core', version: '2.1.5'],
            [group: 'com.dangdang', name: 'elastic-job-lite-spring', version: '2.1.5']

SimpleJob 簡單作業

import com.dangdang.ddframe.job.api.ShardingContext;
import 
 com.dangdang.ddframe.job.api.simple.SimpleJob;

public class MyElasticSimpleJob implements SimpleJob{

    @Override
    public void execute(ShardingContext context) {
        switch (context.getShardingItem()) {
            case 0: 
                System.out.println("do something by sharding item 0"); 
                break;
            case 1: 
                System.out.println("do something by sharding item 1"); 
                break;
            case 2: 
                System.out.println("do something by sharding item 2"); 
                break;
            // case n: ...
        }
    }

}

DataFlowJob 資料流作業

import java.util.ArrayList;
import java.util.List;

import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.api.dataflow.DataflowJob;

public class MyElasticDataflowJob implements DataflowJob<String>{

    @Override
    public List<String> fetchData(ShardingContext context) {
        switch (context.getShardingItem()) {
            case 0: 
                // get data from database by sharding item 0
                List<String> data1 = new ArrayList<>();
                data1.add("get data from database by sharding item 0");
                return data1;
            case 1: 
                // get data from database by sharding item 1
                List<String> data2 = new ArrayList<>();
                data2.add("get data from database by sharding item 1");
                return data2;
            case 2: 
                // get data from database by sharding item 2
                List<String> data3 = new ArrayList<>();
                data3.add("get data from database by sharding item 2");
                return data3;
            // case n: ...
        }
        return null;
    }

    @Override
    public void processData(ShardingContext shardingContext, List<String> data) {
        int count=0;
        // process data
        // ...
        for (String string : data) {
            count++;
            System.out.println(string);
            if (count>10) {
                return;
            }
        }

    }

}

測試以上兩種作業

import java.net.InetAddress;

import java.net.UnknownHostException;

import com.dangdang.ddframe.job.api.dataflow.DataflowJob;
import com.dangdang.ddframe.job.api.simple.SimpleJob;
import com.dangdang.ddframe.job.config.JobCoreConfiguration;
import com.dangdang.ddframe.job.config.JobRootConfiguration;
import com.dangdang.ddframe.job.config.dataflow.DataflowJobConfiguration;
import com.dangdang.ddframe.job.config.script.ScriptJobConfiguration;
import com.dangdang.ddframe.job.config.simple.SimpleJobConfiguration;
import com.dangdang.ddframe.job.lite.api.JobScheduler;
import com.dangdang.ddframe.job.lite.config.LiteJobConfiguration;
import com.dangdang.ddframe.job.reg.base.CoordinatorRegistryCenter;
import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperConfiguration;
import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperRegistryCenter;
import com.job.task.MyElasticDataflowJob;
import com.job.task.MyElasticSimpleJob;

public class JobDemo {

    public static void main(String[] args) throws UnknownHostException {
        System.out.println("Start...");
        System.out.println(InetAddress.getLocalHost());
        new JobScheduler(createRegistryCenter(), createSimpleJobConfiguration()).init();
        new JobScheduler(createRegistryCenter(), createDataflowJobConfiguration()).init();
    }

    private static CoordinatorRegistryCenter createRegistryCenter() {
        CoordinatorRegistryCenter regCenter = new ZookeeperRegistryCenter(
                new ZookeeperConfiguration("127.0.0.1:2181", "new-elastic-job-demo"));
        regCenter.init();
        return regCenter;
    }

    private static LiteJobConfiguration createSimpleJobConfiguration() {
        // 定義作業核心配置
        JobCoreConfiguration simpleCoreConfig = JobCoreConfiguration.newBuilder("SimpleJobDemo", "0/15 * * * * ?", 10).build();
        // 定義SIMPLE型別配置
        SimpleJobConfiguration simpleJobConfig = new SimpleJobConfiguration(simpleCoreConfig, MyElasticSimpleJob.class.getCanonicalName());
        // 定義Lite作業根配置
        JobRootConfiguration simpleJobRootConfig = LiteJobConfiguration.newBuilder(simpleJobConfig).build();

        return (LiteJobConfiguration) simpleJobRootConfig;

    }

    private static LiteJobConfiguration createDataflowJobConfiguration() {
        // 定義作業核心配置
        JobCoreConfiguration dataflowCoreConfig = JobCoreConfiguration.newBuilder("DataflowJob", "0/30 * * * * ?", 10).build();
        // 定義DATAFLOW型別配置
        DataflowJobConfiguration dataflowJobConfig = new DataflowJobConfiguration(dataflowCoreConfig, MyElasticDataflowJob.class.getCanonicalName(), true);
        // 定義Lite作業根配置
        JobRootConfiguration dataflowJobRootConfig = LiteJobConfiguration.newBuilder(dataflowJobConfig).build();

        return (LiteJobConfiguration) dataflowJobRootConfig;
    }
}

執行結果

這裡寫圖片描述

現在我們通過配置檔案的方式來實現兩種型別的作業

建立elastic.xml配置檔案

將elastic-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: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 
                        ">

    <!-- 配置作業註冊中心; baseSleepTimeMilliseconds:等待重試的間隔時間的初始值單位：毫秒 ; 
    maxSleepTimeMilliseconds:等待重試的間隔時間的最大值單位：毫秒;maxRetries:最大重試次數 -->
    <reg:zookeeper id="regCenter" server-lists="192.168.6.175:12181"
        namespace="elastic-job" base-sleep-time-milliseconds="1000"
        max-sleep-time-milliseconds="3000" max-retries="3" />

    <!-- 配置簡單作業 -->
    <job:simple id="JobSimpleJob" class="com.job.task.MyElasticSimpleJob"
        registry-center-ref="regCenter" cron="0/30 * * * * ?"
        sharding-total-count="3" sharding-item-parameters="0=A,1=B,2=C" />

    <!-- 配置資料流作業, job-parameter定義的為分頁引數 
    sharding-total-count 作業分片總數
    sharding-item-parameters分片序列號和引數用等號分隔，多個鍵值對用逗號分隔 ,分片序列號從0開始，不可大於或等於作業分片總數
    job-parameter 作業自定義引數，可通過傳遞該引數為作業排程的業務方法傳參，用於實現帶引數的作業
    例：每次獲取的資料量、作業例項從資料庫讀取的主鍵等
    job-sharding-strategy-class 作業分片策略實現類全路徑 預設使用平均分配策略
    streaming-process 是否流式處理資料
    reconcile-interval-minutes 修復作業伺服器不一致狀態服務排程間隔時間，配置為小於1的任意值表示不執行修復
    event-trace-rdb-data-source 作業事件追蹤的資料來源Bean引用
    -->

        <job:dataflow id="JobDataflow" class="com.job.task.MqElasticDataflowJob" 
        registry-center-ref="regCenter" cron="0/10 * * * * ?" sharding-total-count="3" 
        sharding-item-parameters="0=a,1=b,2=c" job-sharding-strategy-class="com.dangdang.ddframe.job.lite.api.strategy.impl.AverageAllocationJobShardingStrategy" 
        job-parameter="100" streaming-process="true" reconcile-interval-minutes="10" 
        overwrite="true" event-trace-rdb-data-source="dataSource"/> 

</beans>

配置datasource

<?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:util="http://www.springframework.org/schema/util"
    xmlns:tx="http://www.springframework.org/schema/tx"
    xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx.xsd">

    <bean id="dataSource" class="com.mchange.v2.c3p0.ComboPooledDataSource"
        destroy-method="close">
        <property name="driverClass" value="com.mysql.jdbc.Driver" />
        <property name="jdbcUrl" value="jdbc:mysql://127.0.0.1:3306/for_test?useUnicode=yes&amp;characterEncoding=UTF-8" />
        <property name="user" value="admin" />
        <property name="password" value="super" />
        <property name="minPoolSize" value="3" />
        <property name="maxPoolSize" value="20" />
        <property name="acquireIncrement" value="1" />
        <property name="testConnectionOnCheckin" value="true" />
        <property name="maxIdleTimeExcessConnections" value="240" />
        <property name="idleConnectionTestPeriod" value="300" />
    </bean>


</beans>

建立applicationContext.xml檔案

將elastic-job與Spring整合

<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:task="http://www.springframework.org/schema/task"
    xsi:schemaLocation="http://www.springframework.org/schema/beans
        http://www.springframework.org/schema/beans/spring-beans-3.2.xsd
        http://www.springframework.org/schema/context
        http://www.springframework.org/schema/context/spring-context.xsd
        http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd ">

    <task:scheduler id="taskScheduler" pool-size="10" />
    <task:executor id="taskExecutor" />
    <task:annotation-driven executor="taskExecutor" scheduler="taskScheduler" />

    <import resource="elastic.xml" />
    <import resource="mysql.xml"/>
</beans>

配置web.xml

<?xml version="1.0" encoding="UTF-8"?>
<web-app version="2.5" xmlns="http://java.sun.com/xml/ns/javaee" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_2_5.xsd">

    <display-name>elastic-job</display-name>

    <!-- 用來設定web應用的環境引數(context) -->
    <context-param>
        <param-name>contextConfigLocation</param-name>
        <param-value>classpath:applicationContext.xml</param-value>
    </context-param>

    <!-- listener元素用來定義Listener介面,對事件監聽程式 -->
    <listener>
        <listener-class>
            org.springframework.web.context.ContextLoaderListener
        </listener-class>
    </listener>

</web-app>

運作結果

這裡寫圖片描述

elastic-job入門例項

說明 Elastic-Job是一個分散式排程解決方案，由兩個相互獨立的子專案Elastic-Job-Lite和Elastic-Job-Cloud組成。 Elastic-Job-Lite定位為輕量級無中心化解決方案，使用jar包的形式提供分散式任務的

Elastic Job 入門

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

Elastic Job入門（1） - 簡介

介紹構建一般的業務系統來說，使用 Quartz或者 Spring Task即可基本滿足我們的單體服用應用需要。然而隨著線上業務量的不斷髮展，這兩種定時任務已經日漸無法滿足我們的需求。一般，使用這兩種定時任務框架都會遇到如下的兩個痛點問題： 1、如果業務工程採用叢集化的部署，可能會多

Elastic Job入門（3） - 整合Springboot

引入pom檔案　　　　 <dependency> <groupId>com.dangdang</groupId> <artifactId>elastic-job-lite-core</artifa

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

在Elastic Job 入門教程（一）— 與Spring Boot整合這篇文章中，我們簡單介紹了Spring Boot與Elastic Job 的整合，並簡單實現了SimpleJob型別作業。本章，我

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

Elastic-Job提供了事件追蹤功能，用於查詢、統計和監控作業執行歷史和執行狀態。Elastic-Job-Lite在配置中提供了JobEventConfiguration，目前支援資料庫方式配置。事

Elastic-job入門介紹

Elastic-Job是一個分散式的定時任務框架基於成熟的定時任務作業框架Quartz cron表示式執行定時任務相對於Quartz它增加了任務分片。這樣能有效的防止單節點部署專案帶來的種種不便

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

1.接上篇內容：https://blog.csdn.net/seanme/article/details/802564602.本次介紹流式處理任務型別 * 流式任務型別：業務實現兩個介面：抓取(fetchData)和處理(processData)資料 * a.流式處

Elastic Job入門

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

快速入門Elastic-Job

0.前言　　本案例基於springboot 1.5.9 1.maven依賴 1 <dependency> 2 <groupId>com.dangdang</groupId> 3 <artifactId>elastic-job-li

Elastic-Job定時入門

主要功能定時任務：基於成熟的定時任務作業框架Quartz cron表示式執行定時任務。作業註冊中心：基於Zookeeper和其客戶端Curator實現的全域性作業註冊控制中心。用於註冊，控制和協調分散式作業執行。作業分片：將一個任務分片成為多個

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

count 任務不同的應該 center shc 偶數 int ext 數據分片的目的在於把一個任務分散到不同的機器上運行，既可以解決單機計算能力上限的問題，也能降低部分任務失敗對整體系統的影響。elastic-job並不直接提供數據處理的功能，框架只會將分片項分配至各

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

方法 idt image blog per tle cnblogs ack display elastic-job的任務都是使用quartz來觸發的，quartz表達式一般都是定期執行。但有時候一些周期較長的任務，比如一天一次，幾小時一次的任務，我們需要等待很久才能觸發一次

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

shard out utm monit 設置 borde 點滴 title 等於 elastic-job中最關鍵的特性之一就是失效轉移。配置了失效轉移之後，如果在任務執行過程中有一個執行實例掛了，那麽之前被分配到這個實例的任務（或者分片）會在下次任務執行之前被重新分配到其他

分布式任務elastic-job

idata blob vnr mas 編譯安裝當當 pre cor clas # elastic-job簡介目前Elastic job的最新版本已經由原來的elastic-job-core分離除了兩個項目，分別為Elastic-Job-Lite和Elastic-Job-

elastic job簡單用法

uil rman source simple pat 名稱 cat out ont public class JobMain { //配置註冊中心 private ZookeeperConfiguration zkConfig = new Zookeepe

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

job 哈哈哈配置服務 posit rop cat jet gis nmap 　　本文基於 Elastic-Job V2.1.5 版本分享　　　　1. 概述　　　　2. Maven模塊 elastic-job-common-restful　　　　3. Maven模塊 el

Elastic Job學習

soft clas quic body bsp 分享 elastic mic log 從執行的結果來看，隨著服務器的增加，分片的結果也不一樣當只有一臺服務器的時候，所有分片都到這一臺服務器上當服務器增加到兩臺的時候，分片的結果是0 1 2 3 4和5 6 7

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

哈希 AD hash alloc hub strings put iat 總數摘要: 原創出處 http://www.iocoder.cn/Elastic-Job/job-sharding-strategy/ 「芋道源碼」歡迎轉載，保留摘要，謝謝！本文基於 Elast

Spring Boot入門例項

簡介 Spring Boot是由Pivotal團隊提供的全新框架，其設計目的是用來簡化新Spring應用的初始搭建以及開發過程。該框架使用了特定的方式來進行配置，從而使開發人員不再需要定義樣板化的配置。通過這種方式，Boot致力於在蓬勃發展的快速應用開發領域（rapid application

elastic-job入門例項

說明

功能列表

1. 任務分片

2. 多工型別

3. 雲原生

4. 容錯性

5. 任務聚合

6. 易用性

構建工具

專案結構如下

引入依賴

SimpleJob 簡單作業

DataFlowJob 資料流作業

測試以上兩種作業

執行結果

建立elastic.xml配置檔案

配置datasource

建立applicationContext.xml檔案

配置web.xml

運作結果

相關推薦