工具篇--分布式定时任务springBoot 整合 elasticjob使用（3）

2024-03-13 03:12
文章标签 java 工具 使用 springboot spring 整合 分布式 定时 任务 elasticjob

本文主要是介绍工具篇--分布式定时任务springBoot 整合 elasticjob使用（3），希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

文章目录

  • 前言
  • 一、Springboot 整合：
    • 1.1 引入jar：
    • 1.2 配置zookeeper 注册中心：
    • 1.3 定义job 业务类：
    • 1.4 job 注册到zookeeper：
    • 1.5 项目启动：
      • 1.5.1 zookeeper 注册中心实例：
      • 1.5.2 任务执行日志输出：
  • 二、扩展：
    • 2.1 任务监听器：
    • 2. 2 DataflowJob 流工作：
    • 2.2.1 新建 DataflowJob：
    • 2.2.2 streaming.process 属性配置：
    • 2.2.3 执行效果：
  • 总结

前言

本文对springBoot 整合 elasticjob 进行介绍。 本文环境 jdk:1.8 ; zookeeper : 3.7

一、Springboot 整合：

1.1 引入jar：

<!-- https://mvnrepository.com/artifact/org.apache.shardingsphere.elasticjob/elasticjob-lite-core --><dependency><groupId>org.apache.shardingsphere.elasticjob</groupId><artifactId>elasticjob-lite-core</artifactId><!-- <version>3.0.4</version>--><version>3.0.1</version></dependency><dependency><groupId>org.yaml</groupId><artifactId>snakeyaml</artifactId><version>1.27</version></dependency><dependency><groupId>ch.qos.logback</groupId><artifactId>logback-classic</artifactId><version>1.2.12</version></dependency>

1.2 配置zookeeper 注册中心：

ElasticJobZookeeper：

import org.apache.shardingsphere.elasticjob.reg.base.CoordinatorRegistryCenter;
import org.apache.shardingsphere.elasticjob.reg.zookeeper.ZookeeperConfiguration;
import org.apache.shardingsphere.elasticjob.reg.zookeeper.ZookeeperRegistryCenter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class ElasticJobZookeeper {@Bean(initMethod = "init")public  CoordinatorRegistryCenter createRegistryCenter() {ZookeeperConfiguration zookeeperConfiguration = new ZookeeperConfiguration("localhost:2181", "elastic-job");zookeeperConfiguration.setConnectionTimeoutMilliseconds(10000);zookeeperConfiguration.setSessionTimeoutMilliseconds(10000);zookeeperConfiguration.setMaxSleepTimeMilliseconds(10000);CoordinatorRegistryCenter regCenter = new ZookeeperRegistryCenter(zookeeperConfiguration);return regCenter;}
}

1.3 定义job 业务类：

MyJob：

import lombok.extern.slf4j.Slf4j;
import org.apache.shardingsphere.elasticjob.api.ShardingContext;
import org.apache.shardingsphere.elasticjob.simple.job.SimpleJob;
import org.springframework.stereotype.Component;@Slf4j
@Component
public class MyJob implements SimpleJob {@Overridepublic void execute(ShardingContext shardingContext) {// 分片参数 0=text,1=image,2=radio,3=vedioString  shardingParameter= shardingContext.getShardingParameter();String  jobParameter= shardingContext.getJobParameter();log.debug("job 执行 error,job名称：{},分片数量:{},分片：{},分片参数:{},jobParamer:{}", shardingContext.getJobName(), shardingContext.getShardingTotalCount(),shardingContext.getShardingItem(), shardingParameter,jobParameter);if ("text".equals(jobParameter)) {// do something by sharding}switch (shardingContext.getShardingItem()) {case 0:// do something by sharding item 0break;case 1:// do something by sharding item 1break;case 2:// do something by sharding item 2break;// case n: ...}}
}

MyJob1：

import lombok.extern.slf4j.Slf4j;
import org.apache.shardingsphere.elasticjob.api.ShardingContext;
import org.apache.shardingsphere.elasticjob.simple.job.SimpleJob;
import org.springframework.stereotype.Component;@Slf4j
@Component
public class MyJob1 implements SimpleJob {@Overridepublic void execute(ShardingContext shardingContext) {log.debug("job 执行 error,job名称：{},分片数量:{}",shardingContext.getJobName(),shardingContext.getShardingTotalCount());switch (shardingContext.getShardingItem()) {case 0:// do something by sharding item 0break;case 1:// do something by sharding item 1break;case 2:// do something by sharding item 2break;// case n: ...}}
}

1.4 job 注册到zookeeper：

ElasticJobConfigure

import com.example.springelasticjob.quickstart.MyJob;
import com.example.springelasticjob.quickstart.MyJob1;
import org.apache.shardingsphere.elasticjob.api.JobConfiguration;
import org.apache.shardingsphere.elasticjob.lite.api.bootstrap.impl.ScheduleJobBootstrap;
import org.apache.shardingsphere.elasticjob.reg.base.CoordinatorRegistryCenter;
import org.apache.shardingsphere.elasticjob.simple.job.SimpleJob;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;@Configuration
public class ElasticJobConfigure implements InitializingBean {@Autowiredprivate MyJob myJob;@Autowiredprivate MyJob1 myJob1;@Autowiredprivate CoordinatorRegistryCenter coordinatorRegistryCenter;public JobConfiguration createJobConfiguration(Class<? extends SimpleJob> JobClass, int shardingTotalCount, String cron, String shardingItemParameters) {// 创建作业配置JobConfiguration jobConfiguration = JobConfiguration.newBuilder(JobClass.getName(), shardingTotalCount).cron(cron).overwrite(true).shardingItemParameters(shardingItemParameters).jobListenerTypes().build();return jobConfiguration;}@Overridepublic void afterPropertiesSet() throws Exception {JobConfiguration jobConfiguration = createJobConfiguration(myJob.getClass(), 1, "0/10 * * * * ?", null);new ScheduleJobBootstrap(coordinatorRegistryCenter, myJob, jobConfiguration).schedule();JobConfiguration jobConfiguration1 = createJobConfiguration(myJob1.getClass(), 1, "0/1 * * * * ?", null);new ScheduleJobBootstrap(coordinatorRegistryCenter, myJob1, jobConfiguration1).schedule();}
}

1.5 项目启动：

1.5.1 zookeeper 注册中心实例：

在这里插入图片描述

1.5.2 任务执行日志输出：

在这里插入图片描述

二、扩展：

2.1 任务监听器：

