大熊猫分布式组件开发系列教程（三）

2023-11-01 06:50
文章标签 组件 教程 开发 系列 分布式 大熊猫

本文主要是介绍大熊猫分布式组件开发系列教程（三），希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

今天我们来看看springboot定时任务如何做成分布式组件来供项目集成依赖使用，接下来就跟着大熊猫一起做crontask分布式组件开发。

首先我们先创建一个crontask模块

其实这个定时任务组件最主要的操作就是定时任务记录，以及定时任务日志这两张表

接着就是一些工厂类的封装，监听类的实现

编写JpJob实现Job的excute方法，以及编写执行之前的方法，执行后的方法。

package com.panda.common.crontask.web.schedule;
import com.panda.common.crontask.common.DateUtil;
import com.panda.common.crontask.service.api.QuartzConfigService;
import com.panda.common.crontask.service.api.QuartzLogService;
import com.panda.common.crontask.service.api.dto.QuartzConfigDto;
import com.panda.common.crontask.service.api.dto.QuartzLogDto;
import org.quartz.Job;
import org.quartz.JobExecutionContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import java.io.Serializable;
import java.time.Duration;
import java.time.LocalDateTime;
public abstract class JpJob implements Job, Serializable {private static Logger logger = LoggerFactory.getLogger(JpJob.class);@Autowiredprivate QuartzLogService quartzLogService;@Autowiredprivate QuartzConfigService quartzConfigService;//任务开始时间private ThreadLocal<LocalDateTime> beforTime = new ThreadLocal<>();//日志Idprivate ThreadLocal<String> logId = new ThreadLocal<>();public abstract void runJob();public void beforeRun(){String clazz = this.getClass().getName();QuartzLogDto quartzLogDto = new QuartzLogDto();quartzLogDto.setClazz(clazz);QuartzConfigDto quartzConfigDto = quartzConfigService.findByClazz(clazz);if(quartzConfigDto != null){quartzLogDto.setQuartzId(quartzConfigDto.getId());quartzLogDto.setName(quartzConfigDto.getName());}quartzLogDto = quartzLogService.add(quartzLogDto);beforTime.set(LocalDateTime.now());logId.set(quartzLogDto.getId());}public void error(Exception e){QuartzLogDto quartzLogDto = quartzLogService.get(logId.get());Duration between = Duration.between(beforTime.get(), LocalDateTime.now());quartzLogDto.setSpendTime(between.toMillis());if(e != null && e.getMessage() != null)quartzLogDto.setExceptionMessage(e.getMessage().length() > 500 ? e.getMessage().substring(0, 499) : e.getMessage());quartzLogDto.setResult(0);quartzLogService.update(logId.get(), quartzLogDto);}public void afterRun(){QuartzLogDto quartzLogDto = quartzLogService.get(logId.get());Duration between = Duration.between(beforTime.get(), LocalDateTime.now());quartzLogDto.setSpendTime(between.toMillis());quartzLogDto.setResult(1);quartzLogService.update(logId.get(), quartzLogDto);}@Overridepublic void execute(JobExecutionContext jobExecutionContext) {String clazz = this.getClass().getName();logger.info("==== 定时任务 "+clazz+" ====> 开启 " + DateUtil.getStringToday());beforeRun();try {runJob();} catch (Exception e) {logger.info("==== 定时任务 "+clazz+" ====> 异常 "+e.getMessage());error(e);} finally {logger.info("==== 定时任务 "+clazz+" ====> 结束 " + DateUtil.getStringToday());afterRun();}}
}

编写JpJobFactory继承从而创建单例Job进行注入。

