xxl-job定时调度任务Java代码分析

简介

用xxl-job做后台任务管理, 主要是快速解决定时任务的HA问题, 项目代码量不大, 功能精简, 没有特殊依赖. 因为产品中用到了这个项目, 上午花了点时间研究了一下运行机制. 把看到的记一下.

环境

<!-- http://repo1.maven.org/maven2/com/xuxueli/xxl-job-core/ -->
<dependency>
    <groupId>com.xuxueli</groupId>
    <artifactId>xxl-job-core</artifactId>
    <version>${最新稳定版本}</version>
</dependency>

运行需要 JDK1.8, MySQL5.7

数据库结构

代码结构

项目文件结构如下

├───doc
│   ├───db                                               # 初始化的sql
│   └───images
├───xxl-job-admin                                        # 运行的服务端模块, 提供界面和调度
│   └───src
│       ├───main
│       │   ├───java
│       │   │   └───com
│       │   │       └───xxl
│       │   │           └───job
│       │   │               └───admin
│       │   │                   ├───controller
│       │   │                   │   ├───annotation
│       │   │                   │   ├───interceptor
│       │   │                   │   └───resolver
│       │   │                   ├───core
│       │   │                   ├───dao
│       │   │                   └───service
│       │   │                       └───impl
│       │   └───resources
│       │       ├───i18n                                 # 多国化, 简繁英
│       │       ├───mybatis-mapper                       # xml形式的mapper
│       │       ├───static                               # 前端静态文件
│       │       └───templates                            # Freemarker模板
│       └───test
│           └───java
│
├───xxl-job-core                                         # 公用jar包, 模块内部依赖
│   └───src
│       └───main
│           └───java
│
└───xxl-job-executor-samples
    ├───xxl-job-executor-sample-frameless                # 任务执行层示例
    │   └───src
    │       ├───main
    │       │   ├───java
    │       │   └───resources
    │       └───test
    │           └───java
    └───xxl-job-executor-sample-springboot               # 使用SpringBoot的执行层示例
        └───src
            ├───main
            │   ├───java
            │   └───resources
            └───test

运行机制

执行端需要准备以下信息

非 Spring 的场景

通过调用 FrameLessXxlJobConfig.getInstance().initXxlJobExecutor() 这个方法将 XxlJobSimpleExecutor 实例化, 并注册到xxl_job服务端

Spring 场景

远程调用服务

xxl_job 并未使用Spring的服务机制, 而是内部实现了一个侦听指定IP+端口的服务. 这个实现对应的类是 EmbedServer, 服务基于 Netty, 核心代码是

// start server
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup)
        .channel(NioServerSocketChannel.class)
        .childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            public void initChannel(SocketChannel channel) throws Exception {
                channel.pipeline()
                        .addLast(new IdleStateHandler(0, 0, 30 * 3, TimeUnit.SECONDS))  // beat 3N, close if idle
                        .addLast(new HttpServerCodec())
                        .addLast(new HttpObjectAggregator(5 * 1024 * 1024))  // merge request & reponse to FULL
                        .addLast(new EmbedHttpServerHandler(executorBiz, accessToken, bizThreadPool));
            }
        })
        .childOption(ChannelOption.SO_KEEPALIVE, true);

这行代码注册了内部的XxlJob方法

.addLast(new EmbedHttpServerHandler(executorBiz, accessToken, bizThreadPool)

处理远程请求时, 在下面的代码中, 通过executorBiz.run(triggerParam)调用XxlJob方法

private Object process(HttpMethod httpMethod, String uri, String requestData, String accessTokenReq) {
    //...
    // services mapping
    try {
        switch (uri) {
            case "/beat":
                return executorBiz.beat();
            case "/idleBeat":
                IdleBeatParam idleBeatParam = GsonTool.fromJson(requestData, IdleBeatParam.class);
                return executorBiz.idleBeat(idleBeatParam);
            case "/run":
                TriggerParam triggerParam = GsonTool.fromJson(requestData, TriggerParam.class);
                return executorBiz.run(triggerParam);
            case "/kill":
                KillParam killParam = GsonTool.fromJson(requestData, KillParam.class);
                return executorBiz.kill(killParam);
            case "/log":
                LogParam logParam = GsonTool.fromJson(requestData, LogParam.class);
                return executorBiz.log(logParam);
            default:
                return new ReturnT<String>(ReturnT.FAIL_CODE, "invalid request, uri-mapping(" + uri + ") not found.");
        }
    } catch (Exception e) {
    //...
}

锁机制

通过select ... for update实现的, 这个表并没有放到 MyBatis, 在 JobScheduleHelper 中, 通过

preparedStatement = conn.prepareStatement(  "select * from xxl_job_lock where lock_name = 'schedule_lock' for update" );
preparedStatement.execute();

得到锁, 在方法末尾释放

// close PreparedStatement
if (null != preparedStatement) {
    try {
        preparedStatement.close();
    } catch (SQLException e) {
        if (!scheduleThreadToStop) {
            logger.error(e.getMessage(), e);
        }
    }
}
标签 , ,

发表回复