  1. 定义监听器：
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.time.DateFormatUtils;
import org.apache.shardingsphere.elasticjob.infra.listener.ElasticJobListener;
import org.apache.shardingsphere.elasticjob.infra.listener.ShardingContexts;import java.util.Date;@Slf4j
public class MyElasticJobListener implements ElasticJobListener {private long beginTime = 0;@Overridepublic void beforeJobExecuted(ShardingContexts shardingContexts) {beginTime = System.currentTimeMillis();log.info("===>{} MyElasticJobListener BEGIN TIME: {} <===",shardingContexts.getJobName(),  DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss"));}@Overridepublic void afterJobExecuted(ShardingContexts shardingContexts) {long endTime = System.currentTimeMillis();log.info("===>{} MyElasticJobListener END TIME: {},TOTAL CAST: {} <===",shardingContexts.getJobName(), DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss"), endTime - beginTime);}@Overridepublic String getType() {return "myElasticJobListener";}}

2） 在项目resources 新建文件夹： META-INF\services
在这里插入图片描述
3）新建文件，名称为：org.apache.shardingsphere.elasticjob.infra.listener.ElasticJobListener
文集内容：

# 监听器实现类的 类全路径
com.example.springelasticjob.config.MyElasticJobListener

4）job 配置增加监听器：

// 创建作业配置JobConfiguration jobConfiguration = JobConfiguration.newBuilder("myjob-param", 1).cron("0/5 * * * * ?").overwrite(true).shardingItemParameters("0=Beijing,1=Shanghai,2=Guangzhou").jobParameter("0=a,1=b,2=c").jobListenerTypes("myElasticJobListener").build();

jobListenerTypes(“myElasticJobListener”) 中 “myElasticJobListener” 要和 MyElasticJobListener getType() 返回的保持一致，否则启动无法找到 监听器：
在这里插入图片描述

2. 2 DataflowJob 流工作：

2.2.1 新建 DataflowJob：


import lombok.extern.slf4j.Slf4j;
import org.apache.shardingsphere.elasticjob.api.ShardingContext;
import org.apache.shardingsphere.elasticjob.dataflow.job.DataflowJob;import java.util.ArrayList;
import java.util.List;/*** 流任务*/
@Slf4j
public class MyDataFlowJob implements DataflowJob {@Overridepublic List fetchData(ShardingContext shardingContext) {// 抓取数据// 分片参数 0=text,1=image,2=radio,3=vedioString jobParameter = shardingContext.getJobParameter();log.debug("job 执行 error,job名称：{},分片数量:{},分片：{},分片参数:{}", shardingContext.getJobName(), shardingContext.getShardingTotalCount(), shardingContext.getShardingItem(), jobParameter);List list = new ArrayList(1);list.add("lgx");return list;}@Overridepublic void processData(ShardingContext shardingContext, List list) {// 数据处理System.out.println("list.toString() = " + list.toString());}
}

2.2.2 streaming.process 属性配置：

