xxl-job动态添加执行任务

雨点打透心脏的1/2处 2022-10-05 01:52 515阅读 0赞

# 1.概述 #

在很多场景中，需要根据用户选择，延期执行某些事情。

# 2.背景 #

在资源分发事件管理中，审批事件可以选择，事件具体的执行时间。这就需要任务调度器协助。

> 定时任务轮训也可以,在分布式场景中存在缺陷。

# 3.开始 #

## 3.1xxl-job配置 ##

在application.yml

xxl:
      job:
        accessToken:   # 访问token
        enabled: true # 是否启用xxl-job
        admin:
          addresses: http://10.0.248.38:9980/xxl-job-admin/
          login-username: admin   #管理平台登录用户名
          login-pwd: 123456   #管理平台
        executor:
          port: 9993   #执行器端口
          appname: id-sync-dev  #执行器地址
          logpath: logs/xxl-job/jobhandler  #日志路径
          logretentiondays: 30   #日志保存天数

config配置

@Slf4j
    @Configuration
    @ConditionalOnProperty(prefix = "xxl.job", name = "enabled", havingValue = "true")
    public class XxlJobConfig { 
    
        @Value("${xxl.job.admin.addresses}")
        private String adminAddresses;
    
        @Value("${xxl.job.executor.appname}")
        private String appName;
    
        @Value("${xxl.job.executor.ip:}")
        private String ip;
    
        @Value("${xxl.job.executor.port:-1}")
        private int port;
    
        @Value("${xxl.job.accessToken:}")
        private String accessToken;
    
        @Value("${xxl.job.executor.logpath}")
        private String logPath;
    
        @Value("${xxl.job.executor.logretentiondays}")
        private int logRetentionDays;
    
        /** * 注册xxlJob执行器 * * @return */
        @Bean
        public XxlJobSpringExecutor xxlJobExecutor() { 
            log.info(">>>>>>>>>>> xxl-job config init.");
            XxlJobSpringExecutor xxlJobSpringExecutor = new XxlJobSpringExecutor();
            xxlJobSpringExecutor.setAdminAddresses(adminAddresses);
            //执行器 appname
            xxlJobSpringExecutor.setAppname(appName);
            //执行器 注册ip 为空 或null 会自动获取
            xxlJobSpringExecutor.setIp(ip);
            //执行器端口 为空 或null 会自动设置未使用的端口
            xxlJobSpringExecutor.setPort(port);
            xxlJobSpringExecutor.setAccessToken(accessToken);
            xxlJobSpringExecutor.setLogPath(logPath);
            xxlJobSpringExecutor.setLogRetentionDays(logRetentionDays);
            return xxlJobSpringExecutor;
        }
    
    }

如果没有设置执行器ip,在注册执行器时会自动获取

## 3.2添加任务流程 ##

id\_sync调用添加任务方法1 获取执行器id(添加任务时，需要执行器id,相当于是对任务进行分组)2 根据appname查询，执行器id(需要携带登录之后返回的cookie，因为此url xxl-job需要进行认证)3 创建执行器(携带登录之后返回的cookie)4 注册执行器,类型设置为自动注册5 获取第一个执行器id6alt\[查询结果为空\]\[返回结果不为空\]id\_sync

> 官方提供的RESTful API 触发任务接口不能满足项目，官方提供的为，根据任务id,触发任务
> 
> https://www.xuxueli.com/xxl-job/\#c%E3%80%81%E8%A7%A6%E5%8F%91%E4%BB%BB%E5%8A%A1

## 3.3核心代码 ##

> 以下代码，发送http请求使用的是，hutool-http(基于HttpUrlConnection的Http客户端封装),详见
> 
> https://hutool.cn/docs/\#/