package com.panda.common.crontask.web.schedule;
import org.quartz.spi.TriggerFiredBundle;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.AutowireCapableBeanFactory;
import org.springframework.scheduling.quartz.AdaptableJobFactory;
import org.springframework.stereotype.Component;
@Component
public class JpJobFactory extends AdaptableJobFactory {@Autowiredprivate AutowireCapableBeanFactory capableBeanFactory;@Overrideprotected Object createJobInstance(TriggerFiredBundle bundle) throws Exception {// 调用父类的方法Object jobInstance = super.createJobInstance(bundle);// 进行注入capableBeanFactory.autowireBean(jobInstance);return jobInstance;}
}

调度工厂类实现任务配置读取服务，项目启动激活所有定时任务，任务暂停，恢复，以及执行一次任务的方法。

package com.panda.common.crontask.web.schedule;
import com.panda.base.service.api.exception.ServiceException;
import com.panda.common.crontask.service.api.QuartzConfigService;
import com.panda.common.crontask.service.api.dto.QuartzConfigDto;
import org.quartz.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.quartz.SchedulerFactoryBean;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;
import java.util.List;
/*** 调度工厂类**/
@Service
@Component
public class JpSchedulerFactory {private static Logger logger = LoggerFactory.getLogger(JpSchedulerFactory.class);@AutowiredSchedulerFactoryBean schedulerFactoryBean;// 任务配置读取服务@Autowiredprivate QuartzConfigService quartzConfigService;public void scheduleJobs() {Scheduler scheduler = getScheduler();startJob(scheduler);}// 获取schedulerprivate Scheduler getScheduler(){return schedulerFactoryBean.getScheduler();}// 项目启动,开启所有激活的任务private void startJob(Scheduler scheduler)  {try {// 获取所有激活的任务List<QuartzConfigDto> jobList = quartzConfigService.findByStatus(1);for (QuartzConfigDto config : jobList) {Class<? extends Job> clazz = (Class<? extends Job>) Class.forName(config.getClazz());JobDetail jobDetail = JobBuilder.newJob(clazz).withIdentity(config.getId()).build();CronScheduleBuilder scheduleBuilder = CronScheduleBuilder.cronSchedule(config.getCron());CronTrigger cronTrigger = TriggerBuilder.newTrigger().withIdentity(config.getId()).withSchedule(scheduleBuilder).build();scheduler.scheduleJob(jobDetail, cronTrigger);}} catch (ClassNotFoundException e) {e.printStackTrace();}catch (SchedulerException e) {e.printStackTrace();}}// 任务暂停public void pauseJob(String id) throws SchedulerException {QuartzConfigDto quartzConfigDto = quartzConfigService.get(id);if(quartzConfigDto == null) throw new ServiceException(502,"不存在的任务");JobKey jobKey = JobKey.jobKey(id);Scheduler scheduler = getScheduler();scheduler.deleteJob(jobKey);}// 任务恢复public void resumeJob(String id) throws SchedulerException, ClassNotFoundException {QuartzConfigDto quartzConfigDto = quartzConfigService.get(id);if(quartzConfigDto == null) throw new ServiceException(502,"不存在的任务");JobKey jobKey = JobKey.jobKey(id);Scheduler scheduler = getScheduler();Class<? extends Job> clazz = (Class<? extends Job>) Class.forName(quartzConfigDto.getClazz());JobDetail detail = scheduler.getJobDetail(jobKey);if (detail == null){JobDetail jobDetail = JobBuilder.newJob(clazz).withIdentity(id).build();CronScheduleBuilder scheduleBuilder = CronScheduleBuilder.cronSchedule(quartzConfigDto.getCron());CronTrigger cronTrigger = TriggerBuilder.newTrigger().withIdentity(id).withSchedule(scheduleBuilder).build();scheduler.scheduleJob(jobDetail, cronTrigger);}else {scheduler.resumeJob(jobKey);}}// 执行一次任务public void runJob(String id) throws SchedulerException {QuartzConfigDto quartzConfigDto = quartzConfigService.get(id);if(quartzConfigDto == null) throw new ServiceException(502,"不存在的任务");JobKey jobKey = JobKey.jobKey(id);Scheduler scheduler = getScheduler();scheduler.triggerJob(jobKey);}
}

定时任务运行工厂类用来springboot启动监听和注入SchedulerFactoryBean

package com.panda.common.crontask.web.schedule;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationListener;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.scheduling.quartz.SchedulerFactoryBean;
/*** 定时任务运行工厂类**/
@Configuration
public class StartSchedulerListener implements ApplicationListener<ContextRefreshedEvent> {@Autowiredpublic com.panda.common.crontask.web.schedule.JpSchedulerFactory jpSchedulerFactory;@Autowiredprivate com.panda.common.crontask.web.schedule.JpJobFactory jpJobFactory;// springboot 启动监听@Overridepublic void onApplicationEvent(ContextRefreshedEvent event) {jpSchedulerFactory.scheduleJobs();}//注入SchedulerFactoryBean@Beanpublic SchedulerFactoryBean schedulerFactoryBean() {SchedulerFactoryBean schedulerFactoryBean = new SchedulerFactoryBean();schedulerFactoryBean.setJobFactory(jpJobFactory);return schedulerFactoryBean;}
}

下面贴出api层代码

package com.panda.common.crontask.web.api;
import com.panda.auth.client.Authorization;
import com.panda.base.service.api.exception.ServiceException;
import com.panda.base.web.api.jersey.IResponse;
import com.panda.base.web.api.jersey.security.ISecurityContext;
import com.panda.base.web.api.response.WebApiResponse;
import com.panda.common.crontask.common.ApplicationConstants;
import com.panda.common.crontask.service.api.QuartzConfigService;
import com.panda.common.crontask.service.api.QuartzLogService;
import com.panda.common.crontask.service.api.dto.QuartzConfigDto;
import com.panda.common.crontask.service.api.dto.QuartzLogDto;
import com.panda.common.crontask.web.schedule.JpSchedulerFactory;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiParam;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Pageable;
import org.springframework.data.domain.Sort;
import org.springframework.stereotype.Component;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;
import javax.annotation.security.RolesAllowed;
import javax.ws.rs.*;
import javax.ws.rs.core.Context;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.SecurityContext;
import java.time.LocalDateTime;
@Path("quartz")
@Component
@Produces(MediaType.APPLICATION_JSON)
@Consumes(MediaType.APPLICATION_JSON)
@Api(tags = "【公共组件-定时任务】") 
@RolesAllowed("*")
@Authorization(permission = "quartz:*")
public class QuartzApi implements ISecurityContext, IResponse {private final static Logger log = LoggerFactory.getLogger(QuartzApi.class);@AutowiredQuartzConfigService quartzConfigService;@AutowiredQuartzLogService quartzLogService;//事务模版@Autowiredprivate TransactionTemplate transactionTemplate;@AutowiredJpSchedulerFactory JpSchedulerFactory;@ApiOperation(value = "获取任务分页")@GETpublic WebApiResponse<Page<QuartzConfigDto>> getQuartzConfigPage(@Context SecurityContext context,// 如果@Authorization存在，则不可缺少次参数。
@ApiParam(value = "页码(从0开始),默认0") @QueryParam("pageIndex") Integer pageIndex,
@ApiParam(value = "页大小,默认10") @QueryParam("pageSize") Integer pageSize,
@ApiParam(value = "任务名称") @QueryParam("name") String name,
@ApiParam(value = "任务状态 1-正常，0-停止") @QueryParam("status") Integer status) {if (pageIndex == null) pageIndex = 0;if (pageSize == null) pageSize = 10;Pageable pageable = PageRequest.of(pageIndex, pageSize,Sort.by(Sort.Direction.DESC, "updateTime"));return response(quartzConfigService.findQuartzConfigPage(status,name,pageable));}@ApiOperation(value = "通过Id获取任务详情")@GET@Path("{id}")public WebApiResponse<QuartzConfigDto> getQuartzConfigById(@Context SecurityContext context,@ApiParam(required = true, value = "quartzConfig Id") @PathParam("id") String id) {return response(quartzConfigService.get(id));}@ApiOperation(value = "通过类名获取任务详情")@GET@Path("getByClazz/{clazz}")public WebApiResponse<QuartzConfigDto> getQuartzConfigByClazz(@Context SecurityContext context,@ApiParam(required = true, value = "clazz") @PathParam("clazz") String clazz ) {return response(quartzConfigService.findByClazz(clazz));}@ApiOperation(value = "添加任务")@POST@Path("add")public WebApiResponse<QuartzConfigDto> addQuartzConfig(QuartzConfigDto quartzConfigDto, @Context SecurityContext context) {return response(quartzConfigService.addOrUpdateQuartzConfig(quartzConfigDto));}@ApiOperation(value = "删除任务")@POST@Path("delete/{id}")public WebApiResponse deleteQuartzConfig(@Context SecurityContext context, @ApiParam(required = true, value = "quartzConfig Id") @PathParam("id") String id) {return transactionTemplate.execute(new TransactionCallback<WebApiResponse>() {@Overridepublic WebApiResponse doInTransaction(TransactionStatus transactionStatus) {try {JpSchedulerFactory.pauseJob(id);quartzConfigService.delete(id);} catch (Exception e) {log.error("删除任务任务失败！", e);transactionStatus.setRollbackOnly();throw new ServiceException(502,"删除任务失败");}log.info("删除任务成功！");return done();}});}@ApiOperation(value = "更新任务")@POSTpublic WebApiResponse<QuartzConfigDto> updateQuartzConfig(QuartzConfigDto quartzConfigDto, @Context SecurityContext context) {try {JpSchedulerFactory.pauseJob(quartzConfigDto.getId());quartzConfigDto = quartzConfigService.updateQuartzConfig(quartzConfigDto);JpSchedulerFactory.resumeJob(quartzConfigDto.getId());} catch (Exception e) {log.error("更新任务失败！", e);throw new ServiceException(502,"暂停任务失败");}log.info("更新任务成功！");return response(quartzConfigDto);}@ApiOperation(value = "暂停任务")@GET@Path("pause/{id}")public WebApiResponse pauseQuartzConfig(@Context SecurityContext context, @ApiParam(required = true, value = "quartzConfig Id") @PathParam("id") String id) {return transactionTemplate.execute(new TransactionCallback<WebApiResponse>() {@Overridepublic WebApiResponse doInTransaction(TransactionStatus transactionStatus) {try {JpSchedulerFactory.pauseJob(id);quartzConfigService.updateJobStatus(id, ApplicationConstants.PAUSE_CODE);} catch (Exception e) {log.error("暂停任务失败！", e);transactionStatus.setRollbackOnly();throw new ServiceException(502,"暂停任务失败");}log.info("暂停任务成功！");return done();}});}@ApiOperation(value = "恢复任务")@GET@Path("resume/{id}")public WebApiResponse resumeQuartzConfig(@Context SecurityContext context, @ApiParam(required = true, value = "quartzConfig Id") @PathParam("id") String id) {return transactionTemplate.execute(new TransactionCallback<WebApiResponse>() {@Overridepublic WebApiResponse doInTransaction(TransactionStatus transactionStatus) {try {JpSchedulerFactory.resumeJob(id);quartzConfigService.updateJobStatus(id,ApplicationConstants.ACTIVE_CODE);} catch (Exception e) {log.error("恢复任务失败！", e);transactionStatus.setRollbackOnly();throw new ServiceException(502,"恢复任务失败");}log.info("恢复任务成功！");return done();}});}@ApiOperation(value = "执行一次任务")@GET@Path("run/{id}")public WebApiResponse runQuartzConfig(@Context SecurityContext context, @ApiParam(required = true, value = "quartzConfig Id") @PathParam("id") String id) {try {JpSchedulerFactory.runJob(id);} catch (Exception e) {log.error("执行任务失败！", e);throw new ServiceException(502,"执行任务失败");}log.info("执行任务成功！");return done();}@ApiOperation(value = "获取任务日志分页")@GET@Path("log")public WebApiResponse<Page<QuartzLogDto>> getQuartzLogPage(@Context SecurityContext context,@ApiParam(required = false, value = "页码(从0开始),默认0") @QueryParam("pageIndex") Integer pageIndex,@ApiParam(required = false, value = "页大小,默认10") @QueryParam("pageSize") Integer pageSize,@ApiParam(value = "任务Id") @QueryParam("quartzId") String quartzId,@ApiParam(value = "任务名称") @QueryParam("name") String name,@ApiParam(value = "执行结果 1-成功，0-失败") @QueryParam("result") Integer result,@ApiParam(value = "开始时间 yyyy-MM-dd 00:00:00") @QueryParam("startTime") LocalDateTime startTime,
@ApiParam(value = "结束时间 yyyy-MM-dd 00:00:00") @QueryParam("endTime") LocalDateTime endTime) {if (pageIndex == null) pageIndex = 0;if (pageSize == null) pageSize = 10;Pageable pageable = PageRequest.of(pageIndex, pageSize, Sort.by(Sort.Direction.DESC, "createTime"));return response(quartzLogService.findQuartzLogPage(quartzId,name,result,startTime,endTime,pageable));}
}

