2. 程式人生 > >分布式任務elastic-job

分布式任務elastic-job

阿新 • • 發佈：2017-12-04
idata blob vnr mas 編譯安裝 當當 pre cor clas

# elastic-job簡介
目前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的方式部署 .[轉自官網:https://github.com/dangdangdotcom/elastic-job/blob/master/README_cn.md]

一般的技術quartz、spring task、java.util.Timer，這幾種如果在單一機器上跑其實問題不大，但是如果一旦應用於集群環境做分布式部署，就會帶來一個致命的問題，那就是重復執行，當然解決方案有，但是必須依賴數據庫，將任務執行狀態持久化下來。所以當當就把quartz和zookeeper結合起來達到分布式調度，並且添加其他功能，形成了elastic-job。

elastic-job主要的設計理念是無中心化的分布式定時調度框架，思路來源於Quartz的基於數據庫的高可用方案。但數據庫沒有分布式協調功能，所以在高可用方案的基礎上增加了彈性擴容和數據分片的思路，以便於更大限度的利用分布式服務器的資源。


# 功能

1. 主要功能

a) 分布式：重寫Quartz基於數據庫的分布式功能，改用Zookeeper實現註冊中心。

b) 並行調度：采用任務分片方式實現。將一個任務拆分為n個獨立的任務項，由分布式的服務器並行執行各自分配到的分片項。

c) 彈性擴容縮容：將任務拆分為n個任務項後，各個服務器分別執行各自分配到的任務項。一旦有新的服務器加入集群，或現有服務器下線，elastic-job將在保留本次任務執行不變的情況下，下次任務開始前觸發任務重分片。

d) 集中管理：采用基於Zookeeper的註冊中心，集中管理和協調分布式作業的狀態，分配和監聽。外部系統可直接根據Zookeeper的數據管理和監控elastic-job。

e) 定制化流程型任務：作業可分為簡單和數據流處理兩種模式，數據流又分為高吞吐處理模式和順序性處理模式，其中高吞吐處理模式可以開啟足夠多的線程快速的處理數據，而順序性處理模式將每個分片項分配到一個獨立線程，用於保證同一分片的順序性，這點類似於kafka的分區順序性。

2. 其他功能

a) 失效轉移：彈性擴容縮容在下次作業運行前重分片，但本次作業執行的過程中，下線的服務器所分配的作業將不會重新被分配。失效轉移功能可以在本次作業運行中用空閑服務器抓取孤兒作業分片執行。同樣失效轉移功能也會犧牲部分性能。

b) Spring命名空間支持：elastic-job可以不依賴於spring直接運行，但是也提供了自定義的命名空間方便與spring集成。

c) 運維平臺：提供web控制臺用於管理作業。

下載源碼
https://github.com/elasticjob


#spring boot quick start
Add maven dependency

<!-- https://mvnrepository.com/artifact/com.dangdang/elastic-job-lite-core --> 
<!-- import elastic-job lite core -->
<dependency>
<groupId>com.dangdang</groupId>
<artifactId>elastic-job-lite-core</artifactId>
<version>2.1.6</version>
</dependency>

<!-- import other module if need -->
<dependency>
<groupId>com.dangdang</groupId>
<artifactId>elastic-job-lite-spring</artifactId>
<version>2.1.6</version>
</dependency>

RegCenter configuration

@Configuration
public class RegistryCenterConfig {

    @Bean(initMethod = "init")
    public ZookeeperRegistryCenter regCenter() {
        String serverList = ConfigOne.getProperty(ConfigConstants.JOB_REG_ZK);
        String namespace = ConfigOne.getProperty(ConfigConstants.JOB_REG_NS);
        return new ZookeeperRegistryCenter(new ZookeeperConfiguration(serverList, namespace));
    }

}


Job configuration

@Configuration
public class SimpleJobConfig {

    //private String defaultCron = "0/5 * * * * ?";
    //默認每天0點10分開始統計
    @Value("${job.default.cron}")
    private String defaultCron = "0 10 0 * * ?";
    @Value("${job.default.shardingTotalCount}")
    private int defaultShardTotal = 1;
    @Value("${job.default.shardingItemParameters}")
    private String defaultShardPrams = "";

