点击上方“Java基基”，选择“设为星标”

做积极的人，而不是积极废人！

每天 14:00 更新文章，每天掉亿点点头发...

源码精品专栏

 

来源：码猿技术专栏

d9ff4d02c4269e250d12f6bd1603f0f9.png

任务调度是java项目中常用的一种组件，可以指定任务在何时进行触发，最熟悉的是spring框架里面的quartz ；

较流行的有一些分布式调度组件，比如elastic-job /azkaban ,都是基于quartz 二次开发

今天介绍一款分布式的任务调度框架：xxl-job 。

项目介绍

xxl-job是一款极容易学习上手的轻量级开源分布式调度框架,分为管理端和执行端两块,管理端负责配置任务信息以及查看任务执行日志,执行端只需要配置与管理端的连接信息就可以进行具体的任务逻辑开发了,目前版本还在持续迭代中,使用简单,功能强大,具体功能特性可以看下官方介绍。废话不多说,直接进入实战吧。

基于 Spring Boot + MyBatis Plus + Vue & Element 实现的后台管理系统 + 用户小程序，支持 RBAC 动态权限、多租户、数据权限、工作流、三方登录、支付、短信、商城等功能。

项目地址：https://github.com/YunaiV/ruoyi-vue-pro

实战

1.服务端部署

从https://github.com/xuxueli/xxl-job下载项目,用mysql客户端工具Navicat执行项目根目录下doc/db/table_xxl_job.sql文件,库名自己可以自行修改,一共8张表,如下:

6e109fccc7ea08443a3b17737675b655.png

创建一个新的spring boot项目,将下载的xxl-job-admin目录下的文件以及pom.xml文件都拷贝到新建的项目中(如果不想新建项目可以直接用下载下来的项目进行修改部署),修改application.properties中的数据库连接信息。

61be88911730e13bacd7d4cf952b9476.png

小编是自己创建的新项目,需要手动改了pom.xml依赖xxl-job-core的版本为2.2.0

3bf1ab515693bbdf22bef61cc2dc51ce.png

修改logback.xml中的日志输出路径。

227399c13e1c5ac83c2c94ecdef20aa2.png

好了,以上3步曲就搞定整个服务端配置了,启动项目,并访问http://localhost:8080/xxl-job-admin/ ,默认管理员账号admin/123456进行登录。

4e5635f8d45bcb743f8928289917e654.png

这交互,可以啊,是不是很带感。

2.执行端配置

创建一个新的module,跟服务端一样,也需要修改下logback.xml以及在pom.xml添加xxl-job-core的依赖。

为了模拟分布式效果,小编创建了2个配置文件来区分2个执行服务。

application-9998.properties

cdf6a066c4f3573839748b56e50a9cd1.png

application-9999.properties

c65bbbff134f611064b4621f4fc41d7d.png

细心的童鞋会发现只有server.port和xxl.job.executor.port不同,执行器服务跟spring boot一样,自带内嵌tomcat,也会暴露一个端口注册到服务端,进行高可用负载。

创建一个java config类,定义一个使用配置的XxlJobSpringExecutor执行类,如下

9921a7ca70844fa4950403c2a0814c0c.png

3cbae238f0727bd3fba083e99d4e5839.png配置2个启动配置,分别启动,效果如下:

ca8e694f7f9cab7bbf0a0f0d8575fb7c.png dcd34d4c67c82822502e25f7d9166089.png

完美启动2个服务,看下服务端平台是不是有这两台执行服务的注册信息。

1390b2303e2ec0d2426f0d5a6353be55.png

注意:为了演示,事先创建了一个执行器,AppName一定要与配置文件中xxl.job.executor.appname一致。

3.任务开发

3.1 基于方法注解任务

话不多说,直接上代码把,毕竟代码是程序员最好的交流方式。

3a7699301206c9ffb939a00aa111deea.png
3.2 基于api任务
f1ecc4ab8c7d0c597f2d7e210201c743.png
3.3 分片广播任务
b16ac1677db55efd2c679dc9958c6e44.png

上面是整理的比较实用的任务创建方式,个人偏好于注解形式,方法上加一个注解就完事了。

4.任务执行

剩下的就是傻白甜的界面操作了,走起。

4.1 单任务执行

创建一个路由策略为轮询的任务,指定corn表达式,并填入JobHandler为myJobAnnotationHandler,myJobAnnotationHandler其实就是spring IOC容器中管理bean的名称,有兴趣的童鞋可以看下源码。

113674f9a0918d6524332713f803632f.png

为了演示效果,点击执行一次并进行任务参数输入。

0e2a0c4e4b7676b9429c3f1f4ca2600f.png 2481fd87db4331d59c26596cbda9c4d0.png

轮询调用执行器服务效果如下:

93cf988585d6eca49fa6802f340d234a.png e9ca91ee30593dec83d3a09db5e6dab2.png
4.2 子任务执行

更新任务,并指定子任务id为5,多个子任务的需要以逗号隔开

e9309cde4dcc8a3f2c68786013113738.png

执行任务结果如下

5f28f5ef46795b76920ecdcd3e0ef1a1.png
4.3 分片广播任务执行

分片任务其实就是广播功能,每次触发,每个执行服务的业务执行类都会被调用,类似于kafka里面的不同消费组都要对同一个topic进行消费一样。

74f5bcb8841eba2a8e75b782b69b45c3.png

执行后的效果如下

1c7b5c49ff7f82b75c71b816b6d1269b.png a2f420aa0bb26a454984980bf29a959b.png

太强势了,需要定时刷新项目中的配置信息,用这个方式很完美。

5.任务日志

任务日志其实是很重要的一块,方便回溯任务历史执行情况, 以便跟踪问题并矫正丢失的业务数据

f450351bc833bdc0e1720e145de1626e.png

查看调度备注,父子任务调度信息非常详细,子任务可以通过执行备注查看执行情况

b1d0d8a523b382a4fae3897ef1ed9b30.png 9ada03720e91a8d4cd45d466dc256483.png

查看控制台输出,里面的日志是执行器中XxlJobLogger类打印出来的

d48de01abbea468181ac86e20cbed493.png

基于微服务的思想，构建在 B2C 电商场景下的项目实战。核心技术栈，是 Spring Boot + Dubbo 。未来，会重构成 Spring Cloud Alibaba 。

项目地址：https://github.com/YunaiV/onemall

通信底层介绍

xxl-job 使用 netty http 的方式进行通信，虽然也支持 Mina，jetty，netty tcp 等方式，但是代码里面固定写死的是 netty http。

通信整体流程

我以调度器通知执行器执行任务为例，绘制的活动图：

4c058275fcf140dae64cce24eed28862.png
活动图

惊艳的设计

看完了整个处理流程代码，设计上可以说独具匠心，将 netty，多线程的知识运用得行云流水。

我现在就将这些设计上出彩的点总结如下：

1. 使用动态代理模式，隐藏通信细节

xxl-job 定义了两个接口 ExecutorBiz，AdminBiz，ExecutorBiz 接口中封装了向心跳，暂停，触发执行等操作，AdminBiz 封装了回调，注册，取消注册操作，接口的实现类中，并没有通信相关的处理。

XxlRpcReferenceBean 类的 getObject() 方法会生成一个代理类，这个代理类会进行远程通信。

2. 全异步处理

执行器收到消息进行反序列化，并没有同步执行任务代码，而是将任务信息存储在 LinkedBlockingQueue 中，异步线程从这个队列中获取任务信息，然后执行。

而任务的处理结果，也不是说处理完之后，同步返回的，也是放到回调线程的阻塞队列中，异步的将处理结果返回回去。

这样处理的好处就是减少了 netty 工作线程的处理时间，提升了吞吐量。

3. 对异步处理的包装

对异步处理进行了包装，代码看起来是同步调用的。

我们看下调度器，XxlJobTrigger 类触发任务执行的代码：

public static ReturnT<String> runExecutor(TriggerParam triggerParam, String address){
    ReturnT<String> runResult = null;
    try {
        ExecutorBiz executorBiz = XxlJobScheduler.getExecutorBiz(address);
        //这里面做了很多异步处理，最终同步得到处理结果
        runResult = executorBiz.run(triggerParam);
    } catch (Exception e) {
        logger.error(">>>>>>>>>>> xxl-job trigger error, please check if the executor[{}] is running.", address, e);
        runResult = new ReturnT<String>(ReturnT.FAIL_CODE, ThrowableUtil.toString(e));
    }

    StringBuffer runResultSB = new StringBuffer(I18nUtil.getString("jobconf_trigger_run") + "：");
    runResultSB.append("<br>address：").append(address);
    runResultSB.append("<br>code：").append(runResult.getCode());
    runResultSB.append("<br>msg：").append(runResult.getMsg());

    runResult.setMsg(runResultSB.toString());
    return runResult;
}

ExecutorBiz.run 方法我们说过了，是走的动态代理，和执行器进行通信，执行器执行结果也是异步处理完，才返回的，而这里看到的 run 方法是同步等待处理结果返回。

我们看下xxl-job是如何同步获取处理结果的：调度器向执行器发出消息后，该线程阻塞。等到执行器处理完毕后，将处理结果返回，唤醒被阻塞的线程，调用处拿到返回值。

动态代理代码如下：