执行实例

package com.panda.common.crontask.web.task;
import com.panda.common.crontask.web.schedule.JpJob;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.stereotype.Component;
import java.io.Serializable;
/*** 定时任务实现类*/
@Component
@EnableScheduling
public class ScheduleTaskTestJob extends JpJob implements Serializable {private static Logger logger = LoggerFactory.getLogger(ScheduleTaskTestJob.class);/*** 执行任务** @throws RuntimeException*/@Overridepublic void runJob() throws RuntimeException {//在这里写定时任务执行的内容logger.info("==== 定时任务 ScheduleTaskTestJob ====> 执行中...... ");}
}

以上就是实现定时任务分布式组件的主要代码，写一个run模块集成一下数据库，做一个启动类就可以运行了，然后上传到neuxs做成依赖就可以供其他项目使用了，想看源码可以关注“蛋皮皮”公众号找大熊猫给你源码。

这篇关于大熊猫分布式组件开发系列教程（三）的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


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

相关文章

Spring Security 从入门到进阶系列教程

Spring Security 从入门到进阶系列教程

Spring Security 入门系列 《保护 Web 应用的安全》 《Spring-Security-入门（一）：登录与退出》 《Spring-Security-入门（二）：基于数据库验证》 《Spring-Security-入门（三）：密码加密》 《Spring-Security-入门（四）：自定义-Filter》 《Spring-Security-入门（五）：在 Sprin
阅读更多...
JS常用组件收集

JS常用组件收集

收集了一些平时遇到的前端比较优秀的组件，方便以后开发的时候查找！！！ 函数工具： Lodash 页面固定： stickUp、jQuery.Pin 轮播： unslider、swiper 开关: switch 复选框： icheck 气泡： grumble 隐藏元素： Headroom
阅读更多...
这15个Vue指令，让你的项目开发爽到爆

这15个Vue指令，让你的项目开发爽到爆

1. V-Hotkey 仓库地址: github.com/Dafrok/v-ho… Demo: 戳这里 https://dafrok.github.io/v-hotkey 安装: npm install --save v-hotkey 这个指令可以给组件绑定一个或多个快捷键。你想要通过按下 Escape 键后隐藏某个组件，按住 Control 和回车键再显示它吗？小菜一碟： <template
阅读更多...
Hadoop企业开发案例调优场景

Hadoop企业开发案例调优场景

需求 （1）需求：从1G数据中，统计每个单词出现次数。服务器3台，每台配置4G内存，4核CPU，4线程。 （2）需求分析： 1G / 128m = 8个MapTask；1个ReduceTask；1个mrAppMaster 平均每个节点运行10个 / 3台 ≈ 3个任务（4    3    3） HDFS参数调优 （1）修改：hadoop-env.sh export HDFS_NAMENOD
阅读更多...
Makefile简明使用教程

Makefile简明使用教程

文章目录 规则makefile文件的基本语法：加在命令前的特殊符号：.PHONY伪目标: Makefilev1 直观写法v2 加上中间过程v3 伪目标v4 变量 make 选项-f-n-C Make 是一种流行的构建工具，常用于将源代码转换成可执行文件或者其他形式的输出文件(如库文件、文档等)。Make 可以自动化地执行编译、链接等一系列操作。 规则 makefile文件
阅读更多...
嵌入式QT开发：构建高效智能的嵌入式系统

嵌入式QT开发：构建高效智能的嵌入式系统

摘要： 本文深入探讨了嵌入式 QT 相关的各个方面。从 QT 框架的基础架构和核心概念出发，详细阐述了其在嵌入式环境中的优势与特点。文中分析了嵌入式 QT 的开发环境搭建过程，包括交叉编译工具链的配置等关键步骤。进一步探讨了嵌入式 QT 的界面设计与开发，涵盖了从基本控件的使用到复杂界面布局的构建。同时也深入研究了信号与槽机制在嵌入式系统中的应用，以及嵌入式 QT 与硬件设备的交互，包括输入输出设
阅读更多...
OpenHarmony鸿蒙开发（ Beta5.0）无感配网详解

OpenHarmony鸿蒙开发（ Beta5.0）无感配网详解

1、简介 无感配网是指在设备联网过程中无需输入热点相关账号信息，即可快速实现设备配网，是一种兼顾高效性、可靠性和安全性的配网方式。 2、配网原理 2.1 通信原理 手机和智能设备之间的信息传递，利用特有的NAN协议实现。利用手机和智能设备之间的WiFi 感知订阅、发布能力，实现了数字管家应用和设备之间的发现。在完成设备间的认证和响应后，即可发送相关配网数据。同时还支持与常规Sof
阅读更多...
如何在页面调用utility bar并传递参数至lwc组件

如何在页面调用utility bar并传递参数至lwc组件

1.在app的utility item中添加lwc组件： 2.调用utility bar api的方式有两种： 方法一，通过lwc调用： import {LightningElement,api ,wire } from 'lwc';import { publish, MessageContext } from 'lightning/messageService';import Ca
阅读更多...
活用c4d官方开发文档查询代码

活用c4d官方开发文档查询代码

当你问AI助手比如豆包，如何用python禁止掉xpresso标签时候，它会提示到 这时候要用到两个东西。https://developers.maxon.net/论坛搜索和开发文档 比如这里我就在官方找到正确的id描述 然后我就把参数标签换过来
阅读更多...
科研绘图系列：R语言扩展物种堆积图（Extended Stacked Barplot）

科研绘图系列：R语言扩展物种堆积图（Extended Stacked Barplot）

介绍 R语言的扩展物种堆积图是一种数据可视化工具，它不仅展示了物种的堆积结果，还整合了不同样本分组之间的差异性分析结果。这种图形表示方法能够直观地比较不同物种在各个分组中的显著性差异，为研究者提供了一种有效的数据解读方式。 加载R包 knitr::opts_chunk$set(warning = F, message = F)library(tidyverse)library(phyl
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2