2. 程式人生 > >Elastic-Job:動態新增任務,支援動態分片

Elastic-Job:動態新增任務,支援動態分片

阿新 • • 發佈：2019-09-19

多情只有春庭月，猶為離人照落花。

概述

因專案中使用到定時任務，且服務部署多例項，因此需要解決定時任務重複執行的問題。即在同一時間點，每一個定時任務只在一個節點上執行。常見的開源方案，如 elastic-job 、 xxl-job 、quartz 、 saturn、 opencron 、 antares 等。最終決定使用elastic-job。elastic-job的亮點主要如下：

  1. 基於quartz 定時任務框架為基礎的，因此具備quartz的大部分功能
  2. 使用zookeeper做協調，排程中心，更加輕量級
  3. 支援任務的分片
  4. 支援彈性擴容 ， 可以水平擴充套件 ， 當任務再次執行時，會檢查當前的伺服器數量，重新分片，分片結束之後才會繼續執行任務
  5. 失效轉移，容錯處理，當一臺排程伺服器宕機或者跟zookeeper斷開連線之後，會立即停止作業，然後再去尋找其他空閒的排程伺服器，來執行剩餘的任務
  6. 提供運維介面，可以管理作業和註冊中心

但在實際開發中發現elastic-job對於動態新增的定時任務不支援分片。即在多例項情況下，在某個例項上動態新增任務，則該任務會一直在這一臺節點上執行。如果需要在其它例項上執行，則需要以相同的引數呼叫其它例項介面。參考:elastic-job：動態進行任務的新增。在多次百度+google下發現Elastic-Job動態新增任務這裡與樓主遇到了相同的問題。但經樓主測試動態新增任務的分片時好時壞,且只要在zookeeper

中註冊了任務,重啟時任務還是會自動初始化。（關於對動態呢任務的描述，可以參考上面連結的描述，此處不在做過多的解釋）。

解決

順著尹大的思路,將任務的節點都集中管理起來，無論動態任務在哪個節點上進行註冊，都需要將這個請求轉發到其他的節點上進行初始化操作，這樣就可以保證多節點分片的任務正常執行。

程式碼如下:

/**
     * 開啟任務監聽,當有任務新增時，監聽zk中的資料增加，自動在其他節點也初始化該任務
     */
    public void monitorJobRegister() {
        CuratorFramework client = zookeeperRegistryCenter.getClient();
        @SuppressWarnings("resource")
        PathChildrenCache childrenCache = new PathChildrenCache(client, "/", true);
        PathChildrenCacheListener childrenCacheListener = new PathChildrenCacheListener() {
            public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
                ChildData data = event.getData();
                switch (event.getType()) {
                    case CHILD_ADDED:
                        String config = new String(client.getData().forPath(data.getPath() + "/config"));
                        Job job = JsonUtils.toBean(Job.class, config);
                        Object bean = null;
                        // 獲取bean失敗則新增任務
                        try {
                            bean = ctx.getBean("SpringJobScheduler" + job.getJobName());
                        } catch (BeansException e) {
                            logger.error("ERROR NO BEAN,CREATE BEAN SpringJobScheduler" + job.getJobName());
                        }
                        if (Objects.isNull(bean)) {
                            addJob(job);
                        }
                        break;
                    default:
                        break;
                }
            }
        };
        childrenCache.getListenable().addListener(childrenCacheListener);
        try {
            // https://blog.csdn.net/u010402202/article/details/79581575
            childrenCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

測試

測試動態新增定時任務,支援分片失效轉移。

  1. 下載elastic-job-spring-boot-starter 使用maven 命令install到本地
  2. 建立 demo-elastic-job專案
    目錄結構如下：
demo-elastic-job
├── mvnw
├── mvnw.cmd
├── src
│   ├── main
│   │   ├── java
│   │   │   └── com
│   │   │       └── example
│   │   │           └── demo
│   │   │               ├── job
│   │   │               │   ├── DynamicJob.java
│   │   │               │   └── TestJob.java
│   │   │               └── DemoApplication.java
│   │   └── resources
│   │       ├── application.yml
│   │       └── application-dev.yml
│   └── test
│       └── java
│           └── com
│               └── example
│                   └── demo
│                       └── DemoApplicationTests.java
├── pom.xml
└── demo.iml

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<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>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.1.1.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.example</groupId>
    <artifactId>demo</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>demo</name>
    <description>Demo project for Spring Boot</description>

    <properties>
        <java.version>1.8</java.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-recipes</artifactId>
            <version>2.10.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-framework</artifactId>
            <version>2.10.0</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>com.cxytiandi</groupId>
            <artifactId>elastic-job-spring-boot-starter</artifactId>
            <version>1.0.4</version>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

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

</project>

DemoApplication.java

package com.example.demo;

import com.cxytiandi.elasticjob.annotation.EnableElasticJob;
import com.cxytiandi.elasticjob.dynamic.service.JobService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.ComponentScan;

@SpringBootApplication
@EnableElasticJob
@ComponentScan(basePackages = {"com.cxytiandi", "com.example.demo"})
public class DemoApplication implements CommandLineRunner {

    public static void main(String[] args) {
        SpringApplication.run(DemoApplication.class, args);
    }

    @Autowired
    private JobService jobService;

    @Override
    public void run(String... args) throws Exception {
        // 模擬初始化讀取資料庫 新增任務
//        Job job1 = new Job();
//        job1.setJobName("job1");
//        job1.setCron("0/10 * * * * ? ");
//        job1.setJobType("SIMPLE");
//        job1.setJobClass("com.example.demo.job.DynamicJob");
//        job1.setShardingItemParameters("");
//        job1.setShardingTotalCount(2);
//        jobService.addJob(job1);
//        Job job2 = new Job();
//        job2.setJobName("job2");
//        job2.setCron("0/10 * * * * ? ");
//        job2.setJobType("SIMPLE");
//        job2.setJobClass("com.example.demo.job.DynamicJob");
//        job2.setShardingItemParameters("0=A,1=B");
//        job2.setShardingTotalCount(2);
//        jobService.addJob(job2);
    }
}

TestJob.java

package com.example.demo.job;

import com.cxytiandi.elasticjob.annotation.ElasticJobConf;
import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.api.simple.SimpleJob;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Component;

/**
 * Created by zhenglongfei on 2019/7/22
 *
 * @VERSION 1.0
 */
@Component
@Slf4j
@ElasticJobConf(name = "dayJob", cron = "0/10 * * * * ?", shardingTotalCount = 2,
        shardingItemParameters = "0=AAAA,1=BBBB", description = "簡單任務", failover = true)
public class TestJob implements SimpleJob {
    @Override
    public void execute(ShardingContext shardingContext) {
        log.info("TestJob任務名：【{}】, 片數：【{}】, param=【{}】", shardingContext.getJobName(), shardingContext.getShardingTotalCount(),
                shardingContext.getShardingParameter());
    }
}

DynamicJob.java

package com.example.demo.job;

import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.api.simple.SimpleJob;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

/**
 * Created by zhenglongfei on 2019/7/24
 *
 * @VERSION 1.0
 */
@Component
@Slf4j
public class DynamicJob implements SimpleJob {
    @Override
    public void execute(ShardingContext shardingContext) {


        switch (shardingContext.getShardingItem()) {
            case 0:
                log.info("【0】 is running");
                break;
            case 1:
                log.info("【1】 is running");
                break;
        }
    }
}

application.yml

elastic:
  job:
    zk:
      serverLists: 172.25.66.137:2181
      namespace: demo_test
server:
  port: 8082
spring:
  redis:
    host: 127.0.0.1
    port: 6379

測試結果

啟動兩個專案分別為8081和8082埠，使用REST API來動態的註冊任務。

  • job

http://localhost:8081/job post
引數如下：

{
  "jobName": "DynamicJob01",
  "cron": "0/3 * * * * ?",
  "jobType": "SIMPLE",
  "jobClass": "com.example.demo.job.DynamicJob",
  "jobParameter": "test",
  "shardingTotalCount": 2,
  "shardingItemParameters": "0=AAAA,1=BBBB"
}

程式碼下載

  • github:demo-elastic-job
  • github:elastic-job-spring-boot-starter

相關推薦

Elastic-Job:動態新增任務,支援動態分片

多情只有春庭月，猶為離人照落花。 概述 因專案中使用到定時任務，且服務部署多例項，因此需要解決定時任務重複執行的問題。即在同一時間點，每一個定時任務只在一個節點上執行。常見的開源方案，如 elastic-job 、 xxl-job 、quartz 、 saturn、 opencron 、 antare

python協程之動態新增任務

https://blog.csdn.net/qq_29349715/article/details/79730786   python協程只能執行在事件迴圈中，但是一旦事件迴圈執行，又會阻塞當前任務。所以只能在當前程序中再開一個執行緒，這個執行緒的主要任務是執行事件迴圈，就是event_loop

Spring實現動態資料來源，支援動態新增、刪除和設定權重及讀寫分離

當專案慢慢變大，訪問量也慢慢變大的時候，就難免的要使用多個數據源和設定讀寫分離了。 在開題之前先說明下，因為專案多是使用Spring，因此以下說到某些操作可能會依賴於Spring。 在我經歷過的專案中，見過比較多的讀寫分離處理方式，主要分為兩步： 1、對於開發人員，要

IT忍者神龜之Spring Task動態新增任務

<task:scheduled-tasks>標籤的作用  <!-- <task:scheduled-tasks> 自動註冊的 -->   <value>org.springframework.scheduling.confi

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

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

java實現動態許可權(選單管理)動態新增選單，動態新增角色SSM框架

首先是資料庫設計 Menu表（選單表） Role表（角色表） Role_Menu表（角色選單關係表） User表（使用者表） User_Role表（使用者角色關係表） 其他實體類就略過了 在Menu實體類中添加了一個List<Menu&

Elastic-Job——分散式定時任務框架

什麼是作業? 作業即定時任務.無需做複雜的控制,在指定的時間執行指定的任務. 為什麼需要作業?         時間驅動/事件驅動: 內部系統一般可以通過事件來驅動,但涉及到外部系統,則只能使用時間驅動.如：抓取外部系統價格.每小時抓取，由於是外部系統,不能像內部系統一樣傳送事件觸發事件. 批量處理/

select動態新增option與動態設定下拉框預設選項（selected）的問題(原創)

//動態生成下拉框$.ajax({    type:"GET",    async:false,     //設為同步請求（非同步載入的話後面的遍歷方法獲取不到option）    url:context+"/service/demo/hfjd/findJdmc",//獲取

Elastic-Job是否支援動態新增做作業

原文地址：http://dangdangdotcom.github.io/elastic-job/post/faq/ 6. 是否支援動態新增作業? 回答： 動態新增作業這個概念每個人理解不盡相同。 elast

SpringBoot整合Elastic-Job，實現動態建立定時任務，任務持久化

SpringBoot使用Elastic-Job-lite，實現動態建立定時任務，任務持久化 Elastic-Job是噹噹開源的一個分散式排程解決方案，由兩個相互獨立的子專案Elastic-Job-Lite和Elastic-Job-Cloud組成。 Elast

Elastic-Job原始碼解析(三)之分片定時任務執行

通過本篇的閱讀你將學會了解Elastic-Job的定時時機,及如何通過分片方式做一個分散式的定時任務框架。瞭解常用的三種分片策略，及如何自定義分散式分片策略 目錄 Elastic-Job如何通過SpringJobScheduler啟動定時 Ela

Spring4.0.6 +Quartz 2.2.1動態新增、修改、刪除、關閉定時任務（可結合資料庫以及空指標異常問題的解決）

不懂定時器的可以閱讀這位作者的文章：Quartz 2.2 的實現原理和執行過程：http://blog.csdn.net/xlxxcc/article/details/52104463 spring配置 <!-- 定時器 --> <!-- 這個類用來做需要完成的業務--&

JSP頁面傳List集合到Action中（並且支援動態新增行，刪除行）

1：JSP頁面（前端用的是H-UI框架） 1：HTML程式碼 <div class="cl pd-5 bg-1 bk-gray mt-20"> <span class="l"> <a href="javascript:;" onclick="add(

springboot2.0+Elastic Job動態job配置實現微服務呼叫

springboot2.0+Elastic Job實現微服務呼叫 本猿第一次寫部落格，寫的不好勿噴 一、前提 本文只介紹如何把elastic-job和springcloud微服務整合，不對其他知識做介紹,需先掌握以下知識 1.會使用springboot 2.會使用springcl

centos7中使用bash指令碼動態新增crontab定時任務

1.首先將在已建立的定時任務放在某個txt檔案中 例如：/file/test/tab.txt 20 * * * * /tmp/shdir/test.sh  >> /file/test/t

Elastic-Job原理--任務分片策略（三）

上一篇部落格Elastic-Job原理--伺服器初始化、節點選舉與通知(二)介紹了Elastic-Job的啟動流程，這篇部落格我們瞭解學習一下Elastic-Job的任務分片策略，目前提供了三種任務分片策略，分片策略的實現最終是在註冊中心zk中在分片的instance中寫入例

springmvc 動態 新增/修改 定時任務

springmvc  +  quartz 動態定時任務1、jar 包<dependency> <groupId>org.quartz-scheduler</groupId> <artifactId>quartz&

Spring整合Quartz動態定時任務，新增及刪除的一些操作

我是是一個新手，最近公司要用定時任務，才研究了一下。有什麼錯誤的地方，希望多指正。 1、首先需要一個執行的QuartzJob（任務類）實現Job介面 然後重寫execute（），需要執行的具體方法就寫在這裡面。   2、編寫一個QuartzJob（任務類）的實體類Jo

Spring3整合Quartz2實現定時任務及動態任務調整(新增刪除暫停恢復)--推薦

1、常規整合 最近工作中需要用到定時任務的功能，雖然Spring3也自帶了一個輕量級的定時任務實現，但感覺不夠靈活，功能也不夠強大。在考慮之後，決定整合更為專業的Quartz來實現定時任務功能。 首先，當然是新增依賴的jar檔案，我的專案是maven管理的，以下的我專

Spring+Quartz實現動態新增定時任務（一）

在最近工作中，由於涉及到定時任務特別多，而這些工作又是由下屬去完成的，在生成環境中經常會出現業務邏輯錯誤，分析下來多數是定時任務執行問題，所以就希望把定時任務優化一下，主要實現2個方面1.定時任務動態配置及持久化2.視覺化的管理介面，可以非常清晰的管理自己的所有定時任務首先，