• XXL-Job分布式任务调度框架-单机模式和分片模式执行任务4

    一 调度模式分类

    1.1 调度模式

    1.单个任务：一个任务实例便可完成

    a)单机单任务：单机模式下任何路由模式都只有一个实例执行

    b)集群单任务：由路由策略(广播模式除外)选择其中一个实例完成

    2.集群部署：每个实例都同时执行一部分数据。分片方式：取模分片，范围分片。现使用xxljob进行分片任务执行，有两种解决思路。

    a)单机多任务分片：单机模式下，创建同类型任务多个任务计划，手工分片数据作为参数。

     1.自定义业务规则，配置多个xxl任务，来实现分片功能。

     2.每个任务指定不同的参数,但使用相同的jobhanlder:

    b)集群任务分片：只有广播模式会通知所有实例都会运行，每个节点取模执行任务

    注意：单机模式：只启动一个任务执行器实例，修改路由模式即便是广播模式依然只有一个实例运行job

    集群模式：启动多个任务实例，这里可以把每个实例的端口号都改为不同，可以看到同一个任务类有多个机器。除了广播模式，其他模式都只会选择机器列表中的一个执行job。

    XXL-JOB任务分片 - 灰信网（软件开发博客聚合）

    3.流程截图如下：

    1.2 案例说明

    1.2.1 需求说明

    假设有5个地市，每个地市有10个订单执行，总共50个订单，而每个订单中又有一个字段体现出地市信息。

    XXL-JOB任务分片_xxljob分片-CSDN博客

    1.3 同一实例并行运行的设置

    1.同一个实例，通过修改服务端口，并行启动的设置。

    二 案例模式1单机模式

    2.1 思路

     单机单任务：单机模式下任何路由模式都只有一个实例执行

    2.1 编写代码

    1. package com.ljf.xxl.job.task;
    2.  
    3. import com.xxl.job.core.biz.model.ReturnT;
    4. import com.xxl.job.core.context.XxlJobHelper;
    5. import com.xxl.job.core.handler.annotation.XxlJob;
    6. import org.slf4j.Logger;
    7. import org.slf4j.LoggerFactory;
    8. import org.springframework.beans.factory.annotation.Autowired;
    9. import org.springframework.stereotype.Component;
    10. import org.springframework.util.StringUtils;
    11.  
    12. import java.util.*;
    13.  
    14. import static com.xxl.job.core.biz.model.ReturnT.FAIL_CODE;
    15.  
    16. /**
    17.  * @ClassName: SingleTask
    18.  * @Description: TODO
    19.  * @Author: admin
    20.  * @Date: 2023/10/13 20:52:26 
    21.  * @Version: V1.0
    22.  **/
    23. @Component
    24. public class SingleTask {
    25.     private Logger log = LoggerFactory.getLogger( SingleTask.class);
    26.     @XxlJob(value = "singleTasks", init = "init", destroy = "destroy")
    27.     public ReturnT<String> execute(String cities) {
    28.         String param = XxlJobHelper.getJobParam();
    29.         System.out.println("v2 param=" + param);
    30.         XxlJobHelper.log("v2 param=" + param);
    31.         System.out.println("获取cities 参数:==================="+cities);
    32.         if (StringUtils.isEmpty(param)) {
    33.             return new ReturnT(FAIL_CODE, "latnIds不能为空");
    34.         }
    35.         //处理
    36.         if (param != null && param.length() > 0) {
    37.             String cityArray[] = param.split(",");
    38.             for (int k = 0; k < cityArray.length; k++) {
    39.               //  Integer cityId = Integer.parseInt(cityArray[k]);
    40.                String cityId= cityArray[k];
    41.                 List<String> infoList = dbMap.get(cityId);
    42.                 infoList.stream().forEach((x) -> {
    43.                   // System.out.println("【"+Thread.currentThread().getName()+"】执行【{"+cityId+"}】，任务内容为：{"+x+"}" );
    44.                     log.info("【"+Thread.currentThread().getName()+"】执行【{"+cityId+"}】，任务内容为：{"+x+"}");
    45.                   // XxlJobHelper.log("【"+Thread.currentThread().getName()+"】执行【{"+cityId+"}】，任务内容为：{"+x+"}");
    46.                 });
    47.             }
    48.         }
    49.         return ReturnT.SUCCESS;
    50.     }
    51.     private void init() {
    52.  
    53.         log.info("init 方法调用成功");
    54.  
    55.     }
    56.  
    57.     private void destroy() {
    58.         log.info("destroy 方法调用成功");
    59.     }
    60.     static List<String> cityNoList=null;
    61.     static int cityTaskNum=0;
    62.     //数据库
    63.     static Map<String,List<String>> dbMap=null;
    64.     static {
    65.         dbMap=new HashMap<>();
    66.         //城市编号
    67.         cityNoList= Arrays.asList("010","0755","0371","0373","0375");
    68.         //任务数
    69.         cityTaskNum=10;
    70.         for(int k=0;k<cityNoList.size();k++){
    71.             List<String> tasksList = new ArrayList<>();
    72.             for(int m=1;m<=cityTaskNum;m++){
    73.                 tasksList.add("执行任务"+m);
    74.             }
    75.             dbMap.put(cityNoList.get(k)+"",tasksList);
    76.         }
    77.     }
    78.  
    79. }

    2.2 配置单机模式

    2.2.1 配置执行器

    执行器设置为：xxl-single-dingsi

    2.2.2 配置调度任务

    1.选择任务管理

    2.进行任务的配置

    2.3 启动服务查看效果

    2.3.1 程序配置启动

    server.port=8083;  job.port=9993 ；执行器设置为：xxl-single-dingsi；(填写系统页面配置的执行器名称)，然后进行启动

    2.3.2 查看log日志

    所有的任务都指在一个任务实例中执行。

    三 案例模式2单机多分片任务

    3.1 思路

    单机多任务分片：单机模式下，创建同类型任务多个任务计划，手工分片数据作为参数。

     1.自定义业务规则，配置多个xxl任务，来实现分片功能。

     2.每个任务指定不同的参数,但使用相同的jobhanlder:

    3.2 调度配置规则

     1.自定义业务规则，配置多个xxl任务，来实现分片功能。这里配置两个任务，使用相同的执行器【单机多任务分片】，两个job任务使用相同 jobhandler： singleTasks

    2.任务id为6的配置：路由策略手动分配为：第一个；jobhandler为：singleTasks；配置参数为：0371,0373,0375

    3.任务id为8的配置： 路由策略手动分配为：最后一个；jobhandler为：singleTasks；配置参数为：010,0755

     3.2 查看效果

    分别启动两个任务，查看执行日志：

    3.2.1 端口8084任务1

    1.在idea中服务端口设置：server.port=8084;  job.port=9994 ；执行器设置为：xxl-single-dingsi，(填写系统页面配置的执行器名称)，然后进行启动

    2.执行结果如下： 分配到的任务是 【010，0755】

     3.2.2 端口8083任务2

    1.在idea中服务端口设置：server.port=8083;  job.port=9993 ；执行器设置为：xxl-single-dingsi，(填写系统页面配置的执行器名称)，然后进行启动

    2.执行结果如下： 分配到的任务是 【0371，0373，0375】 

     3.2.3 查看执行器的执行实例

    四 案例模式3多机分片任务 

    4.1 思路

    采用多机器取模的方式，来为不同的机器指定各自服务的城市列表。

    4.2 编写多机分片代码

    1.编写代码

    1. package com.ljf.xxl.job.task;
    2.  
    3. import com.xxl.job.core.biz.model.ReturnT;
    4. import com.xxl.job.core.context.XxlJobHelper;
    5. import com.xxl.job.core.handler.annotation.XxlJob;
    6. import org.slf4j.Logger;
    7. import org.slf4j.LoggerFactory;
    8. import org.springframework.stereotype.Component;
    9. import org.springframework.util.StringUtils;
    10.  
    11. import java.util.*;
    12.  
    13. import static com.xxl.job.core.biz.model.ReturnT.FAIL_CODE;
    14.  
    15. /**
    16.  * @ClassName: BatchTask
    17.  * @Description: TODO
    18.  * @Author: admin
    19.  * @Date: 2023/10/14 11:09:10 
    20.  * @Version: V1.0
    21.  **/
    22. @Component
    23. public class BatchTask {
    24.     private Logger log = LoggerFactory.getLogger(BatchTask.class);
    25.     @XxlJob(value = "batchTasks", init = "init", destroy = "destroy")
    26.     public ReturnT<String> execute(String cities) {
    27.         String param = XxlJobHelper.getJobParam();
    28.         System.out.println("进来了param:"+param);
    29.         //当前的执行器编号
    30.         int shardIndex = XxlJobHelper.getShardIndex();
    31.         //总的分片数，就是执行器的集群数量
    32.         int shardTotal = XxlJobHelper.getShardTotal();
    33.         log.info("分片总数:{},当前分片数{}",shardTotal,shardIndex);
    34.         //处理
    35.         for(int m=0;m<cityNoList.size();m++){
    36.             if(m%shardTotal==shardIndex){
    37.                String  cityId=cityNoList.get(m);
    38.                 List<String> infoList = dbMap.get(cityId);
    39.                 final int n=m;
    40.                 infoList.stream().forEach((x) -> {
    41.                     log.info("任务索引编号"+n+"【"+Thread.currentThread().getName()+"】执行【{"+cityId+"}】，任务内容为：{"+x+"}");
    42.                 });
    43.             }
    44.             else{
    45.                 log.info("========不在本分片执行的任务序号m:{},当前分片索引:{}",m,shardIndex);
    46.             }
    47.  
    48.         }
    49.         return ReturnT.SUCCESS;
    50.     }
    51.     private void init() {
    52.  
    53.         log.info("init 方法调用成功");
    54.  
    55.     }
    56.  
    57.     private void destroy() {
    58.         log.info("destroy 方法调用成功");
    59.     }
    60.     static List<String> cityNoList=null;
    61.     static int cityTaskNum=0;
    62.     //数据库
    63.     static Map<String,List<String>> dbMap=null;
    64.     static {
    65.         dbMap=new HashMap<>();
    66.         //城市编号
    67.         cityNoList= Arrays.asList("010","0755","0371","0373","0375");
    68.         //任务数
    69.         cityTaskNum=10;
    70.         for(int k=0;k<cityNoList.size();k++){
    71.             List<String> tasksList = new ArrayList<>();
    72.             for(int m=1;m<=cityTaskNum;m++){
    73.                 tasksList.add("执行任务"+m);
    74.             }
    75.             dbMap.put(cityNoList.get(k)+"",tasksList);
    76.         }
    77.     }
    78. }

    4.3 配置执行器

    4.4 配置调度任务

    1.这里的路由模式为： 分片广播，jobhhandler为：BatchTask

    如图

     4.5 启动实例任务

    4.5.1 分片0号端口8085任务1

    1.在idea中服务端口设置：server.port=8085;  job.port=9995 ；执行器设置为：xxl-batch-dingsi

    ，(填写系统页面配置的执行器名称)，然后进行启动

    2.执行结果如下： 0号分片分配到的任务是：编号0，编号3，不满足是：编号1，编号2，编号4；

    4.5.2 分片1端口8086任务2

    1.在idea中服务端口设置：server.port=8086;  job.port=9996 ；执行器设置为：xxl-batch-dingsi

    ，(填写系统页面配置的执行器名称)，然后进行启动

    2.执行结果如下： 1号分片分配到的任务是：编号1，编号4，不满足是：编号0，编号2，编号3；

    4.5.3 分片2端口8087任务3

    1.在idea中服务端口设置：server.port=8086;  job.port=9996 ；执行器设置为：xxl-batch-dingsi

    ，(填写系统页面配置的执行器名称)，然后进行启动

    2.执行结果如下： 2号分片分配到的任务是：编号2，不满足是：编号0，编号1，编号3，编号4；

    4.5.4 任务启动先后对分片数影响

    实例1，实例2，实例3 这3个不同的任务实例，陆续启动，可以通过log日志看到，分片总数在动态变化，实现动态监控加载。

    初始监控到2个分片

    后续等3个都启动起来后，监控到3个分片 ：

  • 相关阅读:
    基于Python实现的吃豆人游戏设计
    【ARM 嵌入式 C 入门及渐进 10 -- 冒泡排序 选择排序 插入排序 快速排序 归并排序 堆排序 比较介绍】
    linux学习笔记
    ffmpeg跨平台arm编译-ubuntu
    verilog之wire vs reg区别
    Jenkins安装和使用
    No1.详解【2023年全国大学生数学建模竞赛】C题——蔬菜类商品的自动定价与补货决策（代码 + 详细输出 + 数据集&代码 下载）
    【Lua基础 第5章】 unpack()和pack()、Lua 中的文件 I/O、简单模式下io的部分方法、完全模式下file的部分方法、日期和时间、闭包使用
    codesys【按钮】
    【Web】前端框架对微软老旧浏览器的支持
  • 原文地址：https://blog.csdn.net/u011066470/article/details/133820792