    @Resource
    private ZookeeperRegistryCenter regCenter;

    /*@Resource
    private JobEventConfiguration jobEventConfiguration;*/

    @Resource
    private StatsDeviceJob statsDeviceJob;

    @Resource
    private StatsUserJob statsUserJob;

    @Resource
    private StatsDeviceFaultJob statsDeviceFaultJob;

    @Resource
    private StatsDeviceAlarmJob statsDeviceAlarmJob;

    @Resource
    private StatsRuleJob statsRuleJob;

    @Resource
    private StatsYumairJob statsYumairJob;


    //@Bean(initMethod = "init")
    @PostConstruct
    public void init() {
        //statsDeviceJob
        new SpringJobScheduler(statsDeviceJob, regCenter,
                getLiteJobConfiguration(statsDeviceJob.getClass(), defaultCron, defaultShardTotal, defaultShardPrams)).init();
        //statsUserJob
        new SpringJobScheduler(statsUserJob, regCenter,
                getLiteJobConfiguration(statsUserJob.getClass(), defaultCron, defaultShardTotal, defaultShardPrams)).init();
        //statsDeviceFaultJob
        new SpringJobScheduler(statsDeviceFaultJob, regCenter,
                getLiteJobConfiguration(statsDeviceFaultJob.getClass(), defaultCron, defaultShardTotal, defaultShardPrams)).init();
        //statsDeviceAlarmJob
        new SpringJobScheduler(statsDeviceAlarmJob, regCenter,
                getLiteJobConfiguration(statsDeviceAlarmJob.getClass(), defaultCron, defaultShardTotal, defaultShardPrams)).init();
        //statsRuleJob
        new SpringJobScheduler(statsRuleJob, regCenter,
                getLiteJobConfiguration(statsRuleJob.getClass(), defaultCron, defaultShardTotal, defaultShardPrams)).init();
        //statsYumairJob 每小時5分運行一次
        new SpringJobScheduler(statsYumairJob, regCenter,
                getLiteJobConfiguration(statsYumairJob.getClass(), "0 5 * * * ?", defaultShardTotal, defaultShardPrams)).init();

    }


    private LiteJobConfiguration getLiteJobConfiguration(final Class<? extends SimpleJob> jobClass, final String cron, final int defaultShardTotal, final String defaultShardPrams) {
        return LiteJobConfiguration.newBuilder(new SimpleJobConfiguration(JobCoreConfiguration.newBuilder(
                jobClass.getName(), cron, defaultShardTotal).shardingItemParameters(defaultShardPrams).build(), jobClass.getCanonicalName())).overwrite(true).build();
    }
}

Job development

@Service
public class StatsDeviceJob implements SimpleJob {
    private static Logger logger = LoggerFactory.getLogger(StatsDeviceJob.class);

    @Autowired
    private DeviceStatsFeignClient deviceStatsFeignClient;
    @Autowired
    private DeviceDataFeignClient deviceDataFeignClient;
    @Autowired
    private IDataStatsFacade dataStatsFacade;

    /**
     * 1.當分片數為1時,在同一個zookepper和jobname情況下,多臺機器部署了Elastic job時,
     * 只有拿到shardingContext.getShardingItem()為0的機器得以執行,其他的機器不執行
     * 2.當分片數大於1時,假如有3臺服務器，分成10片，則分片項分配結果為服務器A=0,1,2;服務器B=3,4,5;服務器C=6,7,8,9。
     * 此時每臺服務器可根據拿到的shardingItem值進行相應的處理
     * 目前job分片數全部置為1,即不使用分片
     * @param shardingContext
     */
    @Override
    public void execute(ShardingContext shardingContext) {
        logger.info(String.format("ShardingItem: %s | Thread: %s | %s",
                shardingContext.getShardingItem(), Thread.currentThread().getId(), "SIMPLE"));

        deviceStatsJob();
    }


    public void deviceStatsJob() {
        //TODO 
    }



 }

# 運維平臺和RESTFul API部署(可選)
1. 下載或者克隆elastic-job源碼
地址：https://github.com/dangdangdotcom/elastic-job


2. maven編譯安裝
進入到elastic-job目錄，按住Shift+鼠標右鍵，選擇“在此處打開命令窗口（W）”，執行如下命令：

