• 分布式任务调度平台 xxl-job 搭建

      1.简介               

            XXL-JOB是一个分布式任务调度平台，它有两个核心模块，一个调度中心，另外一个为执行器，它把任务调度和任务执行分成两个部分。这样调度只需要负责任务的调度属性，触发调度信号。执行模块只需要接收调度信号，去执行具体的业务逻辑，两者可以各自的进行扩容和缩容。

    架构图：

    2. 搭建

     因为是开源项目，所以我们直接去github下载源码，我选择的是release版本 

     源码地址：github.com/xuxueli/xxl-job

    下载下来后的代码结构： 

     

     可以看到 xxl-job-master 为 父工程

                    xxl-job-admin  为 调度中心

                    xxl-job-core     为 基础配置，为调度中心提供支持 

                    xxl-job-executor-samples   为  不同版本的执行器  本文使用springboot执行器

    构建数据库表：

    代码build后，先去数据库把db做好，官网代码里提供了相应sql，直接执行

     修改调度中心db配置：

     然后调度中心可以直接启动了，端口初始是8080，为了防止占用，建议改成别的

    访问url: localhost:8080/xxl-job-admin

     

     搭建执行器：

            这里我们用开源代码给的demo做，后边整合到外部的springboot服务，父工程内的执行器moudle代码结构，分为三部分：job配置中心，暴露的调度接口和要执行的逻辑接口

     1.xxl-job配置

    1. xxl:
    2.   job:
    3.     accessToken: default_token
    4.     admin:
    5.       addresses: 'http://127.0.0.1:8080/xxl-job-admin '
    6.     executor:
    7.       address: ''
    8.       appname: xxl-job-executor-meeting
    9.       ip: 127.0.0.1
    10.       logpath: /data/applogs/xxl-job/jobhandler
    11.       logretentiondays: 30
    12.       port: 9999

     2.xxl-job配置类

    1. /**
    2.  * xxl-job config
    3.  *
    4.  * @author xuxueli 2017-04-28
    5.  */
    6. @Configuration
    7. public class XxlJobConfig {
    8.     private Logger logger = LoggerFactory.getLogger(XxlJobConfig.class);
    9.  
    10.     @Value("${xxl.job.admin.addresses}")
    11.     private String adminAddresses;
    12.  
    13.     @Value("${xxl.job.accessToken}")
    14.     private String accessToken;
    15.  
    16.     @Value("${xxl.job.executor.appname}")
    17.     private String appname;
    18.  
    19.     @Value("${xxl.job.executor.address}")
    20.     private String address;
    21.  
    22.     @Value("${xxl.job.executor.ip}")
    23.     private String ip;
    24.  
    25.     @Value("${xxl.job.executor.port}")
    26.     private int port;
    27.  
    28.     @Value("${xxl.job.executor.logpath}")
    29.     private String logPath;
    30.  
    31.     @Value("${xxl.job.executor.logretentiondays}")
    32.     private int logRetentionDays;
    33.  
    34.  
    35.     @Bean
    36.     public XxlJobSpringExecutor xxlJobExecutor() {
    37.         logger.info(">>>>>>>>>>> xxl-job config init.");
    38.         XxlJobSpringExecutor xxlJobSpringExecutor = new XxlJobSpringExecutor();
    39.         xxlJobSpringExecutor.setAdminAddresses(adminAddresses);
    40.         xxlJobSpringExecutor.setAppname(appname);
    41.         xxlJobSpringExecutor.setAddress(address);
    42.         xxlJobSpringExecutor.setIp(ip);
    43.         xxlJobSpringExecutor.setPort(port);
    44.         xxlJobSpringExecutor.setAccessToken(accessToken);
    45.         xxlJobSpringExecutor.setLogPath(logPath);
    46.         xxlJobSpringExecutor.setLogRetentionDays(logRetentionDays);
    47.  
    48.         return xxlJobSpringExecutor;
    49.     }
    50.  
    51.     /**
    52.      * 针对多网卡、容器内部署等情况，可借助 "spring-cloud-commons" 提供的 "InetUtils" 组件灵活定制注册IP；
    53.      *
    54.      *      1、引入依赖：
    55.      *          
    56.      *             org.springframework.cloud
    57.      *             spring-cloud-commons
    58.      *             ${version}
    59.      *         
    60.      *
    61.      *      2、配置文件，或者容器启动变量
    62.      *          spring.cloud.inetutils.preferred-networks: 'xxx.xxx.xxx.'
    63.      *
    64.      *      3、获取IP
    65.      *          String ip_ = inetUtils.findFirstNonLoopbackHostInfo().getIpAddress();
    66.      */
    67.  
    68.  
    69. }

    3.定时任务

    1. /**
    2.  * XxlJob开发示例（Bean模式）
    3.  *
    4.  * 开发步骤：
    5.  *      1、任务开发：在Spring Bean实例中，开发Job方法；
    6.  *      2、注解配置：为Job方法添加注解 "@XxlJob(value="自定义jobhandler名称", init = "JobHandler初始化方法", destroy = "JobHandler销毁方法")"，注解value值对应的是调度中心新建任务的JobHandler属性的值。
    7.  *      3、执行日志：需要通过 "XxlJobHelper.log" 打印执行日志；
    8.  *      4、任务结果：默认任务结果为 "成功" 状态，不需要主动设置；如有诉求，比如设置任务结果为失败，可以通过 "XxlJobHelper.handleFail/handleSuccess" 自主设置任务结果；
    9.  *
    10.  * @author xuxueli 2019-12-11 21:52:51
    11.  */
    12. @Component
    13. public class SampleXxlJob {
    14.     private static Logger logger = LoggerFactory.getLogger(SampleXxlJob.class);
    15.  
    16.  
    17.     /**
    18.      * 1、简单任务示例（Bean模式）
    19.      */
    20.     @XxlJob("demoJobHandler")
    21.     public void demoJobHandler() throws Exception {
    22.         XxlJobHelper.log("XXL-JOB, Hello World.");
    23.  
    24.         for (int i = 0; i < 5; i++) {
    25.             XxlJobHelper.log("beat at:" + i);
    26.             TimeUnit.SECONDS.sleep(2);
    27.         }
    28.         // default success
    29.     }
    30.  
    31.  
    32.     /**
    33.      * 2、分片广播任务
    34.      */
    35.     @XxlJob("shardingJobHandler")
    36.     public void shardingJobHandler() throws Exception {
    37.  
    38.         // 分片参数
    39.         int shardIndex = XxlJobHelper.getShardIndex();
    40.         int shardTotal = XxlJobHelper.getShardTotal();
    41.  
    42.         XxlJobHelper.log("分片参数：当前分片序号 = {}, 总分片数 = {}", shardIndex, shardTotal);
    43.  
    44.         // 业务逻辑
    45.         for (int i = 0; i < shardTotal; i++) {
    46.             if (i == shardIndex) {
    47.                 XxlJobHelper.log("第 {} 片, 命中分片开始处理", i);
    48.             } else {
    49.                 XxlJobHelper.log("第 {} 片, 忽略", i);
    50.             }
    51.         }
    52.  
    53.     }
    54.  
    55.  
    56.     /**
    57.      * 3、命令行任务
    58.      */
    59.     @XxlJob("commandJobHandler")
    60.     public void commandJobHandler() throws Exception {
    61.         String command = XxlJobHelper.getJobParam();
    62.         int exitValue = -1;
    63.  
    64.         BufferedReader bufferedReader = null;
    65.         try {
    66.             // command process
    67.             ProcessBuilder processBuilder = new ProcessBuilder();
    68.             processBuilder.command(command);
    69.             processBuilder.redirectErrorStream(true);
    70.  
    71.             Process process = processBuilder.start();
    72.             //Process process = Runtime.getRuntime().exec(command);
    73.  
    74.             BufferedInputStream bufferedInputStream = new BufferedInputStream(process.getInputStream());
    75.             bufferedReader = new BufferedReader(new InputStreamReader(bufferedInputStream));
    76.  
    77.             // command log
    78.             String line;
    79.             while ((line = bufferedReader.readLine()) != null) {
    80.                 XxlJobHelper.log(line);
    81.             }
    82.  
    83.             // command exit
    84.             process.waitFor();
    85.             exitValue = process.exitValue();
    86.         } catch (Exception e) {
    87.             XxlJobHelper.log(e);
    88.         } finally {
    89.             if (bufferedReader != null) {
    90.                 bufferedReader.close();
    91.             }
    92.         }
    93.  
    94.         if (exitValue == 0) {
    95.             // default success
    96.         } else {
    97.             XxlJobHelper.handleFail("command exit value("+exitValue+") is failed");
    98.         }
    99.  
    100.     }
    101.  
    102.  
    103.     /**
    104.      * 4、跨平台Http任务
    105.      *  参数示例：
    106.      *      "url: http://www.baidu.com\n" +
    107.      *      "method: get\n" +
    108.      *      "data: content\n";
    109.      */
    110.     @XxlJob("httpJobHandler")
    111.     public void httpJobHandler() throws Exception {
    112.  
    113.         // param parse
    114.         String param = XxlJobHelper.getJobParam();
    115.         if (param==null || param.trim().length()==0) {
    116.             XxlJobHelper.log("param["+ param +"] invalid.");
    117.  
    118.             XxlJobHelper.handleFail();
    119.             return;
    120.         }
    121.  
    122.         String[] httpParams = param.split("\n");
    123.         String url = null;
    124.         String method = null;
    125.         String data = null;
    126.         for (String httpParam: httpParams) {
    127.             if (httpParam.startsWith("url:")) {
    128.                 url = httpParam.substring(httpParam.indexOf("url:") + 4).trim();
    129.             }
    130.             if (httpParam.startsWith("method:")) {
    131.                 method = httpParam.substring(httpParam.indexOf("method:") + 7).trim().toUpperCase();
    132.             }
    133.             if (httpParam.startsWith("data:")) {
    134.                 data = httpParam.substring(httpParam.indexOf("data:") + 5).trim();
    135.             }
    136.         }
    137.  
    138.         // param valid
    139.         if (url==null || url.trim().length()==0) {
    140.             XxlJobHelper.log("url["+ url +"] invalid.");
    141.  
    142.             XxlJobHelper.handleFail();
    143.             return;
    144.         }
    145.         if (method==null || !Arrays.asList("GET", "POST").contains(method)) {
    146.             XxlJobHelper.log("method["+ method +"] invalid.");
    147.  
    148.             XxlJobHelper.handleFail();
    149.             return;
    150.         }
    151.         boolean isPostMethod = method.equals("POST");
    152.  
    153.         // request
    154.         HttpURLConnection connection = null;
    155.         BufferedReader bufferedReader = null;
    156.         try {
    157.             // connection
    158.             URL realUrl = new URL(url);
    159.             connection = (HttpURLConnection) realUrl.openConnection();
    160.  
    161.             // connection setting
    162.             connection.setRequestMethod(method);
    163.             connection.setDoOutput(isPostMethod);
    164.             connection.setDoInput(true);
    165.             connection.setUseCaches(false);
    166.             connection.setReadTimeout(5 * 1000);
    167.             connection.setConnectTimeout(3 * 1000);
    168.             connection.setRequestProperty("connection", "Keep-Alive");
    169.             connection.setRequestProperty("Content-Type", "application/json;charset=UTF-8");
    170.             connection.setRequestProperty("Accept-Charset", "application/json;charset=UTF-8");
    171.  
    172.             // do connection
    173.             connection.connect();
    174.  
    175.             // data
    176.             if (isPostMethod && data!=null && data.trim().length()>0) {
    177.                 DataOutputStream dataOutputStream = new DataOutputStream(connection.getOutputStream());
    178.                 dataOutputStream.write(data.getBytes("UTF-8"));
    179.                 dataOutputStream.flush();
    180.                 dataOutputStream.close();
    181.             }
    182.  
    183.             // valid StatusCode
    184.             int statusCode = connection.getResponseCode();
    185.             if (statusCode != 200) {
    186.                 throw new RuntimeException("Http Request StatusCode(" + statusCode + ") Invalid.");
    187.             }
    188.  
    189.             // result
    190.             bufferedReader = new BufferedReader(new InputStreamReader(connection.getInputStream(), "UTF-8"));
    191.             StringBuilder result = new StringBuilder();
    192.             String line;
    193.             while ((line = bufferedReader.readLine()) != null) {
    194.                 result.append(line);
    195.             }
    196.             String responseMsg = result.toString();
    197.  
    198.             XxlJobHelper.log(responseMsg);
    199.  
    200.             return;
    201.         } catch (Exception e) {
    202.             XxlJobHelper.log(e);
    203.  
    204.             XxlJobHelper.handleFail();
    205.             return;
    206.         } finally {
    207.             try {
    208.                 if (bufferedReader != null) {
    209.                     bufferedReader.close();
    210.                 }
    211.                 if (connection != null) {
    212.                     connection.disconnect();
    213.                 }
    214.             } catch (Exception e2) {
    215.                 XxlJobHelper.log(e2);
    216.             }
    217.         }
    218.  
    219.     }
    220.  
    221.     /**
    222.      * 5、生命周期任务示例：任务初始化与销毁时，支持自定义相关逻辑；
    223.      */
    224.     @XxlJob(value = "demoJobHandler2", init = "init", destroy = "destroy")
    225.     public void demoJobHandler2() throws Exception {
    226.         XxlJobHelper.log("XXL-JOB, Hello World.");
    227.     }
    228.     public void init(){
    229.         logger.info("init");
    230.     }
    231.     public void destroy(){
    232.         logger.info("destroy");
    233.     }
    234.  
    235.  
    236. }

    启动执行器，服务启动端口8081 ，调度端口默认9999

    3.调度平台配置

            1. 在执行器管理里新建一个执行器，Appname配置执行器yml里映射的appname，如果调度中心和执行器在一个单一机器内，可以使用自动注册，如果是不同机器或者同一机器不同VDC或不同集群内需要手动录入，也就是配置执行器的服务

         2.建完执行器后，创建调度任务，选择上面创建的执行器，配置Corn表达式，这里和定时任务的配置统一

     重点在任务配置里的JobHander需要和执行器里的调度接口的映射匹配

     执行器和调度任务配置好后，就可以测试运行了，可以通过日志查看运行状态

     4.xxl-job整合现有springframe项目

    整合现有项目，只需要将xxl-job的依赖打包到maven里

    1. <dependency>
    2.             <groupId>com.xuxueli</groupId>
    3.             <artifactId>xxl-job-core</artifactId>
    4.             <version>2.0.2</version>
    5.         </dependency>

     剩下的调度配置项和调度接口，定时任务和上面一样需要做相同配置

    end

  • 相关阅读:
    【Vue3+Vite+Ts+element-plus】 使用tsx 动态表格封装
    2023最新SSM计算机毕业设计选题大全（附源码+LW）之java招生管理系统2ij21
    猿创征文｜UnitySqlite持久化数据
    R--基础知识总结
    vue3.0--1.vue3.0环境集成、setup、ref函数、reactive函数、计算属性（computed）
    安装2023最新版PyCharm来开发Python应用程序
    面试：List View和RecyclerView缓存策略对比
    【C++】初阶模板
    【优化选址】遗传算法求解物流配送中心选址【含Matlab源码 1917期】
    设计模式day13
  • 原文地址：https://blog.csdn.net/Zxiaobinggan/article/details/127801365