分布式系统：xxl-job改造spring-cloud

修改后的源码仓库地址:GitHub. ：

改造原因#

原有的xxl-job使用自己实现的http协议进行注册以及调度等，与目前框架中本身的注册中心格格不入，会影响健康检查、日志处理、问题排查。
技术栈统一。避免执行器内包含两套注册逻辑。
提高分布式健壮性，原有的服务注册以及发现等功能较弱，且与实际应用可用与否完全无关，经常存在xxl-job线程出问题，但主服务正常，或主服务出问题，但xxl-job线程正常。
灰度扩展，目前系统灰度使用eureka定制实现，为执行器支持灰度，必须进行改造。

主要改造思路#

调度中心#

调度中心侧获取服务时，将原有的基于数据库的地址list，修改为动态从eureka中心获取服务的地址列表，两者通过xxl-job-admin配置的执行器app-name，与执行器的spring.application.name（即注册到eureka的服务标识）关联。

1
XxlJobGroup group = XxlJobAdminConfig.getAdminConfig().getXxlJobGroupDao().load(jobInfo.getJobGroup());
2
//@edit 如果是自动获取地址的话，则使用
3
if (group.getAddressType() == 0) {
4
    group.setAddressList(SpringAdminContext.getEurekaAddressList(group.getAppname()));
5
}

1
//这种静态和spring容器不分的写法，还是有点别扭的
2
@Component("springAdminContext")
3
public class SpringAdminContext {
4
    @Autowired
5
    DiscoveryClient discoveryClient;
6

7
    private static SpringAdminContext springAdminContext;
8
    @PostConstruct
9
    public void initialize() {
10
        springAdminContext = this;
11
        springAdminContext.discoveryClient=this.discoveryClient;
12
    }
13
    public static String getEurekaAddressList(String appName){
14
        //may be springContext not init
15
        if(springAdminContext !=null){
16
            DiscoveryClient discoveryClient = SpringAdminContext.springAdminContext.discoveryClient;
17
            List<ServiceInstance> instances = discoveryClient.getInstances(appName);
18
            StringBuilder addressBuilder = new StringBuilder();
19
            for (int i = 0; i < instances.size(); i++) {
20
                addressBuilder.append(instances.get(i).getUri().toString());
21
                if(i!=instances.size()) {
22
                    addressBuilder.append(",");
23
                }
24
            }
25
            return addressBuilder.toString();
26
        }else {
27
            return "";
28
        }
29

30
    }
31
}

调度中心侧调度服务时，增加灰度策略，在获取到eureka的instanceList后，从instance的meta原数据中取出灰度标识，进行灰度调度。代码结合1中的列表获取，具体灰度实现与百度到的eureka灰度相同，略。

调度中心执行器侧#

通过修改core包，将原有的注册线程删除，并删除embedServer的实现，修改为springMVC。