```
mvn clean install -Dmaven.test.skip=true
```

```
[INFO] ------------------------------------------------------------------------
[INFO] Reactor Summary:
[INFO]
[INFO] elastic-job ....................................... SUCCESS [23.570s]
[INFO] elastic-job-common ................................ SUCCESS [0.053s]
[INFO] elastic-job-common-core ........................... SUCCESS [27.108s]
[INFO] elastic-job-common-restful ........................ SUCCESS [29.844s]
[INFO] elastic-job-lite .................................. SUCCESS [0.078s]
[INFO] elastic-job-lite-core ............................. SUCCESS [7.249s]
[INFO] elastic-job-lite-lifecycle ........................ SUCCESS [3.766s]
[INFO] elastic-job-lite-spring ........................... SUCCESS [1:08:42.613s
]
[INFO] elastic-job-lite-console .......................... SUCCESS [2:31.964s]
[INFO] elastic-job-cloud ................................. SUCCESS [0.031s]
[INFO] elastic-job-cloud-executor ........................ SUCCESS [3.728s]
[INFO] elastic-job-cloud-scheduler ....................... SUCCESS [33.803s]
[INFO] ------------------------------------------------------------------------
[INFO] BUILD SUCCESS
[INFO] ------------------------------------------------------------------------
[INFO] Total time: 1:13:24.136s
[INFO] Finished at: Sat Dec 02 18:33:33 CST 2017
[INFO] Final Memory: 55M/272M
[INFO] ------------------------------------------------------------------------
```


3. 解壓上一步打好的包

路徑：elastic-job\elastic-job-lite\elastic-job-lite-console\target\elastic-job-lite-console-2.1.6.tar.gz
elastic-job-lite-console-2.1.6\bin目錄下是啟動腳本

windows環境用：start.bat

linux環境用：start.sh

elastic-job-lite-console-2.1.6\conf目錄下是配置文件auth.properties，配置的用戶名和密碼

4. 解壓縮elastic-job-lite-console-${version}.tar.gz並執行bin\start.sh。

5. 打開瀏覽器訪問http://localhost:8899/即可訪問控制臺。8899為默認端口號，可通過啟動腳本輸入-p自定義端口號。

6. 訪問RESTFul API方法同控制臺。

分布式任務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開源項目的10

判斷 oba 早期 cep gin bean 社區 之前 都是 原文地址：https://www.cnblogs.com/qiumingcheng/p/5573845.html 作者簡介： 張亮，當當網架構師、當當技術委員會成員、消息中間件組負責人。對架構設計、分布式、

分布式任務調度平臺xxl-job的使用（java）

jdb 啟動 extends 好的 nic broadcast 心跳 return mode 　　xxl-job是一個開源的任務調度平臺（github地址https://github.com/xuxueli/xxl-job），且在分布式場景下有很好的支持性，可以對單個的任務

Celery 分布式任務隊列入門

beat 路徑 call 一行 祝福 app after height sch 一、Celery介紹和基本使用 Celery 是一個 基於python開發的分布式異步消息任務隊列，通過它可以輕松的實現任務的異步處理， 如果你的業務場景中需要用到異步任務，就可以考慮使用ce

Celery 分布式任務隊列快速入門

ade sunday reat 失敗 繼續 complete port 機器 single Celery介紹和基本使用 在項目中如何使用celery 啟用多個workers Celery 定時任務 與django結合 通過django配置celery period

spring quartz 分布式任務計劃

des red field and 文件 是我 網絡 ech startup 通過maven管理的spring mvc工程，且已經成功連接數據庫。數據庫表結構/*Table structure for table `qrtz_calendars` */ DROP TABLE

分布式任務隊列：Celery

== backend localhost blog 分布 關於 com edi 存儲 一、Celery 安裝與應用 關於 Celery 的內容，先依次看如下兩篇文章，讓你有個了解： http://blog.csdn.net/happyanger6/article/det

Quartz.Net分布式任務管理平臺（第二版）

增加 rem 技術分享 記錄 下載 探討 類型 cut 安裝 前言：在Quartz.Net項目發布第一版後，有挺多園友去下載使用，我們通過QQ去探討，其中項目中還是存在一定的不完善。所以有了現在這個版本。這個版本的編寫完成其實有段時間了一直沒有放上去。現在已經

