springBoot集成 Elastic-job

Elastic-job 是由当当网开源的轻量级无中心化定时任务解决方案，使用jar包的形式提供最轻量级的分布式任务的协调服务，外部依赖仅Zookeeper。

其中基本的概念是分片，任务的分布式执行，需要将一个任务拆分为多个独立的任务项，然后由分布式的服务器分别执行某一个或几个分片项。例如，一个任务需要处理100w的数据，现在将任务设置为两个分片，两台服务，一台服务获取一个分片，并处理不通的数据。Elastic-Job并不直接提供数据处理的功能，框架只会将分片项分配至各个运行中的作业服务器，开发者需要自行处理分片项与真实数据的对应关系。

Elastic-job 没有作业调度中心节点，而是基于部署作业框架的程序在到达相应时间点时各自触发调度，注册中心的作用只是用户作业的注册和监控信息的存储，主节点的作用是主要用来处理分片。

基于分片的设计思想，一旦执行作业的服务器崩溃，等待执行的服务器将会在下次作业启动时重新触发分片操作，并进行执行操作。

架构图如下：

接入步骤：1）<!-- 引入elastic-job-lite核心模块 --><dependency><groupId>com.dangdang</groupId><artifactId>elastic-job-lite-core</artifactId><version>${latest.release.version}</version></dependency><!-- 使用springframework自定义命名空间时引入 --><dependency><groupId>com.dangdang</groupId><artifactId>elastic-job-lite-spring</artifactId><version>${latest.release.version}</version></dependency>2）作业开发public class MyElasticJob implements SimpleJob {@Overridepublic void execute(ShardingContext context) {switch (context.getShardingItem()) {case 0: // do something by sharding item 0 break;case 1: // do something by sharding item 1 break;case 2: // do something by sharding item 2 break; //case n: ...}}}

Elastic-Job提供Simple、Dataflow和Script 3种作业类型。方法参数shardingContext包含作业配置、片和运行时信息。可通过getShardingTotalCount(), getShardingItem()等方法分别获取分片总数，运行在本作业服务器的分片序列号等。

a: Simple类型作业

意为简单实现，未经任何封装的类型。需实现SimpleJob接口。该接口仅提供单一方法用于覆盖，此方法将定时执行。与Quartz原生接口相似，但提供了弹性扩缩容和分片等功能

b: Dataflow类型作业

Dataflow类型用于处理数据流，需实现DataflowJob接口。该接口提供2个方法可供覆盖，分别用于抓取(fetchData)和处理(processData)数据。

可通过DataflowJobConfiguration配置是否流式处理。

流式处理数据只有fetchData方法的返回值为null或集合长度为空时，作业才停止抓取，否则作业将一直运行下去；非流式处理数据则只会在每次作业执行过程中执行一次fetchData方法和processData方法，随即完成本次作业。

如果采用流式作业处理方式，建议processData处理数据后更新其状态，避免fetchData再次抓取到，从而使得作业永不停止。流式数据处理参照TbSchedule设计，适用于不间歇的数据处理。

c: Script类型作业

Script类型作业意为脚本类型作业，支持shell，python，perl等所有类型脚本。只需通过控制台或代码配置scriptCommandLine即可，无需编码。执行脚本路径可包含参数，参数传递完毕后，作业框架会自动追加最后一个参数为作业运行时信息。

3）spring命名空间配置<?xml version="1.0" encoding="UTF-8"?><beansxmlns="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 "><!--配置作业注册中心 --><reg:zookeeper id="regCenter" server-lists=" yourhost:2181" namespace="dd-job" base-sleep-time-milliseconds="1000" max-sleep-time-milliseconds="3000" max-retries="3" /><!-- 配置作业--><job:simple id="oneOffElasticJob" class="xxx.MyElasticJob" registry-center-ref="regCenter" cron="0/10 * * * * ?" sharding-total-count="3" sharding-item-parameters="0=A,1=B,2=C" /></beans>

Springboot 使用elasticjob 方法：

在pom中引入

<!-- https://mvnrepository.com/artifact/com.dangdang/elastic-job-lite-core --><dependency><groupId>com.dangdang</groupId><artifactId>elastic-job-lite-core</artifactId><version>2.1.5</version></dependency><!-- 使用springframework自定义命名空间时引入 --><dependency><groupId>com.dangdang</groupId><artifactId>elastic-job-spring</artifactId><version>1.1.1</version></dependency>

配置注册中心

@Data@Configurationpublic class ZkConfig {private String zkHost = "10.143.117.32:2181,10.143.117.32:2182,10.143.117.32:2183";private String namespace = "gome-job";@Beanpublic ZookeeperConfiguration zkConfiguration() {ZookeeperConfiguration zk = new ZookeeperConfiguration(zkHost, namespace);zk.setBaseSleepTimeMilliseconds(1000);zk.setMaxSleepTimeMilliseconds(3000);zk.setMaxRetries(3);return zk;}@Bean(initMethod = "init")public  ZookeeperRegistryCenter regCenter(ZookeeperConfiguration config) {return new ZookeeperRegistryCenter(config);}}

编写simpleJob任务

public class SimpleJob implements com.dangdang.ddframe.job.api.simple.SimpleJob {@Overridepublic void execute(ShardingContext context) {System.out.println("任务执行");System.out.println("shardItems : " + context.getShardingItem() );String sql = "select * from t_user where right(id,1) = " + context.getShardingItem();System.out.println(sql);}}

springboot启动后初始化任务

@Component@RequiredArgsConstructorpublic class JobInitRunner implements CommandLineRunner {private final CoordinatorRegistryCenter regCenter;@Overridepublic void run(String... strings) throws Exception {System.out.println("任务初始化。。。");jobInit();}public void jobInit() {addJob(createLiteJobConfiguration("SimpleJob","0/5 * * * * ?",10,SimpleJob.class, JobType.SIMPLE));addJob(createLiteJobConfiguration("DataJob", "0/5 * * * * ?", 10,DateJob.class,JobType.DATAFLOW));}private void addJob(LiteJobConfiguration jobConfiguration) {new JobScheduler(regCenter, jobConfiguration).init();}private LiteJobConfiguration createLiteJobConfiguration(String JobName, String cron, int shardingCount, Class c, JobType type){JobCoreConfiguration coreConfig = JobCoreConfiguration.newBuilder(JobName, cron, shardingCount).build();JobTypeConfiguration jobConfig = new SimpleJobConfiguration(coreConfig, c.getCanonicalName());if(type.compareTo(JobType.DATAFLOW) == 0){jobConfig = new DataflowJobConfiguration(coreConfig,c.getCanonicalName(),true);}return LiteJobConfiguration.newBuilder(jobConfig).build();}}

springboot启动

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

springBoot集成 Elastic-job相关推荐

springboot集成elastic job
文章目录集成 pom.xml添加依赖 application.properties 添加配置任务其他 @ElasticJobConf 是最核心的注解基本结构 cron如何设置为可配置集成 p ...
SpringBoot集成Elasticsearch7.4 实战（一）
在网上已经有好多关于Elasticsearch的介绍,就不在翻来覆去讲一些基本概念,大家感兴趣的可以自己去找一些资料巩固下.这次只为了顾及众多首次接触Elasticsearch,案例都讲的很浅显,还有 ...
springboot集成elasticsearch6.8.23设置密码xpack连接，及遇到的None of the configured nodes are available
遇到一个 None of the configured nodes are available 的坑一.背景: 因现网扫描出来几个漏洞,目前版本使用的springboot2.1.17.RELEASE ...
SpringBoot集成Elasticseach
目录一.Elasticseach介绍 1.简单介绍 2.对比关系: 3.详细说明: 4.查出数据的解释二.SpringBoot集成Elasticseach 1.引入依赖 2.添加配置 3.创建po ...
springboot集成es7.2自定义注解创建索引
各位好,我们把之前的坑填一下,我在上上篇文章中写了springboot集成es7 的方法,并且集成了es原生客户端 High Level Rest Client, 也说明了原因, 我用的版本较高, ...
从ElasticSearch 认识到实战（SpringBoot集成ES）
ElasticSearch 认识到实战目录搜索引擎介绍 ElasticSearch知识安装使用restful风格查询ES SpringBoot配置ES SpringBoot集成使用一.搜索引 ...
Elasticsearch-Sql+Mybatis+SpringBoot集成
Elasticsearch-Sql+Mybatis+SpringBoot集成本文介绍了springboot如何集成elasticsearch-sql+mybatis rest方式使用es-sql J ...
elasticsearch集群搭建及springboot集成使用
elasticsearch集群搭建及springboot集成使用 1. ES介绍 2. 原理 2.1 核心概念 2.2 索引功能 2.3 ES特性 3. 分词器(analyzer) 4. 集群搭建 4 ...
springboot集成Elasticsearch实现各种搜索功能
springboot集成Elasticsearch各类搜索功能实现 springboot集成Elasticsearch使用completion suggest实现自动关键字补全建立学生的索引和映射: ...
ES入门学习：ElasticSearch、Kibana、ik分词器的安装、简单使用及SpringBoot集成
前言 es是采用Java语言开发,因此,想要安装运行es需要提前准备好jdk环境,关于linux配置jdk在前文linux配置jdk 本文主要介绍es的安装.kibana的安装和简单使用及ik分词器的 ...

springBoot集成 Elastic-job

springBoot集成 Elastic-job相关推荐

最新文章

热门文章