xxl-job入门级教程（二）

XxlJobHelper使用在 jobHandler 中提供了 XxlJobHelper 类完成一些调度任务命令UI界面参数传递String jobParam XxlJobHelper.getJobParam();打印日志XxlJobHelper.log(hello world);返回调度任务结果失败// XxlJobHelper.handleFail(); XxlJobHelper.handleFail(error.);返回调度任务结果成功// XxlJobHelper.handleSuccess(); XxlJobHelper.handleSuccess(success.);xxl-job集群xxl-job集群为了避免单点故障任务调度系统通常需要通过集群实现系统高可用由于任务调度系统的特殊性“调度”和“任务”两个模块需要均支持集群部署由于职责不同因此各自集群侧重点也有有所不同。调度中心集群目标为避免调度模块单点故障集群节点需要通过锁或命名服务保证单个任务的单次触发只在其中一个节点上生效以防止任务的重复触发。执行器集群目标为避免任务模块单点故障进一步可以通过自定义路由策略实现Failover等高级功能从而在执行器某台机器节点故障时自动转移不会影响到任务的正常触发执行执行器配置调度中心集群# 调度中心部署地址,多个配置逗号分隔 http://address01,http://address02 xxl.job.admin.addresseshttp://127.0.0.1:8080/xxl-job-admin,http://127.0.0.1:8079/xxl-job-adminxxl-job海量数据处理-分片需求有一个任务需要处理100W条数据每条数据的业务逻辑处理要0.1s对于普通任务来说只有一个线程来处理 可能需要10万秒才能处理完业务则严重受影响案例双十一大促给1000万用户发营销短信什么是分片任务执行器集群部署如果任务的路由策略选择【分片广播】一次任务调度将会【广播触发】对应集群中所有执行器执行一次任务同时系统自动传递分片参数执行器可根据分片参数开发分片任务需要处理的海量数据以执行器为划分每个执行器分配一定的任务数并行执行XXL-Job支持动态扩容执行器集群从而动态增加分片数量到达更快处理任务分片的值是调度中心分配的// 当前分片数从0开始即执行器的序号 int shardIndex XxlJobHelper.getShardIndex(); //总分片数执行器集群总机器数量 int shardTotal XxlJobHelper.getShardTotal();解决思路如果将100W数据均匀分给集群里的10台机器同时处理每台机器耗时1万秒即可耗时会大大缩短也能充分利用集群资源在xxl-job里可以配置执行器集群有10个机器那么分片总数是10分片序号0~9 分别对应那10台机器。分片方式id % 分片总数 余数是0 的在第1个执行器上执行id % 分片总数 余数是1 的在第2个执行器上执行id % 分片总数 余数是2 的在第3个执行器上执行...id % 分片总数 余数是9 的在第10个执行器上执行在【任务管理】配置任务的路由策略选择【分片广播】代码示例需求100个用户分片处理新增job测试案例。在每个执行器上都配置相同的调度任务如下XxlJob(shardingJobHandler) public void shardingJobHandler() throws Exception { // 当前分片数从0开始即执行器的序号 int shardIndex XxlJobHelper.getShardIndex(); //总分片数执行器集群总机器数量 int shardTotal XxlJobHelper.getShardTotal(); XxlJobHelper.log(分片参数当前分片序号 {}, 总分片数 {}, shardIndex, shardTotal); // 业务逻辑 for (int i 0; i shardTotal; i) { if (i shardIndex) { log.info(第 {} 片, 命中分片开始处理, i); } else { log.info(第 {} 片, 忽略, i); } } }xxl-job底层API对接需求背景项目中需要后端以无入侵的方式调用 xxl-job 调度中心API服务。然而调度中心设置了登录调度中心API接口对cookie进行了验证feign访问调度中心API服务时需通过其登录验证。在 xxl-job-admin 中JobInfoController 提供了完整的API调用xxl-job底层API对接步骤在项目中引入 Feign 的maven依赖dependency groupIdorg.springframework.cloud/groupId artifactIdspring-cloud-starter-openfeign/artifactId version4.0.0/version /dependencyXxlJobClientFeignClient(name xxlJobClient, url ${xxl.job.admin.addresses}) public interface XxlJobClient { /** * xxl-job登录接口 * * param params 参数 * return 响应信息 */ PostMapping(value /login, consumes MediaType.APPLICATION_FORM_URLENCODED_VALUE) Response login(RequestBody MapString, Object params); /** * 创建定时任务 * * param cookie cookie * param params 定时任务参数 * return 定时任务ID */ PostMapping(value /jobinfo/add, consumes MediaType.APPLICATION_FORM_URLENCODED_VALUE) ReturnTString add(RequestHeader(Cookie) String cookie, RequestBody MapString, Object params); /** * 更新定时任务 * * param cookie cookie * param params 定时任务更新参数 * return 执行结果 */ PutMapping(value /jobinfo/update, consumes MediaType.APPLICATION_FORM_URLENCODED_VALUE) ReturnTString update(RequestHeader(Cookie) String cookie, RequestBody MapString, Object params); /** * 删除定时任务 * * param cookie cookie * param id 定时任务更新参数 * return 执行结果 */ DeleteMapping(value /jobinfo/remove) ReturnTString remove(RequestHeader(Cookie) String cookie, RequestParam(id) int id); /** * 开启任务 * * param cookie cookie * param id 定时任务ID * return 执行结果 */ PutMapping(value /jobinfo/start) ReturnTString start(RequestHeader(Cookie) String cookie, RequestParam(id) int id); /** * 结束任务 * * param cookie cookie * param id 定时任务ID * return 执行结果 */ PutMapping(value /jobinfo/stop) ReturnTString stop(RequestHeader(Cookie) String cookie, RequestParam(id) int id); /** * 结束任务 * * param cookie cookie * param params 查询参数 * return 执行结果 */ GetMapping(value /joblog/pageList) MapString, Object log(RequestHeader(Cookie) String cookie, RequestParam(params) MapString, Object params); }具体调用的类Slf4j Component public class XxlJobComponent { /** * xxljob账号 */ Value(${xxl.job.admin.username}) private String userName; /** * xxljob密码 */ Value(${xxl.job.admin.password}) private String password; /** * xxl-job客户端 */ Resource private XxlJobClient xxlJobClient; /** * redis操作类 */ Resource private RedisTemplate redisTemplate; /** * 登录xxl-job */ public void login() { MapString, Object userInfo new HashMap(); userInfo.put(userName, userName); userInfo.put(password, password); // 设置cookie永久有效对应xxl-job记住密码 userInfo.put(ifRemember, on); Response response xxlJobClient.login(userInfo); if (HttpStatus.HTTP_OK response.status()) { response.headers().get(XxlJobConstant.COOKIE_KEY).forEach(e - { if (e.contains(XxlJobConstant.XXL_JOB_LOGIN_IDENTITY)) { redisTemplate.opsForValue().set(CacheConstant.XXL_JOB_COOKIE_REDIS_KEY, e, CacheConstant.XXL_JOB_COOKIE_REDIS_TIMEOUT, TimeUnit.HOURS); } }); } else { throw ExceptionFactory.systemException(ErrorCode.LOGIN_XXL_JOB_FAILURE_EXCEPTION); } } /** * 创建任务 * * param addOrUpdateXxlJobInfoRequest 任务参数 * return 任务ID */ public Integer add(AddOrUpdateXxlJobInfoRequest addOrUpdateXxlJobInfoRequest) { if (Boolean.FALSE.equals(redisTemplate.hasKey(CacheConstant.XXL_JOB_COOKIE_REDIS_KEY))) { login(); } if (ObjectUtils.isNotEmpty(addOrUpdateXxlJobInfoRequest)) { HttpResultForXxlJob result xxlJobClient.add(String.valueOf(redisTemplate.opsForValue().get(CacheConstant.XXL_JOB_COOKIE_REDIS_KEY)), BeanUtil.beanToMap(addOrUpdateXxlJobInfoRequest)); if (HttpStatus.HTTP_OK result.getCode()) { log.info(result.getMsg()); return (int) result.getContent(); } } return null; } /** * 更新任务 * * param addOrUpdateXxlJobInfoRequest 任务参数 */ public void update(AddOrUpdateXxlJobInfoRequest addOrUpdateXxlJobInfoRequest) { if (Boolean.FALSE.equals(redisTemplate.hasKey(CacheConstant.XXL_JOB_COOKIE_REDIS_KEY))) { login(); } if (ObjectUtils.isNotEmpty(addOrUpdateXxlJobInfoRequest)) { HttpResultForXxlJob result xxlJobClient.update(String.valueOf(redisTemplate.opsForValue().get(CacheConstant.XXL_JOB_COOKIE_REDIS_KEY)), BeanUtil.beanToMap(addOrUpdateXxlJobInfoRequest)); log.info(result.getMsg()); AbstractAssert.isTrue(HttpStatus.HTTP_OK result.getCode(), ErrorCode.UPDATE_JOB_FAILURE_EXCEPTION); } } /** * 删除任务 * * param id 任务ID */ public void remove(Integer id) { if (Boolean.FALSE.equals(redisTemplate.hasKey(CacheConstant.XXL_JOB_COOKIE_REDIS_KEY))) { login(); } if (ObjectUtils.isNotEmpty(id)) { HttpResultForXxlJob result xxlJobClient.remove(String.valueOf(redisTemplate.opsForValue().get(CacheConstant.XXL_JOB_COOKIE_REDIS_KEY)), id); log.info(result.getMsg()); AbstractAssert.isTrue(HttpStatus.HTTP_OK result.getCode(), ErrorCode.REMOVE_JOB_FAILURE_EXCEPTION); } } /** * 启动任务 * * param id 任务ID */ public void start(Integer id) { if (Boolean.FALSE.equals(redisTemplate.hasKey(CacheConstant.XXL_JOB_COOKIE_REDIS_KEY))) { login(); } if (ObjectUtils.isNotEmpty(id)) { HttpResultForXxlJob result xxlJobClient.start(String.valueOf(redisTemplate.opsForValue().get(CacheConstant.XXL_JOB_COOKIE_REDIS_KEY)), id); log.info(result.getMsg()); AbstractAssert.isTrue(HttpStatus.HTTP_OK result.getCode(), ErrorCode.START_JOB_FAILURE_EXCEPTION); } } /** * 停止任务 * * param id 任务ID */ public void stop(Integer id) { if (Boolean.FALSE.equals(redisTemplate.hasKey(CacheConstant.XXL_JOB_COOKIE_REDIS_KEY))) { login(); } if (ObjectUtils.isNotEmpty(id)) { HttpResultForXxlJob result xxlJobClient.stop(String.valueOf(redisTemplate.opsForValue().get(CacheConstant.XXL_JOB_COOKIE_REDIS_KEY)), id); log.info(result.getMsg()); AbstractAssert.isTrue(HttpStatus.HTTP_OK result.getCode(), ErrorCode.STOP_JOB_FAILURE_EXCEPTION); } } /** * 查询日志 * * param params 查询参数 * return 结果集 */ Retryable(value SystemException.class, backoff Backoff(delay 2000L, multiplier 1.5)) public List log(Map params) { try { if (Boolean.FALSE.equals(redisTemplate.hasKey(CacheConstant.XXL_JOB_COOKIE_REDIS_KEY))) { login(); } if (ObjectUtils.isEmpty(params)) { return Lists.newArrayList(); } JSONObject result xxlJobClient.log(String.valueOf(redisTemplate.opsForValue().get(CacheConstant.XXL_JOB_COOKIE_REDIS_KEY)), params); if (ObjectUtils.isNotEmpty(result) ObjectUtils.isNotEmpty(result.getString(data))) { return JSON.parseArray(result.getString(data), XxlJobLogDTO.class); } } catch (FeignException e) { redisTemplate.delete(CacheConstant.XXL_JOB_COOKIE_REDIS_KEY); throw ExceptionFactory.systemException(远程操作xxl-job失败进行重试, e); } return Lists.newArrayList(); } /** * 重试次数达到最大后回调处理 * * param systemException 重试异常 */ Recover public void recoverCallback(SystemException systemException) { log.error(远程操作xxl-job异常, systemException); } }