1
//修改后的代替embedServer的处理类
2
@Controller
3
public class XxlJobHandlerController {
4
    private static final Logger logger = LoggerFactory.getLogger(XxlJobHandlerController.class);
5

6
    private ExecutorBiz executorBiz;
7
    @Value("${xxl.job.accessToken:}")
8
    private String accessToken;
9
    @Autowired
10
    ThreadPoolExecutor bizThreadPool;
11
    @PostConstruct
12
    public void start() {
13
        executorBiz = new ExecutorBizImpl();
14
    }
15

16
    @PostMapping("/job/{method}")
17
    @ResponseBody
18
    public ReturnT jobHandle(HttpServletRequest httpServletRequest, HttpServletResponse httpServletResponse, @PathVariable("method") String methodName) {
19
        return doHandlerReq(httpServletRequest,httpServletResponse,"/"+methodName);
20
    }
21
    // ---------------------- registry ----------------------
22

23
    protected ReturnT doHandlerReq(HttpServletRequest httpServletRequest,HttpServletResponse httpServletResponse,String method) {
24
        try {
25
            //read request
26
            //@edit 这里请求都模拟的原有处理方式，包括header这些
27
            int contentLength = httpServletRequest.getContentLength();
28
            byte[] reqBody=new byte[contentLength];
29
            httpServletRequest.getInputStream().read(reqBody,0,contentLength);
30
            String requestData=new String(reqBody, StandardCharsets.UTF_8);
31
            String uri = method;
32
            String accessTokenReq = httpServletRequest.getHeader(XxlJobRemotingUtil.XXL_JOB_ACCESS_TOKEN);
33
            //@edit 这里原有是netty纯异步，但是到这边http就不太合适了，这么写虽然看起来吞吐会下降，但是一般web容器现在底层也支持nio了，应该关系不大。
34
            FutureTask<ReturnT> stringFutureTask=new FutureTask<ReturnT>(() -> process(uri, requestData, accessTokenReq));
35
            // invoke
36
            bizThreadPool.execute(stringFutureTask);
37
            ReturnT returnT = stringFutureTask.get();
38
            httpServletResponse.setHeader(HttpHeaders.CONTENT_TYPE, "text/html;charset=UTF-8");
39

40

41
            return returnT;
42
        } catch (Exception e) {
43
            return new ReturnT<String>(ReturnT.FAIL_CODE, "invalid request, uri-mapping empty.");
44
        }
45
    }
46

47
    private ReturnT process(String uri, String requestData, String accessTokenReq) {
48

49
        if (uri == null || uri.trim().length() == 0) {
50
            return new ReturnT<String>(ReturnT.FAIL_CODE, "invalid request, uri-mapping empty.");
51
        }
52
        if (accessToken != null
53
                && accessToken.trim().length() > 0
54
                && !accessToken.equals(accessTokenReq)) {
55
            return new ReturnT<String>(ReturnT.FAIL_CODE, "The access token is wrong.");
56
        }
57

58
        // services mapping
59
        try {
60
            if ("/beat".equals(uri)) {
61
                return executorBiz.beat();
62
            } else if ("/idleBeat".equals(uri)) {
63
                IdleBeatParam idleBeatParam = GsonTool.fromJson(requestData, IdleBeatParam.class);
64
                return executorBiz.idleBeat(idleBeatParam);
65
            } else if ("/run".equals(uri)) {
66
                TriggerParam triggerParam = GsonTool.fromJson(requestData, TriggerParam.class);
67
                return executorBiz.run(triggerParam);
68
            } else if ("/kill".equals(uri)) {
69
                KillParam killParam = GsonTool.fromJson(requestData, KillParam.class);
70
                return executorBiz.kill(killParam);
71
            } else if ("/log".equals(uri)) {
72
                LogParam logParam = GsonTool.fromJson(requestData, LogParam.class);
73
                return executorBiz.log(logParam);
74
            } else {
75
                return new ReturnT<String>(ReturnT.FAIL_CODE, "invalid request, uri-mapping(" + uri + ") not found.");
76
            }
77
        } catch (Exception e) {
78
            logger.error(e.getMessage(), e);
79
            return new ReturnT<String>(ReturnT.FAIL_CODE, "request error:" + ThrowableUtil.toString(e));
80
        }
81
    }
82
}

修改处理callback的逻辑，使得任务执行结果可以回调到admin

1
//@edit 这个方法主要用于执行器在获取调度中心列表时调用，目的是执行器获取调度中心列表进行callback回调通知执行结果。
2
    //如果callback失败或者这里出问题，那么会导致在管理台看到的执行结果永远没有。
3
    //管理台的调度结果和执行结果是分开的，调度结果依赖单次http请求，执行结果依赖callback
4
    // @see TriggerCallbackThread
5
    public static List<AdminBiz> getAdminBizList(){
6
        List<AdminBiz> adminBizs=new ArrayList<>();
7
        String adminAppName="xxl-job-admin-cloud";
8
        List<String> addressList = Arrays.asList(SpringContext.getEurekaAddressList(adminAppName).split(","));
9
        addressList.forEach(e ->{
10
            adminBizs.add(new AdminBizClient(e.concat("/").concat("xxl-job-admin"),accessToken));
11
        });
12
        return adminBizs;
13
    }

总结#

其实还有一些细节的修改包括eureka的配置，原有注册代码等的调整，就不好一一列出来了，具体可以拉下来项目搜索@edit，主要修改的地方我都加了这个。

乘桴

改造原因#

主要改造思路#

调度中心#

调度中心执行器侧#

总结#

继续阅读

换条路线

乘桴

改造原因#

主要改造思路#

调度中心#

调度中心 执行器侧#

总结#

分享文章

沿着主题读

换条路线

调度中心执行器侧#