xxl-job源码之执行器执行任务的核心流程

Love The Way You Lie 2022-11-09 11:24 264阅读 0赞

# xxl-job源码之执行器执行任务的核心流程 #

## 先来一张图大致看看 ##

如果看的不清晰，可以前往地址：https://www.processon.com/view/link/604e0b860791291f25613424  
密码： xiaowan  
绿色那一块就是执行器的流程  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NtcXdhbg_size_16_color_FFFFFF_t_70]

## 看看启动后的控制台有什么 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NtcXdhbg_size_16_color_FFFFFF_t_70 1]

"nioEventLoopGroup-2-1"@6,081 in group "main": RUNNING
    "nioEventLoopGroup-3-1"@6,231 in group "main": RUNNING
    "nioEventLoopGroup-3-2"@6,424 in group "main": RUNNING
    
    "xxl-job, executor ExecutorRegistryThread"@6,099 in group "main": SLEEPING
    "xxl-job, executor JobLogFileCleanThread"@5,708 in group "main": SLEEPING
    "xxl-job, executor TriggerCallbackThread"@5,711 in group "main": WAIT
    "xxl-rpc, EmbedServer bizThreadPool-632554680"@6,439 in group "main": WAIT

## 启动 ##

对于不习惯SpringBoot，官方提供了2个demo执行器，一个是xxl-job-executor-sample-springboot，一个是xxl-job-executor-sample-frameless。在2个都能运行的情况下，我们就使用 xxl-job-executor-sample-frameless 吧，框架依赖的少，那么对于我们来说，需要屏蔽无用的内容就少。

这个目录就3个源码文件，肯定是猜测启动一个服务，然后将执行器的配置文件数据读入，最后与服务端进行通讯。

最后就锁定 FrameLessXxlJobConfig ，里面其他方法，稍微过一眼，核心就在 FrameLessXxlJobConfig\#initXxlJobExecutor–>xxlJobExecutor.start()，而这个方法里面前面一堆set方法，肯定最后落到 start方法上  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NtcXdhbg_size_16_color_FFFFFF_t_70 2]

### XxlJobSimpleExecutor\#start ###

public void start() { 
    
            // init JobHandler Repository (for method)
            // 这里找到所有的 @xxl-job 注解方法，然后放到本地map中
            initJobHandlerMethodRepository(xxlJobBeanList);
    
            // super start
            try { 
                // 真正启动
                super.start();
            } catch (Exception e) { 
                throw new RuntimeException(e);
            }
        }

### XxlJobExecutor\#start ###

// ---------------------- start + stop ----------------------
    public void start() throws Exception { 
    
            // init logpath
            // 执行日志存放的目录，有配置则用配置，没有就用默认
            XxlJobFileAppender.initLogPath(logPath);
    
            // init invoker, admin-client
            // 就是在配置文件里的admin地址，
            // 在这里封装成 adminBiz 对象 
            initAdminBizList(adminAddresses, accessToken);
    
            // init JobLogFileCleanThread
            // 看名字是job留下的日志 定时清除线程
            JobLogFileCleanThread.getInstance().start(logRetentionDays);
    
            // init TriggerCallbackThread
            // 初始化 执行回调 线程，
            // 看名字是这样，一眼扫过去，扫不出结果
            // 等会细看下
            TriggerCallbackThread.getInstance().start();
    
            // init executor-server
            // 这个应该是起了一个服务，供admin调度中心调用
            // ip和端口都是我们之前配置的
            // 粗看是使用netty，等会详细看下
            initEmbedServer(address, ip, port, appname, accessToken);
        }
    	
    	    private void initEmbedServer(String address, String ip, int port, String appname, String accessToken) throws Exception { 
            // start
            // 这里就是单纯new
            embedServer = new EmbedServer();
            // 这里是核心代码
            embedServer.start(address, port, appname, accessToken);
        }

## EmbedServer\#start ##

