文章目录
- 一、需求
- 1. 场景
- 2. 分析
- 3. 案例
- 二、什么是分⽚任务
- 2.1. 分⽚路由策略
- 2.2. 海量数据处理
- 2.3. 分片数量
- 2.4. 分片值颁发
- 2.5. 案例
- 三、解决思路
- 3.1. 数据拆分
- 3.2. 分片数量
- 3.3. 分⽚⽅式
- 3.4. 路由策略
- 3.5. 程序实战
一、需求
1. 场景
有⼀个任务需要处理100W条数据,每条数据的业务逻辑处理
要0.1s
2. 分析
对于普通任务来说,只有⼀个线程来处理 可能需要10万秒才能处理完,业务则严重受影响
3. 案例
双⼗⼀⼤促,给1000万⽤户发营销短信
二、什么是分⽚任务
2.1. 分⽚路由策略
执⾏器集群部署,如果任务的路由策略选择【分⽚⼴播】,⼀次任务调度将会【⼴播触发】对应集群中所有执⾏器执⾏⼀次任务,同时系统⾃动传递分⽚参数,执⾏器可根据分⽚参数开发分⽚任务。
2.2. 海量数据处理
需要处理的海量数据,以执⾏器为划分,每个执⾏器分配⼀定的任务数,并⾏执⾏。
2.3. 分片数量
XXL-Job⽀持动态扩容执⾏器集群,从⽽动态增加分⽚数量,到达更快处理任务。
2.4. 分片值颁发
分⽚的值是调度中⼼分配的
2.5. 案例
// 当前分⽚数,从0开始,即执⾏器的序号
int shardIndex = XxlJobHelper.getShardIndex();
//总分⽚数,执⾏器集群总机器数量
int shardTotal = XxlJobHelper.getShardTotal();
三、解决思路
3.1. 数据拆分
如果将100W数据均匀分给集群⾥的10台机器同时处理,每台机器耗时,1万秒即可,耗时会⼤⼤缩短,也能充分利⽤集群资源
3.2. 分片数量
在xxl-job⾥,可以配置执⾏器集群有10个机器,那么分⽚总
数是10,分⽚序号0~9 分别对应那10台机器。
3.3. 分⽚⽅式
id % 分⽚总数 余数是0 的,在第1个执⾏器上执⾏
id % 分⽚总数 余数是1 的,在第2个执⾏器上执⾏
id % 分⽚总数 余数是2 的,在第3个执⾏器上执⾏
...
id % 分⽚总数 余数是9 的,在第10个执⾏器上执⾏
3.4. 路由策略
选择为【分⽚⼴播】
3.5. 程序实战
- 基础案例
/**
* 2、分片广播任务
*/
@XxlJob("shardingJobHandler")
public void shardingJobHandler() throws Exception {
XxlJobHelper.log(" shardingJobHandler start");
logger.info("shardingJobHandler execute......");
// 分片参数
int shardIndex = XxlJobHelper.getShardIndex();
int shardTotal = XxlJobHelper.getShardTotal();
XxlJobHelper.log("分片参数:当前分片序号 = {}, 总分片数 = {}", shardIndex, shardTotal);
// 业务逻辑
for (int i = 0; i < shardTotal; i++) {
if (i == shardIndex) {
XxlJobHelper.log("第 {} 片, 命中分片开始处理", i);
} else {
XxlJobHelper.log("第 {} 片, 忽略", i);
}
}
XxlJobHelper.handleSuccess(" shardingJobHandler complete");
}
- 根据id进⾏分⽚取模(部署3个执⾏器)
/**
* 2、分⽚⼴播任务
*/
@XxlJob("shardingJobHandler")
public void shardingJobHandler() throws Exception {
XxlJobHelper.log(" shardingJobHandler start");
logger.info("shardingJobHandler execute......");
// 当前分⽚数,从0开始,即执⾏器的序号
int shardIndex = XxlJobHelper.getShardIndex();
//总分⽚数,执⾏器集群总机器数量
int shardTotal = XxlJobHelper.getShardTotal();
XxlJobHelper.log("分片参数:当前分片序号 = {}, 总分片数 = {}", shardIndex, shardTotal);
List<Integer> allUserIds = getAllUserIds();
allUserIds.forEach(obj -> {
if (obj % shardTotal == shardIndex) {
logger.info("第 {} ⽚, 命中分⽚开始处理⽤户id={}", shardIndex, obj);
}
});
}
private List<Integer> getAllUserIds() {
List<Integer> ids = new ArrayList<>();
for (int i = 0; i < 100; i++) {
ids.add(i);
}
return ids;
}