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分词器的 ...
最新文章
- 面向对象设计原则_聊聊面向对象的6大设计原则
- MIT 更新最大自然灾害图像数据集,囊括 19 种灾害事件
- C++中的构造函数VS析构函数
- python中loop函数运用_使用涉及函数的Python在for循环中填充DataFrame
- Elastic Search入门:架构说明及Docker方式体验
- 第十届蓝桥杯java B组—试题F 特别数的和
- java向量数组异常,数组中空异常指针的Java错误处理
- linux重启鼠标键盘服务,Linux 关闭服务后 鼠标 键盘用不了
- 服务器系统架构的评估,系统架构师:性能评估
- Java 面向对象:多态的理解
- Bouncy Castle Crypto API c# port
- 《众妙之门——自由网站设计师成功之道》一1.6 创造性地处理行政事务性工作...
- GAN详解与PyTorch MINIST手写数字生成实战
- 德国IT民工找工有感
- 黑客全军覆没 书生安全云实战各路高手
- vue实现数字翻页动画
- MVC母版页的使用方法
- PHP导出Excel,亲测有用
- SkinSharp(Skin#)软件换肤库!
- 获取Windows操作系统版本和位数