一种基于springboot、redis的分布式任务引擎的实现(一)
总体思路是,主节点接收到任务请求,将根据任务情况拆分成多个任务块,将任务块标识的主键放入redis。发送redis消息,等待其他节点运行完毕,结束处理。接收到信息的节点注册本节点信息到redis、开启多线程、获取任务块、执行任务、结束处理。
1、主节点接收任务请求
@Overridepublic void executeTaskInfo(PrepareDTO prepareDTO) {//异常标记String taskInfo = prepareDTO.getTaskId();//任务分组状态String taskStatus = "";try {log.info("数据准备任务并设定任务执行状态,{}", prepareDTO);this.dataPrepareBo.doStartGroupJobInfo(prepareDTO);//给redis集合中放计算对象log.info("开始放入计算任务:{}", prepareDTO);boolean getTaskFlag = this.dataPrepareBo.pushCalculationObject(prepareDTO);if (!getTaskFlag) {taskStatus = String.format("没有获取数据或计划已取消,%s", taskInfo);log.error(taskStatus);throw new Exception(taskStatus);}//发消息执行缓存中任务log.info("发消息执行任务:{}", prepareDTO);sendMessage(prepareDTO);//等待任务执行完毕log.info("等待任务执行结果");taskStatus = this.getGroupUpLoadTaskFinsh(prepareDTO);} catch (Exception e) {//捕获日志e.printStackTrace();taskStatus = "获取任务状态异常" + e;log.info(taskStatus);dataPrepareBo.putExceptionMsg2Cache(taskInfo, "数据准备分发计算任务线程异常:" + taskStatus);} finally {//做任务结束处理this.doGroupTaskFinshpPocess(prepareDTO, taskStatus);}}
2,发送消息
@Overridepublic void sendMessage(String topic, String msg) {this.redisTemplate.convertAndSend(topic, msg);}
3,节点接收任务,并执行
public void doUpLoadTask(String msg) throws Exception {log.info("开始执行明细任务{}" + msg);String taskId = this.getTaskId(msg);try {Object cancelFlag = this.redisTemplate.opsForValue().get(String.format(EngineConstant.JOB_CANCEL_FLAG, taskId));if(cancelFlag != null && "1".equals(cancelFlag.toString())){log.info("本次任务已取消");return;}//上传本机执行信息到redisthis.cacheBo.initGroupUpLoadTaskStats(taskId,ENGINE_DISTRIBUTION_RUNNING.getKey());//从缓存获取任务,获取任务后启线程执行任务。如果没获取到任务,则本节点任务执行完毕//循环获取任务this.groupTaskProcessBO.doGroupTaskProcess(taskId, null);//处理结束this.cacheBo.finishGroupUpLoadTaskStats(taskId,ENGINE_DISTRIBUTION_RUNNING.getKey());} catch (Exception e) {//记录日志taskUpldExeLogCDTO.setRunStas("-1");String exceptionInfo = this.taskLogUtils.getExceptionInfo(e) ;taskUpldExeLogCDTO.setAbnInfo(exceptionInfo);throw e;} finally {//记录日志taskUpldExeLogCDTO.setEndtime(DateUtil.getCurrentDate());if("-1".equals(taskUpldExeLogCDTO.getRunStas())){//异常结束this.taskLogUtils.sendLogInfo(taskUpldExeLogCDTO,"执行上传任务异常");} else {//正常结束taskUpldExeLogCDTO.setRunStas("1");this.taskLogUtils.sendLogInfo(taskUpldExeLogCDTO,"执行上传任务正常");}}}
4,开启线程执行任务
@Overridepublic CalculationDTO doGroupTaskProcess(String taskId, TaskUpldExeLogCDTO taskUpldExeLogCDTO) throws Exception {List<Future> futureList = new ArrayList<>();//开始执行明细任务处理ThreadPoolTaskExecutor taskTransferExecutor = ToolUtil.getExecutor("engine-file-tasks-pool-", Math.min(parallelProcessNum,10), 8);ExecutorListHolder.putThreadPool(String.format(GroupConstant.PREPARE_ENGINE_POOL,taskId), taskTransferExecutor.getThreadPoolExecutor());for(int i = 0 ; i < parallelProcessNum ; i++) {DoGroupUpLoadTaskThread doGroupUpLoadTaskThread = new DoGroupUpLoadTaskThread(taskId, redisTemplate, calculationBo, null, null);Future<?> future = taskTransferExecutor.submit(doGroupUpLoadTaskThread);futureList.add(future);}if (!CollectionUtil.isEmpty(futureList)) {futureList.forEach(f -> {try {f.get(GroupTaskProcessBOImpl.maxTime, TimeUnit.SECONDS);} catch (Exception e) {e.printStackTrace();}});}log.info("本节点执行分组任务执行完毕{}", taskId + ":" + GroupConstant.IDENTITY);return null;}
5,线程执行明细
@Overridepublic ResponseDTO call() throws Exception {//执行任务while(true) {FilterTableUniqueDTO filterTableUniqueDTO = (FilterTableUniqueDTO)this.redisTemplate.opsForList().leftPop(String.format(ENGINE_FILTERTABLEUNIQUE_QUEUE.getKey(), taskId));log.debug("取出任务:" + filterTableUniqueDTO);if(null == filterTableUniqueDTO) {break ;}long lastNum = this.redisTemplate.opsForList().size(String.format(ENGINE_FILTERTABLEUNIQUE_QUEUE.getKey(), taskId));log.info("生成文件剩余任务数量:" + lastNum);
// 处理任务calculationBo.GenerateFile(filterTableUniqueDTO, taskUpldDetlLogCDTO);}return null;}
以上是主要入口总体思路涉及代码,详细实现整理起来涉及内容比较繁多,将在第二部分分享。
相关文章:
一种基于springboot、redis的分布式任务引擎的实现(一)
总体思路是,主节点接收到任务请求,将根据任务情况拆分成多个任务块,将任务块标识的主键放入redis。发送redis消息,等待其他节点运行完毕,结束处理。接收到信息的节点注册本节点信息到redis、开启多线程、获取任务块、执…...

基于IDE Eval Resetter延长IntelliJ IDEA等软件试用期的方法(包含新版本软件的操作方法)
本文介绍基于IDE Eval Resetter插件,对集成开发环境IntelliJ IDEA等JetBrains公司下属的多个开发软件,加以试用期延长的方法。 我们这里就以IntelliJ IDEA为例,来介绍这一插件发挥作用的具体方式。不过,需要说明使用IDE Eval Rese…...

RocketMQ消费者可以手动消费但无法主动消费问题,或生成者发送超时
1.大多数是配置问题 修改rocketmq文件夹broker.conf 2.配置与集群IP或本地IPV4一样 重启 在RocketMQ独享实例中支持IPv4和IPv6双栈,主要是通过在网络层面上同时支持IPv4和IPv6协议栈来实现的。RocketMQ的Broker端、Namesrv端和客户端都需要支持IPv4和IPv6协议&…...

【数据库系统】--【2】DBMS架构
DBMS架构 01DBMS架构概述02 DBMS的物理架构03 DBMS的运行和数据架构DBMS的运行架构DBMS的数据架构PostgreSQL的体系结构RMDB的运行架构 04DBMS的逻辑和开发架构DBMS的层次结构DBMS的开发架构DBMS的代码架构 05小结 01DBMS架构概述 02 DBMS的物理架构 数据库系统的体系结构 数据…...

第三章 图论 No.13拓扑排序
文章目录 裸题:1191. 家谱树差分约束拓扑排序:1192. 奖金集合拓扑序:164. 可达性统计差分约束拓扑序:456. 车站分级 拓扑序和DAG有向无环图联系在一起,通常用于最短/长路的线性求解 裸题:1191. 家谱树 119…...

喜报 | 擎创再度入围IDC中国FinTech 50榜单
8月16日,2023年度“IDC中国FinTech 50”榜单正式揭晓,擎创科技继2022年入选该榜单后,再次以创新者姿态成功入选,并以技术赋能业务创新,成为中国金融科技领域创新与活力的重要贡献者。 “IDC中国FinTech 50”旨在评选出…...

【C++ 记忆站】引用
文章目录 一、引用概念二、引用特性1、引用在定义时必须初始化2、一个变量可以有多个引用3、引用一旦引用一个实体,再不能引用其他实体 三、常引用四、使用场景1、做参数1、输出型参数2、大对象传参 2、做返回值1、传值返回2、传引用返回 五、传值、传引用效率比较六…...

Hlang--用Python写个编程语言-变量的实现
文章目录 前言语法规则表示次幂实现变量实现优先级实现步骤解析关键字语法解析解释器总结前言 先前的话,我们终于是把我们整个架子搭起来了,这里重复一下我们的流程,那就是,首先,我们通过解析文本,然后呢遍历文本当中的我们定义的合法关键字,然后呢,把他们封装为一个T…...

多维时序 | MATLAB实现PSO-CNN-BiLSTM多变量时间序列预测
多维时序 | MATLAB实现PSO-CNN-BiLSTM多变量时间序列预测 目录 多维时序 | MATLAB实现PSO-CNN-BiLSTM多变量时间序列预测基本介绍模型特点程序设计参考资料 基本介绍 本次运行测试环境MATLAB2021b,MATLAB实现PSO-CNN-BiLSTM多变量时间序列预测。代码说明:…...

实现Java异步调用的高效方法
文章目录 为什么需要异步调用?Java中的异步编程方式1. 使用多线程2. 使用Java异步框架 异步调用的关键细节结论 🎉欢迎来到Java学习路线专栏~实现Java异步调用的高效方法 ☆* o(≧▽≦)o *☆嗨~我是IT陈寒🍹✨博客主页:IT陈寒的博…...

批量提取文件名到excel,详细的提取步骤
如何批量提取文件名到excel?我们的电脑中可能存储着数量非常多的电子文件,现在需要快速将这些文件的名称全部提取到Excel中。虽然少量数据可以通过复制粘贴的方式轻松完成,但是对于上万个数据而言,复制粘贴都是行不通的࿰…...
C#中的泛型约束可以用在以下几个地方?
1.泛型类型参数: 在定义泛型类型或泛型方法时,可以使用泛型约束来限制泛型类型参数的类型。这可以确保类型参数满足特定的条件,从而在编译时捕获错误并提供更安全和可靠的代码。 public class MyClass<T> where T : IComparable<T&…...

Linux Vm上部署Docker
创建ubutu虚拟机并远程连接, 参考 https://blog.csdn.net/m0_48468018/article/details/132267096 在终端中切换到root用户,并安装docker服务 2.1 切换到root用户 sudo su2.2 安装docker服务 , 参考 https://docs.docker.com/engine/install/ubuntu/ …...
ubuntu bind dns服务配置
sudo apt-get install bind9 内网搭建DNS服务器,大多数是解析纯内网地址使用。但是偶尔也需要解析外网的地址,所以我们可以配置DNS没有添加A记录的URL时,forward到外网DNS服务器或者内网的其他DNS服务器解析。 打开配置文件: sud…...

安卓的代码加固和其他安全问题
文章目录 安卓加固apk文件结构dex加固过程 其它安全问题 安卓加固 从App的加固技术来看:主流分为dex加密和so加密,目前来看保护dex文件更为重要,因为dex反编译后的java代码可读性更强。 android-ndk: Native Development Kit 官网解释:这套工具使您能在 Android 应…...

关于Linux Docker springboot jar 日志时间不正确 问题解决
使用Springboot项目的jar,制作了一个Docker镜像,启动该镜像后发现容器和容器中的Springboot 项目的日志时间不正确。 解决 查看容器时间命令为: docker exec 容器id date 1. 容器与宿主机同步时间 在启动镜像时候把操作系统的时间通过&q…...

提高批量爬虫工作效率
大家好!作为一名专业的爬虫程序员,我今天要和大家分享一些关于提高批量爬虫工作效率的实用技巧。无论你是要批量采集图片、文本还是视频数据,这些经验都能帮助你在大规模数据采集中事半功倍。废话不多说,让我们开始吧!…...

E96系列电阻阻值和代码、乘数对照表
1、为什么要用代码表示? 0805封装还可以简单易懂写下四位丝印,比如10K的1002,但0603的封装上面再想写下四位丝印就没空间了,就算写了也不容易看不清。 2、E96系列电阻阻值和代码、乘数对照表 下面是E96系列的对照表,…...

基于CentOS7.9安装部署docker(简洁版)
安装部署 1基于官方脚本安装(不推荐 不能自行选择版本) 官方文档:https://docs.docker.com/engine/install/centos/ 2 使用yum安装 阿里云文档:docker-ce镜像_docker-ce下载地址_docker-ce安装教程-阿里巴巴开源镜像站 # ste…...
MySQL常用练手题目
数据库表名和字段设计 1.学生表 Student(s_id,s_name,s_birth,s_sex) 学生编号,学生姓名, 出生年月,学生性别 2.课程表 Course(c_id,c_name,t_id) 课程编号, 课程名称, 教师编号 3.教师表 Teacher(t_id,t_name) 教师编号,教师姓名 4.成绩表 Score (s_id,c_id,s_score) 学生编号…...
设计模式和设计原则回顾
设计模式和设计原则回顾 23种设计模式是设计原则的完美体现,设计原则设计原则是设计模式的理论基石, 设计模式 在经典的设计模式分类中(如《设计模式:可复用面向对象软件的基础》一书中),总共有23种设计模式,分为三大类: 一、创建型模式(5种) 1. 单例模式(Sing…...
React Native 开发环境搭建(全平台详解)
React Native 开发环境搭建(全平台详解) 在开始使用 React Native 开发移动应用之前,正确设置开发环境是至关重要的一步。本文将为你提供一份全面的指南,涵盖 macOS 和 Windows 平台的配置步骤,如何在 Android 和 iOS…...
Qt Widget类解析与代码注释
#include "widget.h" #include "ui_widget.h"Widget::Widget(QWidget *parent): QWidget(parent), ui(new Ui::Widget) {ui->setupUi(this); }Widget::~Widget() {delete ui; }//解释这串代码,写上注释 当然可以!这段代码是 Qt …...

NLP学习路线图(二十三):长短期记忆网络(LSTM)
在自然语言处理(NLP)领域,我们时刻面临着处理序列数据的核心挑战。无论是理解句子的结构、分析文本的情感,还是实现语言的翻译,都需要模型能够捕捉词语之间依时序产生的复杂依赖关系。传统的神经网络结构在处理这种序列依赖时显得力不从心,而循环神经网络(RNN) 曾被视为…...

c#开发AI模型对话
AI模型 前面已经介绍了一般AI模型本地部署,直接调用现成的模型数据。这里主要讲述讲接口集成到我们自己的程序中使用方式。 微软提供了ML.NET来开发和使用AI模型,但是目前国内可能使用不多,至少实践例子很少看见。开发训练模型就不介绍了&am…...
基于matlab策略迭代和值迭代法的动态规划
经典的基于策略迭代和值迭代法的动态规划matlab代码,实现机器人的最优运输 Dynamic-Programming-master/Environment.pdf , 104724 Dynamic-Programming-master/README.md , 506 Dynamic-Programming-master/generalizedPolicyIteration.m , 1970 Dynamic-Programm…...

以光量子为例,详解量子获取方式
光量子技术获取量子比特可在室温下进行。该方式有望通过与名为硅光子学(silicon photonics)的光波导(optical waveguide)芯片制造技术和光纤等光通信技术相结合来实现量子计算机。量子力学中,光既是波又是粒子。光子本…...

计算机基础知识解析:从应用到架构的全面拆解
目录 前言 1、 计算机的应用领域:无处不在的数字助手 2、 计算机的进化史:从算盘到量子计算 3、计算机的分类:不止 “台式机和笔记本” 4、计算机的组件:硬件与软件的协同 4.1 硬件:五大核心部件 4.2 软件&#…...
uniapp 集成腾讯云 IM 富媒体消息(地理位置/文件)
UniApp 集成腾讯云 IM 富媒体消息全攻略(地理位置/文件) 一、功能实现原理 腾讯云 IM 通过 消息扩展机制 支持富媒体类型,核心实现方式: 标准消息类型:直接使用 SDK 内置类型(文件、图片等)自…...
Docker拉取MySQL后数据库连接失败的解决方案
在使用Docker部署MySQL时,拉取并启动容器后,有时可能会遇到数据库连接失败的问题。这种问题可能由多种原因导致,包括配置错误、网络设置问题、权限问题等。本文将分析可能的原因,并提供解决方案。 一、确认MySQL容器的运行状态 …...