科技企业网站源码,临沂网站建设优化,怎么用2级目录做网站,做视频网站多大服务器文章目录 xxl-job 任务调度原理1 问题1 时区导致的任务没有执行的问题解决方案 2 执行器注册和下线导致的问题#xff08;IP问题#xff09;解决方案 3 问题3 调度成功#xff0c;但是执行器的定时任务未执行4 问题4 数据库性能问题#xff0c;导致查询任务和操作日志数据卡… 文章目录 xxl-job 任务调度原理1 问题1 时区导致的任务没有执行的问题解决方案 2 执行器注册和下线导致的问题IP问题解决方案 3 问题3 调度成功但是执行器的定时任务未执行4 问题4 数据库性能问题导致查询任务和操作日志数据卡主解决方案 5. 问题5 定时任务方法中有事务 xxl-job 任务调度原理
任务调度的详细原理主要包括以下几个步骤 Cron表达式解析: Cron表达式是一种用于配置定时任务的表达式它描述了一个定时规则。这个规则会被调度中心解析并生成具体的触发时间。 触发时间监控: 调度中心会实时监控当前时间一旦当前时间达到了Cron表达式解析出的触发时间就会开始进行任务调度。 任务参数获取: 调度中心调度任务时首先会从数据库或内存中获取需要调度的任务的信息包括任务类型、执行参数、执行器地址等。 任务路由策略选择: 根据预设的任务路由策略调度中心会选择一个或多个执行器进行任务调度。路由策略可能是轮询、随机、一致性哈希、最少负载等。 执行器调度: 调度中心根据选择的执行器地址通过RPC远程过程调用或者HTTP请求将任务调度请求发送到执行器。 调度日志记录: 调度中心在完成任务调度后会记录任务调度日志包括调度时间、执行器地址、任务参数等信息以便后续分析和排查问题。 调度结果反馈: 执行器在接收到任务调度请求后会立即返回一个调度结果给调度中心调度中心根据这个结果更新任务调度日志。
1 问题1 时区导致的任务没有执行的问题
这个是最常见的问题。因为xxl-job 最核心的是通过时间计算获得下次任务执行的时间点然后通过内置线程检查进行调度。 所以时间计算错误和时间点检查不匹配也是不会调度的。还有一种是调度了执行器实际没有执行。我们后面也会讲到。
XXL-JOB是一种任务调度平台。虽然在配置时可能不需要你直接指定时区但它实际上是根据服务器的时区设定来执行定时任务的。如果你没有特意去修改服务器的时区那么它就会使用服务器默认的时区。
所以你在配置XXL-JOB时实际上是考虑了时区的只不过你可能并没有显式地去设定它。如果你的服务器在不同的时区或者你需要调度的任务需要在特定的时区执行那么你可能需要考虑到时区的影响。
如果需要修改服务器的时区你可以通过在服务器上执行特定的命令来实现。这通常需要管理员权限。具体的命令可能会因操作系统的不同而不同。
如果你需要在XXL-JOB中设置定时任务在特定的时区执行你可能需要查看XXL-JOB的文档或者联系其支持人员看看它是否支持这种设定。
解决方案
将系统时间数据库时间容器时间设置统一。
MySQL数据库时区
show variables like %time_zone%;数据库JDBC URL上的设置时区 urljdbc:mysql://xxxx:3306/dddd?useUnicodetruecharacterEncodingutf8useSSLfalseserverTimezoneGMT%2B8容器内时间设置这一步要确保宿主机时间的正确
FROM openjdk:8-jre-slim
MAINTAINER xuxueliENV PARAMSENV TZPRC
RUN ln -snf /usr/share/zoneinfo/$TZ /etc/localtime echo $TZ /etc/timezoneADD target/xxl-job-admin-*.jar /app.jarENTRYPOINT [sh,-c,java -jar /app.jar $PARAMS]Java应用程序时区问题
启动jar包添加参数 -Duser.timezoneGMT08在Java应用程序中使用参数-Duser.timezoneGMT08启动jar包可以设置应用程序的默认时区为GMT08也就是东八区中国标准时间。
当Java应用程序需要进行日期和时间相关的操作时它会使用默认的时区来解释和表示日期和时间。默认情况下Java使用操作系统的默认时区作为应用程序的默认时区。但是通过使用-Duser.timezone参数可以覆盖默认的时区设置使应用程序在运行时使用指定的时区。
2 执行器注册和下线导致的问题IP问题
我们先来看执行器的配置
### 调度中心部署根地址 [选填]如调度中心集群部署存在多个地址则用逗号分隔。执行器将会使用该地址进行执行器心跳注册和任务结果回调为空则关闭自动注册
xxl.job.admin.addresseshttp://127.0.0.1:8080/xxl-job-admin
### 执行器通讯TOKEN [选填]非空时启用
xxl.job.accessToken
### 执行器AppName [选填]执行器心跳注册分组依据为空则关闭自动注册
xxl.job.executor.appnamexxl-job-executor-sample
### 执行器注册 [选填]优先使用该配置作为注册地址为空时使用内嵌服务 ”IP:PORT“ 作为注册地址。从而更灵活的支持容器类型执行器动态IP和动态映射端口问题。
xxl.job.executor.address
### 执行器IP [选填]默认为空表示自动获取IP多网卡时可手动设置指定IP该IP不会绑定Host仅作为通讯实用地址信息用于 执行器注册 和 调度中心请求并触发任务
xxl.job.executor.ip
### 执行器端口号 [选填]小于等于0则自动获取默认端口为9999单机部署多个执行器时注意要配置不同执行器端口
xxl.job.executor.port9999
### 执行器运行日志文件存储磁盘路径 [选填] 需要对该路径拥有读写权限为空则使用默认路径
xxl.job.executor.logpath/data/applogs/xxl-job/jobhandler
### 执行器日志文件保存天数 [选填] 过期日志自动清理, 限制值大于等于3时生效; 否则, 如-1, 关闭自动清理功能
xxl.job.executor.logretentiondays30这两行配置 xxl.job.executor.address执行器地址一般不需要手动配置系统会自动获取。如果手动配置那么优先级会比自动获取的方式更高。 xxl.job.executor.ip执行器IP地址可以手动配置如果不配置则系统会自动获取。
这两个配置项的值取决于你的具体需求和实际的服务器情况。如果你的执行器是在本地服务器上那么可能不需要填写这两个配置项系统会自动获取。但是如果你的执行器在远程服务器上那么你可能需要手动配置这两个项指定执行器的地址和IP。
需要注意的是通常情况下只需要配置xxl.job.executor.address即可因为XXL-JOB执行器会自动获取本机IP地址进行注册和通信。只有在特定需求下才需要同时配置两个地址。
一般容器环境出现问题核心的点就在这个配置上。
导致调度中心服务无法与定时任务执行器进行通信
容器环境的特殊性在容器环境中网络是隔离的。因此容器中的应用程序在默认情况下只能看到容器内部的网络环境。这就导致了一些问题比如 xxl.job.executor.address 和 xxl.job.executor.ip 配置问题。
无法预知容器的 IP 地址因为它是在容器启动时动态分配的。这就导致了如果在配置文件中硬编码了这些值可能会导致调度中心无法和执行器通信。
解决方案
将IP地址xxl.job.executor.ip或者注册地址xxl.job.executor.address尝试改为k8s服务名如果夸namespace的话服务名前带上namespace 如 nsname.service名称 以下只是示例实际根据你的的serveci命名命名替换 xxl.iob.admin.addresses http://xxl-job-admin-service:8080/xxl-iob-admin
xxl.job.executor.ipservice名称:端口
xxl.job.executor.addresshttp://service名称:port/
3 问题3 调度成功但是执行器的定时任务未执行
这个问题最诡异最坑还无法快速的定位问题。
我们再聊一下xxl-job执行器执行任务的原理 show you code 然后talk package com.xxl.job.core.thread;import com.xxl.job.core.biz.model.HandleCallbackParam;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.biz.model.TriggerParam;
import com.xxl.job.core.context.XxlJobContext;
import com.xxl.job.core.executor.XxlJobExecutor;
import com.xxl.job.core.handler.IJobHandler;
import com.xxl.job.core.log.XxlJobFileAppender;
import com.xxl.job.core.log.XxlJobLogger;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.Collections;
import java.util.Date;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;public class JobThread extends Thread {private static Logger logger LoggerFactory.getLogger(JobThread.class);private int jobId;private IJobHandler handler;private LinkedBlockingQueueTriggerParam triggerQueue;private SetLong triggerLogIdSet;private volatile boolean toStop false;private String stopReason;private boolean running false;private int idleTimes 0;public JobThread(int jobId, IJobHandler handler) {this.jobId jobId;this.handler handler;this.triggerQueue new LinkedBlockingQueue();this.triggerLogIdSet Collections.synchronizedSet(new HashSet());}public IJobHandler getHandler() {return this.handler;}public ReturnTString pushTriggerQueue(TriggerParam triggerParam) {if (this.triggerLogIdSet.contains(triggerParam.getLogId())) {logger.info( repeate trigger job, logId:{}, triggerParam.getLogId());return new ReturnT(500, repeate trigger job, logId: triggerParam.getLogId());} else {this.triggerLogIdSet.add(triggerParam.getLogId());this.triggerQueue.add(triggerParam);return ReturnT.SUCCESS;}}public void toStop(String stopReason) {this.toStop true;this.stopReason stopReason;}public boolean isRunningOrHasQueue() {return this.running || this.triggerQueue.size() 0;}public void run() {// 尝试初始化处理器try {this.handler.init();} catch (Throwable var26) {// 如果有错误记录错误日志logger.error(var26.getMessage(), var26);}final TriggerParam triggerParam;ReturnT executeResult;// 当线程没有被要求停止时while(!this.toStop) {this.running false;this.idleTimes;triggerParam null;executeResult null;boolean var16 false;ReturnT stopResult;label348: {try {var16 true;// 从触发队列中获取触发器参数如果在3秒内获取不到返回nulltriggerParam (TriggerParam)this.triggerQueue.poll(3L, TimeUnit.SECONDS);// 如果触发器参数不为空if (triggerParam ! null) {this.running true;this.idleTimes 0;// 从日志ID集合中移除触发器参数的日志IDthis.triggerLogIdSet.remove(triggerParam.getLogId());// 创建日志文件名称String logFileName XxlJobFileAppender.makeLogFileName(new Date(triggerParam.getLogDateTime()), triggerParam.getLogId());// 设置XxlJob上下文XxlJobContext.setXxlJobContext(new XxlJobContext(triggerParam.getLogId(), logFileName, triggerParam.getBroadcastIndex(), triggerParam.getBroadcastTotal()));// 记录任务启动日志XxlJobLogger.log(br----------- xxl-job job execute start -----------br----------- Param: triggerParam.getExecutorParams(), new Object[0]);// 如果执行超时时间大于0if (triggerParam.getExecutorTimeout() 0) {Thread futureThread null;try {// 创建一个新的任务FutureTaskReturnTString futureTask new FutureTask(new CallableReturnTString() {public ReturnTString call() throws Exception {// 执行处理器return JobThread.this.handler.execute(triggerParam.getExecutorParams());}});// 在新线程中执行任务futureThread new Thread(futureTask);futureThread.start();// 获取执行结果如果在指定的超时时间内没有结果抛出超时异常executeResult (ReturnT)futureTask.get((long)triggerParam.getExecutorTimeout(), TimeUnit.SECONDS);} catch (TimeoutException var24) {// 记录任务超时日志XxlJobLogger.log(br----------- xxl-job job execute timeout, new Object[0]);XxlJobLogger.log(var24);// 设置超时的执行结果executeResult new ReturnT(IJobHandler.FAIL_TIMEOUT.getCode(), job execute timeout );} finally {// 中断线程futureThread.interrupt();}} else {// 如果没有设置超时时间则直接执行处理器executeResult this.handler.execute(triggerParam.getExecutorParams());}// 如果执行结果为空if (executeResult null) {// 设置执行失败的结果executeResult IJobHandler.FAIL;} else {// 否则设置执行结果的消息和内容executeResult.setMsg(executeResult ! null executeResult.getMsg() ! null executeResult.getMsg().length() 50000 ? executeResult.getMsg().substring(0, 50000).concat(...) : executeResult.getMsg());executeResult.setContent((Object)null);}// 记录任务结束日志XxlJobLogger.log(br----------- xxl-job job execute end(finish) -----------br----------- ReturnT: executeResult, new Object[0]);var16 false;} else if (this.idleTimes 30) {// 如果连续30次都没有获取到触发器参数if (this.triggerQueue.size() 0) {// 如果触发队列为空则移除Job线程XxlJobExecutor.removeJobThread(this.jobId, executor idle times over limit.);var16 false;} else {var16 false;}} else {var16 false;}break label348;} catch (Throwable var27) {// 如果在执行过程中抛出异常if (this.toStop) {// 如果被要求停止记录停止原因XxlJobLogger.log(br----------- JobThread toStop, stopReason: this.stopReason, new Object[0]);}// 获取异常信息StringWriter stringWriter new StringWriter();var27.printStackTrace(new PrintWriter(stringWriter));String errorMsg stringWriter.toString();// 设置执行结果为异常信息executeResult new ReturnT(500, errorMsg);// 记录任务异常日志XxlJobLogger.log(br----------- JobThread Exception: errorMsg br----------- xxl-job job execute end(error) -----------, new Object[0]);var16 false;} finally {if (var16) {// 如果触发器参数不为空if (triggerParam ! null) {if (!this.toStop) {// 如果没有被要求停止向回调队列中推送回调参数TriggerCallbackThread.pushCallBack(new HandleCallbackParam(triggerParam.getLogId(), triggerParam.getLogDateTime(), executeResult));} else {// 如果被要求停止设置停止结果并向回调队列中推送回调参数ReturnTString stopResult new ReturnT(500, this.stopReason [job running, killed]);TriggerCallbackThread.pushCallBack(new HandleCallbackParam(triggerParam.getLogId(), triggerParam.getLogDateTime(), stopResult));}}}}// 如果触发器参数不为空if (triggerParam ! null) {if (!this.toStop) {// 如果没有被要求停止向回调队列中推送回调参数TriggerCallbackThread.pushCallBack(new HandleCallbackParam(triggerParam.getLogId(), triggerParam.getLogDateTime(), executeResult));} else {// 如果被要求停止设置停止结果并向回调队列中推送回调参数stopResult new ReturnT(500, this.stopReason [job running, killed]);TriggerCallbackThread.pushCallBack(new HandleCallbackParam(triggerParam.getLogId(), triggerParam.getLogDateTime(), stopResult));}}continue;}// 如果触发队列不为空且队列中还有元素while(this.triggerQueue ! null this.triggerQueue.size() 0) {// 从队列中获取触发器参数triggerParam (TriggerParam)this.triggerQueue.poll();// 如果触发器参数不为空if (triggerParam ! null) {// 设置未执行的结果并向回调队列中推送回调参数executeResult new ReturnT(500, this.stopReason [job not executed, in the job queue, killed.]);TriggerCallbackThread.pushCallBack(new HandleCallbackParam(triggerParam.getLogId(), triggerParam.getLogDateTime(), executeResult));}}// 尝试销毁处理器try {this.handler.destroy();} catch (Throwable var23) {// 如果有错误记录错误日志logger.error(var23.getMessage(), var23);}// 记录线程停止日志logger.info( xxl-job JobThread stoped, hashCode:{}, Thread.currentThread());}
}单从源码我们分析 JobThread 类源码的关键操作 pushTriggerQueue 方法该方法用于将任务触发器参数TriggerParam添加到任务队列triggerQueue中。如果队列中已经存在相同的日志ID那么这个新的任务触发器就不会被添加。否则将会添加到该队列中。 run 方法 开始一个持续运行的循环 在循环中尝试从触发队列中获取一个触发参数 (triggerParam)如果在3秒内没有获取到返回null。 如果获取到触发参数 将此线程标记为运行状态。从日志ID集合中移除该触发参数的日志ID。设置XxlJob的上下文。记录任务开始执行的日志。如果设置了执行超时时间将在新线程中执行处理器如果超过指定的超时时间仍未获取到结果将记录任务超时日志并设置超时的执行结果。如果没有设置超时时间直接执行处理器。根据执行结果是否为空设置相应的执行结果信息。记录任务结束日志。 如果连续30次都没有获取到触发参数且触发队列为空将移除该Job线程。 总结一下就是任务调度和任务执行依靠队列 triggerQueue这个是参数队列调度直接往这个队列里面put参数对象。执行依靠是JobThread run方法从队列中获取参数对象能获取就执行任务和线程是一一绑定的关系调度成功但没有结果回调。 所以核心的排查问题方法是我们要知道这个队列中还有多少个任务对象虽然已经被调度到任务队列但是没有被执行。 里面放到的对象为这个 所以我们只需要看下这个容器
jmap -histo 进程ID | grep “com.xxl.job.core.biz.model.TriggerParam”获取到有多少任务在等待。我们就能判断是不是卡在执行任务当中handler.execute();
这就是这个问题的排查思路。希望能帮助到大家。
4 问题4 数据库性能问题导致查询任务和操作日志数据卡主
表XXL_JOB_QRTZ_TRIGGER_LOG 约有 16.5 GB
执行
DELETE FROM XXL_JOB_QRTZ_TRIGGER_LOG WHERE trigger_time ‘2021-12-17 00:18:59’ AND trigger_time ‘2021-12-18 23:59:20’ 操作可能导致数据库 死锁或者CPU夯住了导致 0 时执行的任务没有执行成功。
它的主要功能是进行任务管理和调度。它在数据库中维护了一系列的日志和状态信息以确保任务能够正确地调度和执行。在某些情况下为了维护系统的性能和稳定性可能会需要清理一些旧的不再需要的数据。
DELETE FROM XXL_JOB_QRTZ_TRIGGER_LOG WHERE trigger_time ‘2021-12-17 00:18:59’ AND trigger_time ‘2021-12-18 23:59:20’这个操作是在XXL-JOB的任务调度日志表XXL_JOB_QRTZ_TRIGGER_LOG中删除特定时间范围内的日志记录。 为了进行日志清理以释放存储空间和保持数据库性能。大量的删除操作可能会对数据库性能产生影响尤其是在高并发或者大数据量的情况下。因此这种操作通常需要在数据库的低峰期进行并可能需要使用分批删除等技术以减少对数据库性能的影响。
解决方案
建议任务调度服务中的 XXL_JOB_QRTZ_TRIGGER_LOG 这张表保留最近一周的日志量在业务低峰期定时执行脚本。
5. 问题5 定时任务方法中有事务
事务的提交是当前方法结束 。由于任务执行时间很长导致后续任务到达执行时间无法执行甚至长事务导致其他服务性能问题引发连锁反应。