//代理类中的触发调用
if (CallType.SYNC == callType) {
   // future-response set
   XxlRpcFutureResponse futureResponse = new XxlRpcFutureResponse(invokerFactory, xxlRpcRequest, null);
   try {
      // do invoke
      client.asyncSend(finalAddress, xxlRpcRequest);

      // future get
      XxlRpcResponse xxlRpcResponse = futureResponse.get(timeout, TimeUnit.MILLISECONDS);
      if (xxlRpcResponse.getErrorMsg() != null) {
         throw new XxlRpcException(xxlRpcResponse.getErrorMsg());
      }
      return xxlRpcResponse.getResult();
   } catch (Exception e) {
      logger.info(">>>>>>>>>>> xxl-rpc, invoke error, address:{}, XxlRpcRequest{}", finalAddress, xxlRpcRequest);

      throw (e instanceof XxlRpcException)?e:new XxlRpcException(e);
   } finally{
      // future-response remove
      futureResponse.removeInvokerFuture();
   }
}

XxlRpcFutureResponse 类中实现了线程的等待，和线程唤醒的处理：

//返回结果，唤醒线程
public void setResponse(XxlRpcResponse response) {
   this.response = response;
   synchronized (lock) {
      done = true;
      lock.notifyAll();
   }
}

@Override
    public XxlRpcResponse get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
        if (!done) {
            synchronized (lock) {
                try {
                    if (timeout < 0) {
            //线程阻塞
                        lock.wait();
                    } else {
                        long timeoutMillis = (TimeUnit.MILLISECONDS==unit)?timeout:TimeUnit.MILLISECONDS.convert(timeout , unit);
                        lock.wait(timeoutMillis);
                    }
                } catch (InterruptedException e) {
                    throw e;
                }
            }
        }

        if (!done) {
            throw new XxlRpcException("xxl-rpc, request timeout at:"+ System.currentTimeMillis() +", request:" + request.toString());
        }
        return response;
    }

有的同学可能会问了，调度器接收到返回结果，怎么确定唤醒哪个线程呢？

每一次远程调用，都会生成 uuid 的请求 id，这个 id 是在整个调用过程中一直传递的，就像一把钥匙，在你回家的的时候，拿着它就带开门。

这里拿着请求 id 这把钥匙，就能找到对应的 XxlRpcFutureResponse，然后调用 setResponse 方法，设置返回值，唤醒线程。

public void notifyInvokerFuture(String requestId, final XxlRpcResponse xxlRpcResponse){

    // 通过requestId找到XxlRpcFutureResponse，
    final XxlRpcFutureResponse futureResponse = futureResponsePool.get(requestId);
    if (futureResponse == null) {
        return;
    }
    if (futureResponse.getInvokeCallback()!=null) {

        // callback type
        try {
            executeResponseCallback(new Runnable() {
                @Override
                public void run() {
                    if (xxlRpcResponse.getErrorMsg() != null) {
                        futureResponse.getInvokeCallback().onFailure(new XxlRpcException(xxlRpcResponse.getErrorMsg()));
                    } else {
                        futureResponse.getInvokeCallback().onSuccess(xxlRpcResponse.getResult());
                    }
                }
            });
        }catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
    } else {
        // 里面调用lock的notify方法
        futureResponse.setResponse(xxlRpcResponse);
    }

    // do remove
    futureResponsePool.remove(requestId);

}

欢迎加入我的知识星球，一起探讨架构，交流源码。加入方式，长按下方二维码噢：

0c8ffecb802501e77ce7e4761204cc44.png

已在知识星球更新源码解析如下：

799f4edf332601a00e6393efa640fe2f.png

91b351da19a2c058c669c00f0830c919.png

5474b486857eb832d6788543ac83dbf4.png

6f21c011f88bc17c8d55935f87afa31b.png

最近更新《芋道 SpringBoot 2.X 入门》系列，已经 101 余篇，覆盖了 MyBatis、Redis、MongoDB、ES、分库分表、读写分离、SpringMVC、Webflux、权限、WebSocket、Dubbo、RabbitMQ、RocketMQ、Kafka、性能测试等等内容。

提供近 3W 行代码的 SpringBoot 示例，以及超 6W 行代码的电商微服务项目。

获取方式：点“在看”，关注公众号并回复 666 领取，更多内容陆续奉上。

文章有帮助的话，在看，转发吧。
谢谢支持哟 (*^__^*）
# 分布式 # spring # java # spring boot # android
Logo
华为云开发者联盟

为开发者提供学习成长、分享交流、生态实践、资源工具等服务，帮助开发者快速成长。

更多推荐

cover

一文为你深度解析LLaMA2模型架构

avatar 华为云开发者联盟
cover

华为云HCSD校园沙龙走进大连理工大学！

avatar 华为云开发者联盟
cover

全球软件技术峰会：数据库峰会

avatar 华为云开发者联盟