工具篇--分布式定时任务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 整合，监听器的使用，任务的流式处理做介绍。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/740470.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

Ubuntu 14.04：PaddleOCR基于PaddleServing的在线服务化部署（失败）

Ubuntu 14.04：PaddleOCR基于PaddleServing的在线服务化部署（失败）

一、 二、安装 注&#xff1a; 安装 PaddleOCR 2.3 。 因为 PaddleOCR 2.4 的 推荐环境 PaddlePaddle > 2.1.2。 https://github.com/PaddlePaddle/PaddleOCR/blob/release/2.4/doc/doc_ch/environment.md 安装前的环境准备 在使用Paddle Serving之前&#xff0c;需要完…
阅读更多...
数据结构和算法：复杂度分析

数据结构和算法：复杂度分析

算法效率评估 时间效率&#xff1a;算法运行速度的快慢。 空间效率&#xff1a;算法占用内存空间的大小。 效率评估方法主要分为两种&#xff1a;实际测试、理论估算 实际测试问题&#xff1a; 1.难以排除测试环境的干扰因素。 硬件配置会影响算法的性能。需要在各种机器上进…
阅读更多...
Web 常用的 扩展开发框架

Web 常用的 扩展开发框架

当谈到提升浏览器功能和用户体验时&#xff0c;浏览器扩展成了一股强大的力量&#xff0c;备受用户青睐。在众多的Web扩展开发框架中&#xff0c;WXT和Plasmo凭借其丰富的工具和特性&#xff0c;以及简化的开发流程&#xff0c;成为开发者们的首选。在本文中&#xff0c;我们将…
阅读更多...
数据分析-Pandas如何画自相关图

数据分析-Pandas如何画自相关图

数据分析-Pandas如何画自相关图 数据分析和处理中&#xff0c;难免会遇到各种数据&#xff0c;那么数据呈现怎样的规律呢&#xff1f;不管金融数据&#xff0c;风控数据&#xff0c;营销数据等等&#xff0c;莫不如此。如何通过图示展示数据的规律&#xff1f; 数据表&#x…
阅读更多...
C#，文字排版的折行问题（Word-wrap problem）的算法与源代码

C#，文字排版的折行问题（Word-wrap problem）的算法与源代码

1、英文的折行问题 给定一个单词序列&#xff0c;以及一行中可以输入的字符数限制&#xff08;线宽&#xff09;。 在给定的顺序中放置换行符&#xff0c;以便打印整齐。 假设每个单词的长度小于线宽。 像MS word这样的文字处理程序负责放置换行符。 这个想法是要有平衡的线条。…
阅读更多...
专业140+总分430+西南交通大学924信号与系统考研经验电子信息与通信工程，真题，大纲，参考书

专业140+总分430+西南交通大学924信号与系统考研经验电子信息与通信工程，真题，大纲，参考书

今年报考西南交通大学&#xff0c;考研分数专业课924信号与系统140&#xff0c;总分430&#xff0c;各门分数都还是比较均衡&#xff0c;经过一年的复习&#xff0c;有得有失&#xff0c;总结一下自己的复习经历&#xff0c;希望给大家有点帮助&#xff0c;在复习中做的更好&am…
阅读更多...
钉钉平台“智”领宠物界，开启萌宠智能新时代！

钉钉平台“智”领宠物界，开启萌宠智能新时代！

在当前数字化转型的浪潮中&#xff0c;钉钉用便捷的数字化解决方案推动了宠物业界的智能升级。一家宠物用品公司采用无雀科技数字化管理系统&#xff0c;与钉钉平台结合&#xff0c;解决了小型企业普遍存在的财务管理不清晰、业务流程不规范、客户信息核对繁琐等痛点问题。 针对…
阅读更多...
AHU 汇编 实验六

AHU 汇编 实验六

一、实验名称&#xff1a;实验6 输入一个16进制数&#xff0c;把它转换为10进制数输出 实验目的&#xff1a; 培养汇编中设计子程序的能力 实验过程&#xff1a; 源代码&#xff1a; data segmentbuff1 db Please input a number(H):$buff2 db 30,?,30 dup(?),13,10buff3 …
阅读更多...
多维时序 | Matlab实现VMD-CNN-LSTM变分模态分解结合卷积神经网络结合长短期记忆神经网络多变量时间序列预测

多维时序 | Matlab实现VMD-CNN-LSTM变分模态分解结合卷积神经网络结合长短期记忆神经网络多变量时间序列预测

多维时序 | Matlab实现VMD-CNN-LSTM变分模态分解结合卷积神经网络结合长短期记忆神经网络多变量时间序列预测 目录 多维时序 | Matlab实现VMD-CNN-LSTM变分模态分解结合卷积神经网络结合长短期记忆神经网络多变量时间序列预测预测效果基本介绍程序设计参考资料 预测效果 基本介…
阅读更多...
鸿蒙Harmony应用开发—ArkTS声明式开发（基础手势：Stepper）

鸿蒙Harmony应用开发—ArkTS声明式开发（基础手势：Stepper）

步骤导航器组件&#xff0c;适用于引导用户按照步骤完成任务的导航场景。 说明&#xff1a; 该组件从API Version 8开始支持。后续版本如有新增内容&#xff0c;则采用上角标单独标记该内容的起始版本。 子组件 仅能包含子组件StepperItem。 接口 Stepper(value?: { index?…
阅读更多...
微信小程序小白易入门基础教程1

微信小程序小白易入门基础教程1