/** * xxl动态添加任务 * * @author Created by niugang on 2021-06-02 08:32 */
    @Component
    @Slf4j
    @RefreshScope
    public class EventXxlJobUtil { 
    
    
        @Value("${xxl.job.admin.login-username:admin}")
        private String loginUsername;
    
        @Value("${xxl.job.admin.login-pwd:123456}")
        private String loginPwd;
    
        @Value("${xxl.job.executor.ip:}")
        private String ip;
    
    
        @Value("${xxl.job.admin.addresses:''}")
        private String adminAddresses;
    
        @Value("${xxl.job.executor.port:-1}")
        private int port;
    
        @Value("${spring.profiles.active}")
        private String profiles;
    
        @Value("${xxl.job.executor.appname}")
        private String appName;
    
        @Value("${xxl.job.accessToken:}")
        private String accessToken;
    
    
        private final static String TITLE = "身份中心资源分发";
    
        /** * 增加执行任务 * * @param eventId 事件id * @param executorDate 执行日期 */
        public void addExecutorTask(Long eventId, Long tenantId, Date executorDate) { 
            Assert.notNull(eventId, "eventId must be not null");
            String desc = DateUtil.format(executorDate, DatePattern.CHINESE_DATE_TIME_PATTERN + "执行资源分发任务(" + eventId + ")");
            Map<String, Object> paramMap = Maps.newHashMapWithExpectedSize(16);
            //执行器主键ID,获取执行器主键
            paramMap.put("jobGroup", getJobGroupId());
            paramMap.put("jobDesc", desc);
            paramMap.put("executorRouteStrategy", "FIRST");
            String cron = getCron(executorDate);
            paramMap.put("cronGen_display", cron);
            //任务执行CRON表达式
            paramMap.put("jobCron", cron);
            //任务模式，可选值参考 com.xxl.job.core.glue.GlueTypeEnum
            paramMap.put("glueType", "BEAN");
            //执行器，任务Handler名称
            paramMap.put("executorHandler", "eventManagementJobHandler");
            //任务阻塞策略，可选值参考 com.xxl.job.core.enums.ExecutorBlockStrategyEnum
            paramMap.put("executorBlockStrategy", "SERIAL_EXECUTION");
            //任务超时时间，单位秒，大于零时生效
            paramMap.put("executorTimeout", 5);
            //失败重试次数
            paramMap.put("executorFailRetryCount", 2);
            //负责人
            paramMap.put("author", "admin");
            //GLUE备注
            paramMap.put("glueRemark", desc);
            //调度状态：0-停止，1-运行
            paramMap.put("triggerStatus", 1);
            HashMap<String, Object> executorParam = Maps.newHashMap();
            executorParam.put("eventId", eventId);
            executorParam.put("tenantId", tenantId);
            //执行器，任务参数
            paramMap.put("executorParam", JSON.toJSONString(executorParam));
            log.info("增加xxl执行任务,请求参数:{}", paramMap);
            HttpResponse response = HttpRequest.post(adminAddresses + "/jobinfo/add").form(paramMap).cookie(getCookie()).execute();
            if (response.isOk()) { 
                JSONObject jsonObject = JSON.parseObject(response.body());
                log.info("增加xxl执行任务成功,返回信息:{}", jsonObject);
                return;
            }
            log.error("调用xxl增加执行任务失败:{}", JSON.parseObject(response.body()));
            throw new BizException(ErrorCode.INVOKE_XXL_ADD_TASK_FAILED);
    
        }
    
        /** * 获取执行id * * @return int */
        private int getJobGroupId() { 
            Map<String, Object> paramMap = new HashMap<>();
            paramMap.put("appname", appName);
            paramMap.put("start", "0");
            paramMap.put("length", "100");
            log.info("获取xxl执行id,请求参数:{}", paramMap);
            //通过appname 查询appname对应的id
            HttpResponse response = HttpRequest.post(adminAddresses + "/jobgroup/pageList").form(paramMap).cookie(getCookie()).execute();
            if (response.isOk()) { 
                XxlSearchDto xxlSearchDto = JSON.parseObject(response.body(), XxlSearchDto.class);
                log.info("获取xxl执行器成功,返回信息:{}", xxlSearchDto);
                List<XxlSearchDto.DataEntity> list = xxlSearchDto.getData();
                if (CollectionUtils.isEmpty(list)) { 
                    //新增成功没有返回id
                    createJobGroup();
                    //在调一次列表查询 获取组id
                    return getJobGroupId();
                }
                XxlSearchDto.DataEntity dataEntity = list.get(0);
                return dataEntity.getId();
            }
            log.error("调用xxl获取执行器失败:{}", JSON.parseObject(response.body()));
            throw new BizException(ErrorCode.INVOKE_XXL_GET_EXECUTOR_FAILED);
        }
    
        /** * 创建执行器 */
        private void createJobGroup() { 
            Map<String, Object> paramMap = new HashMap<>();
            paramMap.put("appname", appName);
            paramMap.put("title", TITLE + profiles);
            //注册方式 0自动 1 为手动注册
            paramMap.put("addressType", "0");
            log.info("调用xxl增加执行器,请求参数：{}", paramMap);
            HttpResponse response = HttpRequest.post(adminAddresses + "/jobgroup/save").form(paramMap).cookie(getCookie()).execute();
            if (response.isOk()) { 
                log.info("调用xxl增加执行器,成功返回信息:{}", JSON.parseObject(response.body()));
                //注册执行器
                registry();
                return;
            }
            throw new BizException(ErrorCode.INVOKE_XXL_ADD_EXECUTOR_FAILED);
        }
    
        /** * 注册执行器 * 不能设置超时 * AdminBiz adminBiz = new AdminBizClient(adminAddresses, accessToken) * ReturnT<String> returnT = adminBiz.registry(registryParam); */
        private void registry() { 
            //copy xxl-job 源码
            port = port > 0 ? port : NetUtil.findAvailablePort(9999);
            ip = (ip != null && ip.trim().length() > 0) ? ip : IpUtil.getIp();
            String ipPort = IpUtil.getIpPort(ip, port);
            String address = "http://{ip_port}/".replace("{ip_port}", ipPort);
            RegistryParam registryParam = new RegistryParam(RegistryConfig.RegistType.EXECUTOR.name(), appName, address);
            ReturnT<String> returnT = XxlJobRemotingUtil.postBody(adminAddresses + "api/registry", accessToken, 6, registryParam, String.class);
            log.info("注册执行器返回结果：{}", returnT);
    
        }
    
    
        /*** * 生成 日期对应的 cron表达式 * convert Date to cron ,eg. "0 06 10 15 1 ? 2014" * @param date : 时间点 * @return String */
        private String getCron(Date date) { 
            String dateFormat = "ss mm HH dd MM ? yyyy";
            return DateUtil.format(date, dateFormat);
        }
    
    
        /** * 获取cookie * * @return String */
        private String getCookie() { 
            String path = adminAddresses + "/login";
            Map<String, Object> hashMap = new HashMap<>();
            hashMap.put("userName", loginUsername);
            hashMap.put("password", loginPwd);
            log.info("获取xxl cookie,请求参数：{}", hashMap);
            HttpResponse response = HttpRequest.post(path).form(hashMap).execute();
            boolean ok = response.isOk();
            if (ok) { 
                List<HttpCookie> cookies = response.getCookies();
                log.info("获取xxl cookie成功,返回信息:{}", cookies);
                StringBuilder sb = new StringBuilder();
                for (HttpCookie cookie : cookies) { 
                    sb.append(cookie.toString());
                }
                return sb.toString();
            }
            log.error("调用xxl获取cookie失败:{}", JSON.parseObject(response.body()));
            throw new BizException(ErrorCode.INVOKE_XXL_GET_COOKIE_FAILED);
    
        }
    }