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

news2025/1/10 10:22:38

文章目录

  • 前言
  • 一、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 {
    @Override
    public void execute(ShardingContext shardingContext) {

        // 分片参数 0=text,1=image,2=radio,3=vedio
        String  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 0
                break;
            case 1:
                // do something by sharding item 1
                break;
            case 2:
                // do something by sharding item 2
                break;
            // 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 {
    @Override
    public void execute(ShardingContext shardingContext) {
        log.debug("job 执行 error,job名称：{},分片数量:{}",shardingContext.getJobName(),shardingContext.getShardingTotalCount());
        switch (shardingContext.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: ...
        }
    }
}

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 {
    @Autowired
    private MyJob myJob;

    @Autowired
    private MyJob1 myJob1;
    @Autowired
    private 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;

    }

    @Override
    public 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;

    @Override
    public void beforeJobExecuted(ShardingContexts shardingContexts) {
        beginTime = System.currentTimeMillis();
        log.info("===>{} MyElasticJobListener BEGIN TIME: {} <===",shardingContexts.getJobName(),  DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss"));
    }

    @Override
    public 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);
    }

    @Override
    public 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 {
    @Override
    public List fetchData(ShardingContext shardingContext) {
        // 抓取数据
        // 分片参数 0=text,1=image,2=radio,3=vedio
        String 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;
    }

    @Override
    public 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.coloradmin.cn/o/1512114.html

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

相关文章

RANDOMIZE-IN-PLACE随机排列算法

RANDOMIZE-IN-PLACE随机排列算法

给定一个长度为 n n n的数组&#xff0c;如何构造出一个随机排列呢&#xff1f;《算法导论》给了我们一个名为RANDOMIZE-IN-PLACE的随机算法&#xff0c;该算法在数组原址上进行排序&#xff0c;时间复杂度为 O ( n ) O(n) O(n)。下面本文将介绍RANDOMIZE-IN-PLACE的设计思想及…
阅读更多...
代码随想录(day3)——链表

代码随想录(day3)——链表

Leetcode.203 移除链表元素&#xff1a; 203. 移除链表元素 - 力扣&#xff08;LeetCode&#xff09; 对于本题&#xff0c;难点就在于对于头部结点的删除&#xff0c;以及给定链表为空时&#xff0c;如何进行遍历。因为需要遍历链表&#xff0c;假设访问链表下一个结点所对应…
阅读更多...
开源绘图工具 PlantUML 入门教程（常用于画类图、用例图、时序图等）

开源绘图工具 PlantUML 入门教程（常用于画类图、用例图、时序图等）

文章目录 一、类图二、用例图三、时序图 一、类图 类的UML图示 startuml skinparam classAttributeIconSize 0 class Dummy {-field1 : String#field2 : int~method1() : Stringmethod2() : void } enduml定义能见度&#xff08;可访问性&#xff09; startumlclass Dummy {-f…
阅读更多...
基于YOLOv8/YOLOv7/YOLOv6/YOLOv5的行人跌倒检测系统（深度学习+UI界面+完整训练数据集）

基于YOLOv8/YOLOv7/YOLOv6/YOLOv5的行人跌倒检测系统（深度学习+UI界面+完整训练数据集）

摘要&#xff1a;开发行人跌倒检测系统在确保老年人安全方面扮演着至关重要的角色。本篇文章详尽地阐述了如何利用深度学习技术构建一个行人跌倒检测系统&#xff0c;并附上了完整的代码实现。该系统采用了先进的YOLOv8算法&#xff0c;并对YOLOv7、YOLOv6、YOLOv5等先前版本进…
阅读更多...
ARM64汇编05 - MOV系列指令

ARM64汇编05 - MOV系列指令

MOV(wide immediate) MOV 可以将一个立即数移动到寄存器中。 .text:0000000000000834 80 46 82 D2 MOV X0, #0x1234 ; Keypatch modified this from:MOV X0, #0x1234 对应的汇编代码为&#xff1a;80 46 82 D2 看手册可知&#xf…
阅读更多...
多维时序 | Matlab实现VMD-CNN-BiLSTM变分模态分解结合卷积神经网络结合双向长短期记忆神经网络多变量时间序列预测

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

多维时序 | Matlab实现VMD-CNN-BiLSTM变分模态分解结合卷积神经网络结合双向长短期记忆神经网络多变量时间序列预测 目录 多维时序 | Matlab实现VMD-CNN-BiLSTM变分模态分解结合卷积神经网络结合双向长短期记忆神经网络多变量时间序列预测预测效果基本介绍程序设计参考资料 预测…
阅读更多...
利用“定时执行专家”软件的25种任务与12种触发器，提升IT系统管理自动化水平

利用“定时执行专家”软件的25种任务与12种触发器，提升IT系统管理自动化水平

在IT系统管理中&#xff0c;自动化是提高工作效率、减少人为错误的关键。而《定时执行专家》这款软件&#xff0c;以其强大的功能、易用性和毫秒级的执行精度&#xff0c;成为了IT系统管理员的得力助手。今天&#xff0c;我们就来探讨一下如何利用这款软件的25种任务类型和12种…
阅读更多...
如何在Linux系统安装SVN并配置固定公网地址远程访问【内网穿透】

如何在Linux系统安装SVN并配置固定公网地址远程访问【内网穿透】

文章目录 前言1. Ubuntu安装SVN服务2. 修改配置文件2.1 修改svnserve.conf文件2.2 修改passwd文件2.3 修改authz文件 3. 启动svn服务4. 内网穿透4.1 安装cpolar内网穿透4.2 创建隧道映射本地端口 5. 测试公网访问6. 配置固定公网TCP端口地址6.1 保留一个固定的公网TCP端口地址6…
阅读更多...
2024-03-11,12（HTML,CSS）

2024-03-11,12（HTML,CSS）

1.HTML的作用就是在浏览器摆放内容。 2.HTML基本骨架 head&#xff1a;网页头部&#xff0c;是给浏览器看的代码&#xff0c;例如CSS body&#xff1a;网页主体&#xff0c;是给用户看的代码&#xff0c;例如图片&#xff0c;文字。 title&#xff1a;网页标题 3.标签的两种…
阅读更多...
Redis中set，zset

Redis中set，zset

集合类型set中的数据是无序的&#xff0c;不能重复的 SET SADD key value [value....] 将一个或者多个元素添加到集合set中&#xff0c;重复的元素是无法进行添加的 返回值为添加成功的数字smembers key 获取set中所有的元素&#xff0c;返回元素的顺序是无序的sismember key…
阅读更多...
React Hooks 那些事儿

React Hooks 那些事儿

翻了波之前写的文章还有笔记&#xff0c;发现关于前端的文章并不多&#xff08;好歹也划水做过点前端开发&#xff09;。巧了&#xff0c;最近没什么好话题可写&#xff0c;做下 React Hooks 学习笔记吧。 Effect Hook 不得不说 Hook 的出现降低了我们在 React 中处理副作用&…
阅读更多...
【漏洞复现】SpringBlade error/list SQL 注入漏洞

【漏洞复现】SpringBlade error/list SQL 注入漏洞

免责声明&#xff1a;文章来源互联网收集整理&#xff0c;请勿利用文章内的相关技术从事非法测试&#xff0c;由于传播、利用此文所提供的信息或者工具而造成的任何直接或者间接的后果及损失&#xff0c;均由使用者本人负责&#xff0c;所产生的一切不良后果与文章作者无关。该…
阅读更多...
定时执行专家 —— 让工作更高效，生活更便捷

定时执行专家 —— 让工作更高效，生活更便捷

在现代社会&#xff0c;高效的时间管理已经成为我们工作和生活中不可或缺的一部分。为了实现这一目标&#xff0c;我们经常会借助各种工具和软件来辅助我们完成定时任务。今天&#xff0c;我要为大家介绍一款功能强大、操作简便的定时任务执行软件——《定时执行专家》。这款软…
阅读更多...
上海计算机学会 2023年11月月赛 丙组T5 推箱子（数学 思维 排序）

上海计算机学会 2023年11月月赛 丙组T5 推箱子（数学 思维 排序）

第五题&#xff1a;T5推箱子 标签&#xff1a;排序、数学、思维题意&#xff1a;给定 t t t组数据&#xff0c;每组数据给定长度为 n n n的字符串&#xff0c; 表示箱子&#xff0c; _ \_ _表示空格&#xff0c;求把箱子都推到一起&#xff08;即两两箱子之间没有空格&#…
阅读更多...
idea+maven+tomcat+spring 创建一个jsp项目

idea+maven+tomcat+spring 创建一个jsp项目

概述&#xff1a;我真服了&#xff0c;这个垃圾学校还在教jsp&#xff0c;这种技术我虽然早会了&#xff0c;但是之前搞的大多都是springboot web类型的&#xff0c;这里我就复习一下&#xff0c;避免以后忘记这种垃圾技术 第一步&#xff1a;创建maven项目 第二步&#xff1a…
阅读更多...
Leetcode 3.12

Leetcode 3.12

leetcode hot 100 链表1.两两交换链表中的节点2.随机链表的复制3.排序链表 链表 1.两两交换链表中的节点 两两交换链表中的节点 1.必须要设置一个dummy (temp) 结点2.保存第二个节点3.先让第一个节点指向第三个节点4.再让第二个节点指向第一个节点5.最后让dummy指向第二个节点…
阅读更多...
基于深度学习的番茄叶片病害检测系统（含UI界面、yolov8、Python代码、数据集）

基于深度学习的番茄叶片病害检测系统（含UI界面、yolov8、Python代码、数据集）

项目介绍 项目中所用到的算法模型和数据集等信息如下&#xff1a; 算法模型&#xff1a;     yolov8 yolov8主要包含以下几种创新&#xff1a;         1. 可以任意更换主干结构&#xff0c;支持几百种网络主干。 数据集&#xff1a;     网上下载的数据集&#x…
阅读更多...
宏景eHR downlawbase SQL注入漏洞复现

宏景eHR downlawbase SQL注入漏洞复现

0x01 产品简介 宏景eHR人力资源管理软件是一款人力资源管理与数字化应用相融合，满足动态化、协同化、流程化、战略化需求的软件。面向复杂单组织或多组织客户，支持流程，B/S架构。特别适合集团化管理和跨地域使用的产品，融合了最新的互联网技术和先进的人力资源管理理念和实…
阅读更多...
浏览器与服务器通信过程(HTTP协议)

浏览器与服务器通信过程(HTTP协议)

目录 1 概念 2 常见的 web 服务器有 3 浏览器与服务器通信过程 3.1 DNS 3.2 URL 4 HTTP请求方法和应答状态码 4.1 HTTP请求报文段实例 4.2 HTTP请求方法 5 HTTP应答报头和应答状态 5.1 HTTP的应答报头结构 5.2 HTTP的应答状态 1 概念 浏览器与 web 服务器在应用层通…
阅读更多...
【DAY11 软考中级备考笔记】数据结构 查找和排序

【DAY11 软考中级备考笔记】数据结构 查找和排序

数据结构 查找和排序 3月12日 – 天气&#xff1a;晴 1. 顺序查找 顺序查找就是简单的从头一个一个的进行比较&#xff0c;注意它的平均查找长度 2. 折半查找 折半查找和二叉排序树一致&#xff1a; 优点&#xff1a;查找效率很高 缺点&#xff1a;要求必须是循序存储并且表中…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023