 private static JobConfiguration createJobConfiguration() {JobConfiguration jobConfiguration = JobConfiguration.newBuilder("myjob-dataflow-param", 1).cron("0/30 * * * * ?").overwrite(true).shardingItemParameters("0=Beijing,1=Shanghai,2=Guangzhou").jobParameter("0=a,1=b,2=c")//  streaming.process 流处理设置为true.setProperty("streaming.process","true").build();return jobConfiguration;}

2.2.3 执行效果：

虽然任务是每隔30s 执行一次，但是因为 fetchData 可以一直获取到数据，使的 processData 方法可以一直被调用：
在这里插入图片描述

总结

本文对 springBoot 整合 elasticjob 整合，监听器的使用，任务的流式处理做介绍。

这篇关于工具篇--分布式定时任务springBoot 整合 elasticjob使用（3）的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/803476。

相关文章

Java实现字节字符转bcd编码

Java实现字节字符转bcd编码

《Java实现字节字符转bcd编码》BCD是一种将十进制数字编码为二进制的表示方式,常用于数字显示和存储,本文将介绍如何在Java中实现字节字符转BCD码的过程,需要的小伙伴可以了解下... 目录前言BCD码是什么Java实现字节转bcd编码方法补充总结前言BCD码（Binary-Coded Decima
阅读更多...
Linux下MySQL数据库定时备份脚本与Crontab配置教学

Linux下MySQL数据库定时备份脚本与Crontab配置教学

《Linux下MySQL数据库定时备份脚本与Crontab配置教学》在生产环境中,数据库是核心资产之一,定期备份数据库可以有效防止意外数据丢失,本文将分享一份MySQL定时备份脚本,并讲解如何通过cr... 目录备份脚本详解脚本功能说明授权与可执行权限使用 Crontab 定时执行编辑 Crontab添加定
阅读更多...
SpringBoot全局域名替换的实现

SpringBoot全局域名替换的实现

《SpringBoot全局域名替换的实现》本文主要介绍了SpringBoot全局域名替换的实现,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一... 目录 项目结构⚙️ 配置文件application.yml️ 配置类AppProperties.Ja
阅读更多...
Java使用Javassist动态生成HelloWorld类

Java使用Javassist动态生成HelloWorld类

《Java使用Javassist动态生成HelloWorld类》Javassist是一个非常强大的字节码操作和定义库,它允许开发者在运行时创建新的类或者修改现有的类,本文将简单介绍如何使用Javass... 目录1. Javassist简介2. 环境准备3. 动态生成HelloWorld类3.1 创建CtC
阅读更多...
JavaScript中的高级调试方法全攻略指南

JavaScript中的高级调试方法全攻略指南

《JavaScript中的高级调试方法全攻略指南》什么是高级JavaScript调试技巧,它比console.log有何优势,如何使用断点调试定位问题,通过本文,我们将深入解答这些问题,带您从理论到实... 目录观点与案例结合观点1观点2观点3观点4观点5高级调试技巧详解实战案例断点调试：定位变量错误性能分
阅读更多...
使用Python批量将.ncm格式的音频文件转换为.mp3格式的实战详解

使用Python批量将.ncm格式的音频文件转换为.mp3格式的实战详解

《使用Python批量将.ncm格式的音频文件转换为.mp3格式的实战详解》本文详细介绍了如何使用Python通过ncmdump工具批量将.ncm音频转换为.mp3的步骤,包括安装、配置ffmpeg环... 目录1. 前言2. 安装 ncmdump3. 实现 .ncm 转 .mp34. 执行过程5. 执行结
阅读更多...
Java实现将HTML文件与字符串转换为图片

Java实现将HTML文件与字符串转换为图片

《Java实现将HTML文件与字符串转换为图片》在Java开发中,我们经常会遇到将HTML内容转换为图片的需求,本文小编就来和大家详细讲讲如何使用FreeSpire.DocforJava库来实现这一功... 目录前言核心实现：html 转图片完整代码场景 1：转换本地 HTML 文件为图片场景 2：转换 H
阅读更多...
Java使用jar命令配置服务器端口的完整指南

Java使用jar命令配置服务器端口的完整指南

《Java使用jar命令配置服务器端口的完整指南》本文将详细介绍如何使用java-jar命令启动应用,并重点讲解如何配置服务器端口,同时提供一个实用的Web工具来简化这一过程,希望对大家有所帮助... 目录1. Java Jar文件简介1.1 什么是Jar文件1.2 创建可执行Jar文件2. 使用java
阅读更多...
C#使用Spire.Doc for .NET实现HTML转Word的高效方案

C#使用Spire.Doc for .NET实现HTML转Word的高效方案

《C#使用Spire.Docfor.NET实现HTML转Word的高效方案》在Web开发中,HTML内容的生成与处理是高频需求,然而,当用户需要将HTML页面或动态生成的HTML字符串转换为Wor... 目录引言一、html转Word的典型场景与挑战二、用 Spire.Doc 实现 HTML 转 Word1
阅读更多...
SpringBoot实现不同接口指定上传文件大小的具体步骤

SpringBoot实现不同接口指定上传文件大小的具体步骤

《SpringBoot实现不同接口指定上传文件大小的具体步骤》：本文主要介绍在SpringBoot中通过自定义注解、AOP拦截和配置文件实现不同接口上传文件大小限制的方法,强调需设置全局阈值远大于... 目录一  springboot实现不同接口指定文件大小1.1 思路说明1.2 工程启动说明二 具体实施2
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2