Celery分布式任務

覆蓋 嘗試 one pytho lse 配置連接 read 當前 serial Celery分布式任務 celery call把任務給一個組件，組件交給rabiitmq放到隊列broker，隊列返回任務id給celery組件再給call，任務完成時call拿著id通過cel

分布式任務隊列celery用法詳解

後來 這一 als 介紹 type Coding 是把 cat 令行 celery基礎介紹：這個圖我們可以看出，celery基本結構也就是三部分1 第一部分 broker也就是中間件消息隊列，作用就是用來接收應用的請求這一部分常見玩法可以是rabbitmq和redis等2

celery --分布式任務隊列

lex object 高可用 大型項目 -s bin 使用 serial cal 一、介紹 celery是一個基於python開發的分布式異步消息任務隊列，用於處理大量消息，同時為操作提供維護此類系統所需的工具。 它是一個任務隊列，專註於實時處理，同時還支持任務調度

django+celery 實現分布式任務

交換機 觸發 本質 seconds 數據報 bit con 定時器1 mps 想用django做一個自動運維平臺，利用netsnmp來獲取交換機及服務器信息，但是snmpget任務需要在後臺實時運行，為了不影響html響應，利用celery來結合django做異步任務隊列。

Go語言開發分布式任務調度 輕松搞定高性能Crontab

監聽 接口 記錄 簡單 inux 寫入 可用 初始化 創建 第1章 課程介紹本章中將介紹一下本課程的基本內容，包括：我們要做什麽、要求什麽基礎、將學會哪些工具、收獲哪些獨家幹貨，以及課程具體安排。1-1 導學視頻第2章 如何執行shell命令執行"定時任務"其實就是執行"s

Celery異步的分布式任務調度理解

生產 業務邏輯 應用 text 發送 mongo 實現 delay 手機號 什麽是Celery呢？ Celery是一個用Python開發的異步的分布式任務調度模塊。 Celery本身不包含消息服務，使用第三方消息服務，也就是Broker，來傳遞任務，目前支持的有Rebbim

Celery 分布式任務隊列快速入門 以及在Django中動態添加定時任務

快速入門 change pwa waiting processes 中間件 peid eas 復雜 Celery 分布式任務隊列快速入門 以及在Django中動態添加定時任務 轉自 金角大王 http://www.cnblogs.com/alex3714/articles/

spring與quartz整合實現分布式動態創建，刪除，改變執行時間定時任務(mysql數據庫)

ces value 我們 job clu xsd collect 註解 common 背景：因為在項目中用到了定時任務，當時想到了spring的quartz，寫完發現費了很大功夫，光是整合就花了一上午，其中最大的問題就是版本問題，項目中用的是spring3.2.8的版本，查

基於spring+quartz的分布式定時任務框架

path simple ngs 分布 方式 fix set 分布式 .cn http://www.cnblogs.com/aaronfeng/p/5537177.html 問題背景 　　我公司是一個快速發展的創業公司，目前有200人，主要業務是旅遊和酒店相關的，應用叠

流程控制：分布式並行任務流程控制

time aid ges stat rdquo net 不同的 分布式部署 red 背景： 目前工作中遇到一個比較急，又有點費事的工作任務： 1）目前系統中已經已經包含了一些比較完善的部分模塊，但是模塊之間沒有一個控制流程來管理，就造成程序沒有辦法自動化； 2

Spring整合Quartz定時任務 在集群、分布式系統中的應用(Mysql數據庫環境)

foreign log ride bigint post 業務 設置 ade 營業額 Spring整合Quartz定時任務 在集群、分布式系統中的應用(Mysql數據庫環境) 轉載：http://www.cnblogs.com/jiafuwei/p/6145280.

17. ZooKeeper常見的分布式系統任務——屏障

per java 開始 例子 重要 行處理 避免 註冊 進程 以後幾節中主要介紹以下內容： 如何執行領導者選舉，組員管理和兩階段提交協議等常見的分布式系統任務 如何實現一些分布式數據結構，如屏障（barrier），鎖（lock）和隊列（queue） 這一章中概述的高層次