!!!important 关于守护线程的问题

Question

有一个疑问，工作中使用一个叫 xxl-job 的开源框架做定时任务调度，我看过里面的源码，其实原理就是有好几个守护线程，我举一个例子。代码进过简化处理，完整代码：https://gitee.com/xuxueli0323/xxl-job/blob/master/xxl-job-core/src/main/java/com/xxl/job/core/thread/TriggerCallbackThread.java
这是一个执行器回调守护线程。

PS：我也正是因为使用这个框架出了一些小问题，看了源码，发现大量使用到线程的知识才来看老师的课程充充电，弥补细节。

我的疑问：
（1）课程中提到不要把用户线程设置为守护线程，这个框架中起码有3-4个线程都是守护线程，我有点疑惑了，开发中我们到底应该怎么做？
（2）我在使用这个框架的时候也出现过守护线程异常退出的情况，查了一些资料，整体思路是当程序出现异常会导致守护线程异常退出，方案try-catch捕获异常，也试过还是有异常退出的时候。关于守护线程异常退出老师有什么好的处理方案吗
（3）是否有可以监控到我这个守护线程异常退出的方案、手段，让我能够及时知道我这个守护线程没有运行了，异常退出了。

简化后代码如下：

package com.xxl.job.core.thread;
/**
 * Created by xuxueli on 16/7/22.
 */
public class TriggerCallbackThread {
    private static Logger logger = LoggerFactory.getLogger(TriggerCallbackThread.class);

    private static TriggerCallbackThread instance = new TriggerCallbackThread();
    public static TriggerCallbackThread getInstance(){
        return instance;
    }

    /**
     * job results callback queue
     */
    private LinkedBlockingQueue<HandleCallbackParam> callBackQueue = new LinkedBlockingQueue<HandleCallbackParam>();
    
    public static void pushCallBack(HandleCallbackParam callback){
        getInstance().callBackQueue.add(callback);
        logger.debug(">>>>>>>>>>> xxl-job, push callback request, logId:{}", callback.getLogId());
    }

    /**
     * callback thread
     */
    private Thread triggerCallbackThread;
    private Thread triggerRetryCallbackThread;
    private volatile boolean toStop = false;
    public void start() {

        // valid
        if (XxlJobExecutor.getAdminBizList() == null) {
            logger.warn(">>>>>>>>>>> xxl-job, executor callback config fail, adminAddresses is null.");
            return;
        }

        // callback
        triggerCallbackThread = new Thread(new Runnable() {

            @Override
            public void run() {

                // normal callback
                while(!toStop){
                    try {
                        HandleCallbackParam callback = getInstance().callBackQueue.take();
                        if (callback != null) {

                            // callback list param
                            List<HandleCallbackParam> callbackParamList = new ArrayList<HandleCallbackParam>();
                            int drainToNum = getInstance().callBackQueue.drainTo(callbackParamList);
                            callbackParamList.add(callback);

                            // callback, will retry if error
                            if (callbackParamList!=null && callbackParamList.size()>0) {
                                doCallback(callbackParamList);
                            }
                        }
                    } catch (Exception e) {
                        if (!toStop) {
                            logger.error(e.getMessage(), e);
                        }
                    }
                }

                // last callback
                try {
                    List<HandleCallbackParam> callbackParamList = new ArrayList<HandleCallbackParam>();
                    int drainToNum = getInstance().callBackQueue.drainTo(callbackParamList);
                    if (callbackParamList!=null && callbackParamList.size()>0) {
                        doCallback(callbackParamList);
                    }
                } catch (Exception e) {
                    if (!toStop) {
                        logger.error(e.getMessage(), e);
                    }
                }
                logger.info(">>>>>>>>>>> xxl-job, executor callback thread destory.");

            }
        });
        triggerCallbackThread.setDaemon(true);
        triggerCallbackThread.setName("xxl-job, executor TriggerCallbackThread");
        triggerCallbackThread.start();


        // retry
        triggerRetryCallbackThread = new Thread(new Runnable() {
            @Override
            public void run() {
                while(!toStop){
                    try {
                        retryFailCallbackFile();
                    } catch (Exception e) {
                        if (!toStop) {
                            logger.error(e.getMessage(), e);
                        }

                    }
                    try {
                        TimeUnit.SECONDS.sleep(RegistryConfig.BEAT_TIMEOUT);
                    } catch (InterruptedException e) {
                        if (!toStop) {
                            logger.error(e.getMessage(), e);
                        }
                    }
                }
                logger.info(">>>>>>>>>>> xxl-job, executor retry callback thread destory.");
            }
        });
        triggerRetryCallbackThread.setDaemon(true);
        triggerRetryCallbackThread.start();

    }
    public void toStop(){
        toStop = true;
        // stop callback, interrupt and wait
        if (triggerCallbackThread != null) {    // support empty admin address
            triggerCallbackThread.interrupt();
            try {
                triggerCallbackThread.join();
            } catch (InterruptedException e) {
                logger.error(e.getMessage(), e);
            }
        }

        // stop retry, interrupt and wait
        if (triggerRetryCallbackThread != null) {
            triggerRetryCallbackThread.interrupt();
            try {
                triggerRetryCallbackThread.join();
            } catch (InterruptedException e) {
                logger.error(e.getMessage(), e);
            }
        }

    }
}

Answer 1

悟空 2021-02-17 22:39:45

课程中说的“不要把用户线程设置为守护线程”，针对的是业务场景，不适用于框架，因为框架需要在业务线程都结束后自动结束，所以需要设置为守护线程。

守护线程也是线程，在异常的时候，可以用try catch来处理，但是要注意，使用守护线程一定一定要注意必须做的一个方案：最外层用try包裹住，防止里面一切可能出现但又忘记了的运行时异常发生，从而终止了守护线程。那就影响非常之大了，代码例子：

//启用一个守护线程 去监听延迟队列执行任务Thread t = new Thread(() -> {
    while (true) {
          try {
             // 在这里面书写你的最核心逻辑。。。。里面抛出任何异常，我都不怕了
          } catch (Exception e) {
              log.error("daemon thread for CATEGORY_QUEUE 执行失败", e);
          }
      }
    });

当你使用守护线程去处理逻辑，而必须确保此守护线程不能退出时，请使用try，不允许任何异常抛出来终止守护线程即可。