Skip to content

X
XXL-JOB
提交 0c03f634 authored 作者: xuxueli's avatar xuxueli
浏览文件
操作

广播分片任务

上级 f5f98bc0
隐藏空白字符变更
内嵌 并排
正在显示 包含 260 行增加 和 31 行删除
doc/XXL-JOB官方文档.md
浏览文件 @ 0c03f634
......@@ -306,7 +306,7 @@ XXL-JOB是一个轻量级分布式任务调度框架，其核心设计目标是
#### 步骤一：执行器项目中，开发JobHandler：
- 1、 新建一个继承com.xxl.job.core.handler.IJobHandler的Java类；
- 2、 该类被Spring容器扫描为Bean实例，如加“@Service注解”；
- 2、 该类被Spring容器扫描为Bean实例，如加“@Component”注解；
- 3、 添加 “@JobHander(value="自定义jobhandler名称")”注解，注解的value值为自定义的JobHandler名称，该名称对应的是调度中心新建任务的JobHandler属性的值。
（可参考xxl-job-executor-example项目中的DemoJobHandler，见下图）
......@@ -889,6 +889,7 @@ Tips: 历史版本(V1.3.x)目前已经Release至稳定版本, 进入维护阶段
#### 6.18 版本 V1.8.1 特性[快照版本]
- 1、任务分片：一个任务被拆分成N个独立的任务单元，然后由分布式部署的执行器分别执行某一个或几个分片单元；
- 2、执行器JobHandler禁止命名冲突；
#### TODO LIST
- 1、任务权限管理：执行器为粒度分配权限，核心操作校验权限；
......
xxl-job-admin/src/main/java/com/xxl/job/admin/core/route/ExecutorRouteStrategyEnum.java
浏览文件 @ 0c03f634
......@@ -15,7 +15,8 @@ public enum ExecutorRouteStrategyEnum {
LEAST_FREQUENTLY_USED("最不经常使用", new ExecutorRouteLFU()),
LEAST_RECENTLY_USED("最近最久未使用", new ExecutorRouteLRU()),
FAILOVER("故障转移", new ExecutorRouteFailover()),
BUSYOVER("忙碌转移", new ExecutorRouteBusyover());
BUSYOVER("忙碌转移", new ExecutorRouteBusyover()),
BROADCAST("广播", null);
ExecutorRouteStrategyEnum(String title, ExecutorRouter router) {
this.title = title;
......
xxl-job-admin/src/main/java/com/xxl/job/admin/core/route/ExecutorRouter.java
浏览文件 @ 0c03f634
......@@ -28,7 +28,7 @@ public abstract class ExecutorRouter {
* run executor
* @param triggerParam
* @param address
* @return
* @return ReturnT.content: final address
*/
public static ReturnT<String> runExecutor(TriggerParam triggerParam, String address){
ReturnT<String> runResult = null;
......@@ -46,6 +46,7 @@ public abstract class ExecutorRouter {
runResultSB.append("<br>msg：").append(runResult.getMsg());
runResult.setMsg(runResultSB.toString());
runResult.setContent(address);
return runResult;
}
......
xxl-job-admin/src/main/java/com/xxl/job/admin/core/trigger/XxlJobTrigger.java
浏览文件 @ 0c03f634
......@@ -5,6 +5,7 @@ import com.xxl.job.admin.core.model.XxlJobGroup;
import com.xxl.job.admin.core.model.XxlJobInfo;
import com.xxl.job.admin.core.model.XxlJobLog;
import com.xxl.job.admin.core.route.ExecutorRouteStrategyEnum;
import com.xxl.job.admin.core.route.ExecutorRouter;
import com.xxl.job.admin.core.schedule.XxlJobDynamicScheduler;
import com.xxl.job.admin.core.thread.JobFailMonitorHelper;
import com.xxl.job.core.biz.model.ReturnT;
......@@ -40,6 +41,79 @@ public class XxlJobTrigger {
ExecutorRouteStrategyEnum executorRouteStrategyEnum = ExecutorRouteStrategyEnum.match(jobInfo.getExecutorRouteStrategy(), null); // route strategy
ArrayList<String> addressList = (ArrayList<String>) group.getRegistryList();
// broadcast
if (ExecutorRouteStrategyEnum.BROADCAST == executorRouteStrategyEnum && CollectionUtils.isNotEmpty(addressList)) {
for (int i = 0; i < addressList.size(); i++) {
String address = addressList.get(i);
// 1、save log-id
XxlJobLog jobLog = new XxlJobLog();
jobLog.setJobGroup(jobInfo.getJobGroup());
jobLog.setJobId(jobInfo.getId());
XxlJobDynamicScheduler.xxlJobLogDao.save(jobLog);
logger.debug(">>>>>>>>>>> xxl-job trigger start, jobId:{}", jobLog.getId());
// 2、prepare trigger-info
//jobLog.setExecutorAddress(executorAddress);
jobLog.setGlueType(jobInfo.getGlueType());
jobLog.setExecutorHandler(jobInfo.getExecutorHandler());
jobLog.setExecutorParam(jobInfo.getExecutorParam());
jobLog.setTriggerTime(new Date());
ReturnT<String> triggerResult = new ReturnT<String>(null);
StringBuffer triggerMsgSb = new StringBuffer();
triggerMsgSb.append("注册方式：").append( (group.getAddressType() == 0)?"自动注册":"手动录入" );
triggerMsgSb.append("<br>阻塞处理策略：").append(blockStrategy.getTitle());
triggerMsgSb.append("<br>失败处理策略：").append(failStrategy.getTitle());
triggerMsgSb.append("<br>地址列表：").append(group.getRegistryList());
triggerMsgSb.append("<br>路由策略：").append(executorRouteStrategyEnum.getTitle()).append("("+i+"/"+addressList.size()+")"); // update01
// 3、trigger-valid
if (triggerResult.getCode()==ReturnT.SUCCESS_CODE && CollectionUtils.isEmpty(addressList)) {
triggerResult.setCode(ReturnT.FAIL_CODE);
triggerMsgSb.append("<br>----------------------<br>").append("调度失败：").append("执行器地址为空");
}
if (triggerResult.getCode() == ReturnT.SUCCESS_CODE) {
// 4.1、trigger-param
TriggerParam triggerParam = new TriggerParam();
triggerParam.setJobId(jobInfo.getId());
triggerParam.setExecutorHandler(jobInfo.getExecutorHandler());
triggerParam.setExecutorParams(jobInfo.getExecutorParam());
triggerParam.setExecutorBlockStrategy(jobInfo.getExecutorBlockStrategy());
triggerParam.setLogId(jobLog.getId());
triggerParam.setLogDateTim(jobLog.getTriggerTime().getTime());
triggerParam.setGlueType(jobInfo.getGlueType());
triggerParam.setGlueSource(jobInfo.getGlueSource());
triggerParam.setGlueUpdatetime(jobInfo.getGlueUpdatetime().getTime());
triggerParam.setBroadcastIndex(i);
triggerParam.setBroadcastTotal(addressList.size()); // update02
// 4.2、trigger-run (route run / trigger remote executor)
triggerResult = ExecutorRouter.runExecutor(triggerParam, address); // update03
triggerMsgSb.append("<br><br><span style=\"color:#00c0ef;\" > >>>>>>>>>>>触发调度<<<<<<<<<<< </span><br>").append(triggerResult.getMsg());
// 4.3、trigger (fail retry)
if (triggerResult.getCode()!=ReturnT.SUCCESS_CODE && failStrategy == ExecutorFailStrategyEnum.FAIL_RETRY) {
triggerResult = ExecutorRouter.runExecutor(triggerParam, address); // update04
triggerMsgSb.append("<br><br><span style=\"color:#F39C12;\" > >>>>>>>>>>>失败重试<<<<<<<<<<< </span><br>").append(triggerResult.getMsg());
}
}
// 5、save trigger-info
jobLog.setExecutorAddress(triggerResult.getContent());
jobLog.setTriggerCode(triggerResult.getCode());
jobLog.setTriggerMsg(triggerMsgSb.toString());
XxlJobDynamicScheduler.xxlJobLogDao.updateTriggerInfo(jobLog);
// 6、monitor triger
JobFailMonitorHelper.monitor(jobLog.getId());
logger.debug(">>>>>>>>>>> xxl-job trigger end, jobId:{}", jobLog.getId());
}
return;
}
// 1、save log-id
XxlJobLog jobLog = new XxlJobLog();
jobLog.setJobGroup(jobInfo.getJobGroup());
......@@ -67,10 +141,6 @@ public class XxlJobTrigger {
triggerResult.setCode(ReturnT.FAIL_CODE);
triggerMsgSb.append("<br>----------------------<br>").append("调度失败：").append("执行器地址为空");
}
if (triggerResult.getCode() == ReturnT.SUCCESS_CODE && executorRouteStrategyEnum == null) {
triggerResult.setCode(ReturnT.FAIL_CODE);
triggerMsgSb.append("<br>----------------------<br>").append("调度失败：").append("执行器路由策略为空");
}
if (triggerResult.getCode() == ReturnT.SUCCESS_CODE) {
// 4.1、trigger-param
......@@ -79,11 +149,13 @@ public class XxlJobTrigger {
triggerParam.setExecutorHandler(jobInfo.getExecutorHandler());
triggerParam.setExecutorParams(jobInfo.getExecutorParam());
triggerParam.setExecutorBlockStrategy(jobInfo.getExecutorBlockStrategy());
triggerParam.setLogId(jobLog.getId());
triggerParam.setLogDateTim(jobLog.getTriggerTime().getTime());
triggerParam.setGlueType(jobInfo.getGlueType());
triggerParam.setGlueSource(jobInfo.getGlueSource());
triggerParam.setGlueUpdatetime(jobInfo.getGlueUpdatetime().getTime());
triggerParam.setLogId(jobLog.getId());
triggerParam.setLogDateTim(jobLog.getTriggerTime().getTime());
triggerParam.setBroadcastIndex(0);
triggerParam.setBroadcastTotal(1);
// 4.2、trigger-run (route run / trigger remote executor)
triggerResult = executorRouteStrategyEnum.getRouter().routeRun(triggerParam, addressList);
......
xxl-job-core/src/main/java/com/xxl/job/core/biz/impl/ExecutorBizImpl.java
浏览文件 @ 0c03f634
......@@ -76,10 +76,13 @@ public class ExecutorBizImpl implements ExecutorBiz {
// valid：jobHandler + jobThread
if (GlueTypeEnum.BEAN==GlueTypeEnum.match(triggerParam.getGlueType())) {
// new jobhandler
IJobHandler newJobHandler = XxlJobExecutor.loadJobHandler(triggerParam.getExecutorHandler());
// valid old jobThread
if (jobThread != null && jobHandler!=null && jobThread.getHandler() != jobHandler) {
if (jobThread!=null && jobHandler != newJobHandler) {
// change handler, need kill old thread
removeOldReason = "更新JobHandler或更换任务模式,终止旧任务线程";
removeOldReason = "更换JobHandler或更换任务模式,终止旧任务线程";
jobThread = null;
jobHandler = null;
......@@ -87,7 +90,7 @@ public class ExecutorBizImpl implements ExecutorBiz {
// valid handler
if (jobHandler == null) {
jobHandler = XxlJobExecutor.loadJobHandler(triggerParam.getExecutorHandler());
jobHandler = newJobHandler;
if (jobHandler == null) {
return new ReturnT<String>(ReturnT.FAIL_CODE, "job handler [" + triggerParam.getExecutorHandler() + "] not found.");
}
......
xxl-job-core/src/main/java/com/xxl/job/core/biz/model/TriggerParam.java
浏览文件 @ 0c03f634
......@@ -14,12 +14,15 @@ public class TriggerParam implements Serializable{
private String executorParams;
private String executorBlockStrategy;
private int logId;
private long logDateTim;
private String glueType;
private String glueSource;
private long glueUpdatetime;
private int logId;
private long logDateTim;
private int broadcastIndex;
private int broadcastTotal;
public int getJobId() {
return jobId;
......@@ -53,6 +56,22 @@ public class TriggerParam implements Serializable{
this.executorBlockStrategy = executorBlockStrategy;
}
public int getLogId() {
return logId;
}
public void setLogId(int logId) {
this.logId = logId;
}
public long getLogDateTim() {
return logDateTim;
}
public void setLogDateTim(long logDateTim) {
this.logDateTim = logDateTim;
}
public String getGlueType() {
return glueType;
}
......@@ -77,20 +96,20 @@ public class TriggerParam implements Serializable{
this.glueUpdatetime = glueUpdatetime;
}
public int getLogId() {
return logId;
public int getBroadcastIndex() {
return broadcastIndex;
}
public void setLogId(int logId) {
this.logId = logId;
public void setBroadcastIndex(int broadcastIndex) {
this.broadcastIndex = broadcastIndex;
}
public long getLogDateTim() {
return logDateTim;
public int getBroadcastTotal() {
return broadcastTotal;
}
public void setLogDateTim(long logDateTim) {
this.logDateTim = logDateTim;
public void setBroadcastTotal(int broadcastTotal) {
this.broadcastTotal = broadcastTotal;
}
@Override
......@@ -100,11 +119,14 @@ public class TriggerParam implements Serializable{
", executorHandler='" + executorHandler + '\'' +
", executorParams='" + executorParams + '\'' +
", executorBlockStrategy='" + executorBlockStrategy + '\'' +
", logId=" + logId +
", logDateTim=" + logDateTim +
", glueType='" + glueType + '\'' +
", glueSource='" + glueSource + '\'' +
", glueUpdatetime=" + glueUpdatetime +
", logId=" + logId +
", logDateTim=" + logDateTim +
", broadcastIndex=" + broadcastIndex +
", broadcastTotal=" + broadcastTotal +
'}';
}
}
xxl-job-core/src/main/java/com/xxl/job/core/executor/XxlJobExecutor.java
浏览文件 @ 0c03f634
......@@ -98,6 +98,9 @@ public class XxlJobExecutor implements ApplicationContextAware, ApplicationListe
if (serviceBean instanceof IJobHandler){
String name = serviceBean.getClass().getAnnotation(JobHander.class).value();
IJobHandler handler = (IJobHandler) serviceBean;
if (loadJobHandler(name) != null) {
throw new RuntimeException("xxl-job jobhandler naming conflicts.");
}
registJobHandler(name, handler);
}
}
......
xxl-job-core/src/main/java/com/xxl/job/core/log/XxlJobFileAppender.java
浏览文件 @ 0c03f634
......@@ -18,8 +18,8 @@ public class XxlJobFileAppender {
// for JobThread (support log for child thread of job handler)
//public static ThreadLocal<String> contextHolder = new ThreadLocal<String>();
public static InheritableThreadLocal<String> contextHolder = new InheritableThreadLocal<String>();
public static SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
public static final InheritableThreadLocal<String> contextHolder = new InheritableThreadLocal<String>();
public static final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
/**
* log filename: yyyy-MM-dd/9999.log
......
xxl-job-core/src/main/java/com/xxl/job/core/thread/JobThread.java
浏览文件 @ 0c03f634
......@@ -7,6 +7,7 @@ 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 com.xxl.job.core.util.ShardingUtil;
import org.eclipse.jetty.util.ConcurrentHashSet;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
......@@ -112,6 +113,7 @@ public class JobThread extends Thread{
String logFileName = XxlJobFileAppender.makeLogFileName(new Date(triggerParam.getLogDateTim()), triggerParam.getLogId());
XxlJobFileAppender.contextHolder.set(logFileName);
ShardingUtil.setShardingVo(new ShardingUtil.ShardingVO(triggerParam.getBroadcastIndex(), triggerParam.getBroadcastTotal()));
XxlJobLogger.log("<br>----------- xxl-job job execute start -----------<br>----------- Params:" + Arrays.toString(handlerParams));
executeResult = handler.execute(handlerParams);
......
xxl-job-core/src/main/java/com/xxl/job/core/util/ShardingUtil.java 0 → 100644
浏览文件 @ 0c03f634
package com.xxl.job.core.util;
/**
* sharding vo
* @author xuxueli 2017-07-25 21:26:38
*/
public class ShardingUtil {
private static InheritableThreadLocal<ShardingVO> contextHolder = new InheritableThreadLocal<ShardingVO>();
public static class ShardingVO {
private int index; // sharding index
private int total; // sharding total
public ShardingVO(int index, int total) {
this.index = index;
this.total = total;
}
public int getIndex() {
return index;
}
public void setIndex(int index) {
this.index = index;
}
public int getTotal() {
return total;
}
public void setTotal(int total) {
this.total = total;
}
}
public static void setShardingVo(ShardingVO shardingVo){
contextHolder.set(shardingVo);
}
public static ShardingVO getShardingVo(){
return contextHolder.get();
}
}
xxl-job-executor-example/src/main/java/com/xxl/job/executor/service/jobhandler/DemoJobHandler.java
浏览文件 @ 0c03f634
......@@ -4,7 +4,7 @@ import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.handler.IJobHandler;
import com.xxl.job.core.handler.annotation.JobHander;
import com.xxl.job.core.log.XxlJobLogger;
import org.springframework.stereotype.Service;
import org.springframework.stereotype.Component;
import java.util.concurrent.TimeUnit;
......@@ -13,15 +13,15 @@ import java.util.concurrent.TimeUnit;
* 任务Handler的一个Demo（Bean模式）
*
* 开发步骤：
* 1、继承 “IJobHandler” ；
* 2、装配到Spring，例如加 “@Service” 注解；
* 3、加 “@JobHander” 注解，注解value值为新增任务生成的JobKey的值;多个JobKey用逗号分割;
* 1、新建一个继承com.xxl.job.core.handler.IJobHandler的Java类；
* 2、该类被Spring容器扫描为Bean实例，如加“@Component”注解；
* 3、添加 “@JobHander(value="自定义jobhandler名称")”注解，注解的value值为自定义的JobHandler名称，该名称对应的是调度中心新建任务的JobHandler属性的值。
* 4、执行日志：需要通过 "XxlJobLogger.log" 打印执行日志；
*
* @author xuxueli 2015-12-19 19:43:36
*/
@JobHander(value="demoJobHandler")
@Service
@Component
public class DemoJobHandler extends IJobHandler {
@Override
......
xxl-job-executor-example/src/main/java/com/xxl/job/executor/service/jobhandler/ShardingJobHandler.java 0 → 100644
浏览文件 @ 0c03f634
package com.xxl.job.executor.service.jobhandler;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.handler.IJobHandler;
import com.xxl.job.core.handler.annotation.JobHander;
import com.xxl.job.core.log.XxlJobLogger;
import com.xxl.job.core.util.ShardingUtil;
import org.springframework.stereotype.Service;
/**
* 广播分片任务
*
* @author xuxueli 2017-07-25 20:56:50
*/
@JobHander(value="shardingJobHandler")
@Service
public class ShardingJobHandler extends IJobHandler {
@Override
public ReturnT<String> execute(String... params) throws Exception {
// 分片参数
ShardingUtil.ShardingVO shardingVO = ShardingUtil.getShardingVo();
XxlJobLogger.log("分片参数：当前分片序号 = {0}, 总分片数 = {1}", shardingVO.getIndex(), shardingVO.getTotal());
// 业务逻辑
for (int i = 0; i < shardingVO.getTotal(); i++) {
if (i == shardingVO.getIndex()) {
XxlJobLogger.log("第 {0} 片, 命中分片开始处理", i);
} else {
XxlJobLogger.log("第 {0} 片, 忽略", i);
}
}
return ReturnT.SUCCESS;
}
}
xxl-job-executor-springboot-example/src/main/java/com/xxl/job/executor/service/jobhandler/ShardingJobHandler.java 0 → 100644
浏览文件 @ 0c03f634
package com.xxl.job.executor.service.jobhandler;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.handler.IJobHandler;
import com.xxl.job.core.handler.annotation.JobHander;
import com.xxl.job.core.log.XxlJobLogger;
import com.xxl.job.core.util.ShardingUtil;
import org.springframework.stereotype.Service;
/**
* 广播分片任务
*
* @author xuxueli 2017-07-25 20:56:50
*/
@JobHander(value="shardingJobHandler")
@Service
public class ShardingJobHandler extends IJobHandler {
@Override
public ReturnT<String> execute(String... params) throws Exception {
// 分片参数
ShardingUtil.ShardingVO shardingVO = ShardingUtil.getShardingVo();
XxlJobLogger.log("分片参数：当前分片序号 = {0}, 总分片数 = {1}", shardingVO.getIndex(), shardingVO.getTotal());
// 业务逻辑
for (int i = 0; i < shardingVO.getTotal(); i++) {
if (i == shardingVO.getIndex()) {
XxlJobLogger.log("第 {0} 片, 命中分片开始处理", i);
} else {
XxlJobLogger.log("第 {0} 片, 忽略", i);
}
}
return ReturnT.SUCCESS;
}
}
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论