微信小程序 基本结构 页面配置 页面配置 app.json 中的部分配置&#xff0c;也支持对单个页面进行配置&#xff0c;可以在页面对应的 .json 文件来对本页面的表现进行配置。 页面中配置项在当前页面会覆盖 app.json 中相同的配置项&#xff08;样式相关的配置项属于 app.js…
阅读更多...
鸿蒙Harmony应用开发—ArkTS声明式开发（基础手势：RichText）

鸿蒙Harmony应用开发—ArkTS声明式开发（基础手势：RichText）

富文本组件&#xff0c;解析并显示HTML格式文本。 说明&#xff1a; 该组件从API Version 8开始支持。后续版本如有新增内容&#xff0c;则采用上角标单独标记该内容的起始版本。该组件无法根据内容自适应设置宽高属性&#xff0c;需要开发者设置显示布局。 子组件 不包含子组…
阅读更多...
基于YOLOv8/YOLOv7/YOLOv6/YOLOv5的石头剪刀布手势识别系统详解（深度学习模型+UI界面代码+训练数据集）

基于YOLOv8/YOLOv7/YOLOv6/YOLOv5的石头剪刀布手势识别系统详解（深度学习模型+UI界面代码+训练数据集）

摘要&#xff1a;本篇博客深入探讨了使用深度学习技术开发石头剪刀布手势识别系统的过程&#xff0c;并分享了完整代码。该系统利用先进的YOLOv8、YOLOv7、YOLOv6、YOLOv5算法&#xff0c;并对这几个版本进行性能对比&#xff0c;如mAP、F1 Score等关键指标。文章详细阐述了YOL…
阅读更多...
基于YOLOv8/YOLOv7/YOLOv6/YOLOv5的远距离停车位检测系统（深度学习代码+UI界面+训练数据集）

基于YOLOv8/YOLOv7/YOLOv6/YOLOv5的远距离停车位检测系统（深度学习代码+UI界面+训练数据集）

摘要&#xff1a;开发远距离停车位检测系统对于提高停车效率具有关键作用。本篇博客详细介绍了如何运用深度学习构建一个远距离停车位检测系统&#xff0c;并提供了完整的实现代码。该系统基于强大的YOLOv8算法&#xff0c;并对比了YOLOv7、YOLOv6、YOLOv5&#xff0c;展示了不…
阅读更多...
TS271IDT运算放大器芯片中文资料PDF数据手册引脚图图片参数价格功能

TS271IDT运算放大器芯片中文资料PDF数据手册引脚图图片参数价格功能

产品描述&#xff1a; TS271 是一款低成本、低功耗的单通道运算放大器&#xff0c;设计用于采用单电源或双电源供电。该运算放大器采用意法半导体硅栅CMOS工艺&#xff0c;具有出色的消耗-速度比。该放大器非常适合低功耗应用。 电源可通过引脚 8 和 4 之间连接的电阻器进行外…
阅读更多...
2024年【高压电工】考试及高压电工考试总结

2024年【高压电工】考试及高压电工考试总结

题库来源&#xff1a;安全生产模拟考试一点通公众号小程序 高压电工考试考前必练&#xff01;安全生产模拟考试一点通每个月更新高压电工考试总结题目及答案&#xff01;多做几遍&#xff0c;其实通过高压电工模拟试题很简单。 1、【单选题】 为了保证频率偏差不超过规定值,必…
阅读更多...
基于Springboot的预报名管理系统（有报告）。Javaee项目，springboot项目。

基于Springboot的预报名管理系统（有报告）。Javaee项目，springboot项目。

演示视频&#xff1a; 基于Springboot的预报名管理系统&#xff08;有报告&#xff09;。Javaee项目&#xff0c;springboot项目。 项目介绍&#xff1a; 采用M&#xff08;model&#xff09;V&#xff08;view&#xff09;C&#xff08;controller&#xff09;三层体系结构&a…
阅读更多...
用miniconda建立PyTorch、Keras、TensorFlow三个环境

用miniconda建立PyTorch、Keras、TensorFlow三个环境

一、配置清华镜像conda源 由于网络问题&#xff0c;直接使用conda默认的源下载包可能会非常慢。为了解决这个问题&#xff0c;可以配置国内镜像源来加速包的下载。清华大学TUNA协会提供了一个常用的conda镜像源。下面是如何配置清华镜像源的步骤&#xff1a; 1. 配置清华conda…
阅读更多...
发布DDD脚手架到Maven仓库，IntelliJ IDEA 配置一下即可使用

发布DDD脚手架到Maven仓库，IntelliJ IDEA 配置一下即可使用

作者&#xff1a;小傅哥 博客&#xff1a;https://bugstack.cn 项目&#xff1a;https://gaga.plus 沉淀、分享、成长&#xff0c;让自己和他人都能有所收获&#xff01;&#x1f604; 大家好&#xff0c;我是技术UP主&#xff0c;小傅哥。 这篇文章将帮助粉丝伙伴们更高效地利…
阅读更多...
使用 Jenkins 和 Spinnaker 构建 Kubernetes CI/CD

使用 Jenkins 和 Spinnaker 构建 Kubernetes CI/CD

无论您是新手还是持续集成和持续交付以及容器化领域的经验丰富&#xff0c;本文都将为您提供设置 Spinnaker 以满足您的软件应用程序交付需求的基本知识。 了解 Jenkins、Spinnaker 和 Kubernetes Kubernetes 和 Jenkins 是两个强大的工具&#xff0c;它们相互配合&#xff0…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023