可以看到这里启动了一个netty的服务，对于接收到的请求，最后都会交给 EmbedHttpServerHandler 这个处理。

EmbedHttpServerHandler 负责去序列化收到的消息内容，最后交给我们的 ExecutorBizImpl 处理。

如果我们收到一个run方法，那么最后就是走向 ExecutorBizImpl\#run

public void start(final String address, final int port, final String appname, final String accessToken) { 
            executorBiz = new ExecutorBizImpl();
            // 开启一个线程
            thread = new Thread(new Runnable() { 
    
                @Override
                public void run() { 
    
                    // param
                    EventLoopGroup bossGroup = new NioEventLoopGroup();
                    EventLoopGroup workerGroup = new NioEventLoopGroup();
                    // 这里设置 core = 0，来一个线程就会创建一个
                    ThreadPoolExecutor bizThreadPool = new ThreadPoolExecutor(
                            0,
                            200,
                            60L,
                            TimeUnit.SECONDS,
                            new LinkedBlockingQueue<Runnable>(2000),
                            new ThreadFactory() { 
                                @Override
                                public Thread newThread(Runnable r) { 
                                    return new Thread(r, "xxl-rpc, EmbedServer bizThreadPool-" + r.hashCode());
                                }
                            },
                            new RejectedExecutionHandler() { 
                                @Override
                                public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { 
                                    throw new RuntimeException("xxl-job, EmbedServer bizThreadPool is EXHAUSTED!");
                                }
                            });
    
    
                    try { 
                        // 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
                                                // 这个挂的就是处理http请求
                                                .addLast(new EmbedHttpServerHandler(executorBiz, accessToken, bizThreadPool));
                                    }
                                })
                                .childOption(ChannelOption.SO_KEEPALIVE, true);
    
                        // bind
                        ChannelFuture future = bootstrap.bind(port).sync();
    
                        logger.info(">>>>>>>>>>> xxl-job remoting server start success, nettype = {}, port = {}", EmbedServer.class, port);
    
                        // start registry
                        startRegistry(appname, address);
    
                        // wait util stop
                        future.channel().closeFuture().sync();
    
                    } catch (InterruptedException e) { 
                        if (e instanceof InterruptedException) { 
                            logger.info(">>>>>>>>>>> xxl-job remoting server stop.");
                        } else { 
                            logger.error(">>>>>>>>>>> xxl-job remoting server error.", e);
                        }
                    } finally { 
                        // stop
                        try { 
                            workerGroup.shutdownGracefully();
                            bossGroup.shutdownGracefully();
                        } catch (Exception e) { 
                            logger.error(e.getMessage(), e);
                        }
                    }
    
                }
    
            });
            thread.setDaemon(true);	// daemon, service jvm, user thread leave >>> daemon leave >>> jvm leave
            thread.start();
        }

## ExecutorBizImpl\#run ##

这里面拿到对应的 IJobHandler，交给我们的JobThread去进行处理

如果我们设置了超时时间，那么这里的 IJobHandler 又会再包一层thread，设置超时时间进行执行

## 源码分析 ##

[分布式作业调度(定时任务)系统xxl-job快速上手及高级功能简述][xxl-job]

[xxl-job源码之admin调度中心的线程们][xxl-job_admin]

[xxl-job源码之执行器执行任务的核心流程][xxl-job 1]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NtcXdhbg_size_16_color_FFFFFF_t_70]: /images/20221023/b160bee9c04440148dadceb96c1a8b84.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NtcXdhbg_size_16_color_FFFFFF_t_70 1]: /images/20221023/88696e5578a14031b7f1e7b8da821a55.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NtcXdhbg_size_16_color_FFFFFF_t_70 2]: /images/20221023/8c018717dfea437d9bad0906ead7f304.png
[xxl-job]: https://lizhao.blog.csdn.net/article/details/114760941
[xxl-job_admin]: https://lizhao.blog.csdn.net/article/details/114786267
[xxl-job 1]: https://lizhao.blog.csdn.net/article/details/114805846