【自省】线程池里的定时任务跑的可欢了，可咋停掉特定的任务？

开启成长之旅！这是我参与「日新方案 12 月更文应战」的第 18 天，点击查看活动详情

我是石页兄，本篇不带情感，只聊干货

欢迎重视微信大众号「架构染色」沟通和学习

一、布景

在《分布式锁自动续期的入门级完成-自省 | 简约而不简略》中经过【自省】的办法讨论了关于分布式锁自动续期功用的入门级完成办法，简略同步一下上下文：

客户端抢到分布式锁之后开始履行使命，履行完毕后再开释分布式锁。
持锁后因客户端反常未能把锁开释，会导致锁成为永恒锁。
为了防止这种状况，在创建锁的时分给锁指定一个过期时刻。
到期之后锁会被自动删去掉，这个角度看是对锁资源的一种维护。
重点：但若锁过期被删去后，使命还没完毕怎样办？
能够经过在一个额外的线程中自动推迟分布式锁的过期时刻，下文也用续期一词来表述；防止当使命还没履行完，锁就被删去了。
但当分布式锁许多的状况下，每个锁都装备一个线程着实浪费，所以是否能够用线程池里的守时使命呢？

在《【自省】运用Executors.xxx违背阿里Java代码规范，那还不写守时使命了？》中仍然经过【自省】的办法讨论了也能够运用 ScheduledExecutorService#scheduleAtFixedRate来完成守时使命，它的运行机制大概是这样：

假如上一个使命的履行时刻大于等待时刻，使命完毕后，下一个使命马上履行。
假如上一个使命的履行时刻小于等待时刻，使命完毕后，下一个使命在（间隔时刻-履行时刻）后开始履行。

在分布式锁自动续期的场景下，它也满意如下规律：

二、理还乱?

用 ScheduledExecutorService#scheduleAtFixedRate逻辑看很简略，也很清晰，但任何事情都有两面性，把使命丢给线程池的办法，完成起来天然简略清晰，但肯定也有坏处。假如要把锁的功用做的强健，总要从不断地自我质疑、自我反思中，理顺思路，寻觅答案，我以为这属于自省式学习，今后也想测验这种形式，一同再看看有啥问题：

问题：锁自动开释的时分，续期的使命要封闭嘛？

是的，当锁被用户自动封闭的时分，自动续期的使命是要自动撤销掉的。
问题：假如我不自动撤销呢？

对于不自动续期的锁，抢锁后装备一个合适的过期时刻，到期之后锁天然会被开释；这种状况下，客户端本就没有续期使命需要撤销。但假如有额外的线程|线程池在守时续期的话，锁用完了需要被开释掉，使命一定要自动撤销掉。
问题：可万一忘记了呢？

有加锁解锁的代码模板，依照模板来；获取锁之后，在finally中履行开释锁的操作。
```
boolean lockResult = lockInstance.tryLock()；
if(lockResult){
    //do work
}finally{
    lockInstance.unLock();
}
```
万一程序反常崩了，没履行finally呢？

假如程序反常崩了，进程消失后，进程内的资源天然就都开释掉了：续期使命没有了，续期的线程|线程池也没有了。但锁资源就需要依赖锁服务，如 Redis ，在锁过期后自动开释掉锁资源。
问题：关于中止使命，在前文独立线程的完成办法中，有介绍可经过中止机制；但是线程池里的使命怎样撤销呢？

遇事不决问百度，排名第一必有解

咱得本意是撤销一个使命，示例给出的办法是要把线程池关掉。
问题：撤销一个使命，要把整个线程池都关掉？

依照示例所给的办法是不可的，每个使命的撤销，都要封闭整个线程池的话，若给每个使命都配有独立的撤销才能，就需要给每个使命都配一个独立的线程池，这就跟每个锁配一个独立的线程没有区别了。
问题：目标是多个使命同享一个线程池，怎样不封闭线程池而只封闭特定的使命呢？

百度出来跟问题相关的文章本就不多，而多数文章供给的奇思妙招并不好使，笔者是浪费了一些时刻的，但不能再耽搁读者朋友的时刻，直接给思路：解铃还须系铃人，scheduleAtFixedRate的返回值是是ScheduledFuture。
问题：看到 xxxFuture 是否想能想起Future接口的才能？

猜测熟悉 get()办法的同学应该特别多，但不知道熟不熟悉cancel办法，假如看到这个办法感到惊喜，欢迎留言互动。
```
public interface Future<V> {
    boolean cancel(boolean mayInterruptIfRunning);
    ...
    V get() throws InterruptedException, ExecutionException;
    ...
}
```

问题：cancel办法好使嘛？

不看理论看实作用，试试看：

public static void testCancel() throws InterruptedException {
    ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
    System.out.println(" start : " + LocalDateTime.now().format(DateTimeFormatter.ISO_LOCAL_DATE_TIME));
    ScheduledFuture<?> scheduledFuture = scheduledExecutorService.scheduleAtFixedRate(() -> {
        System.out.println("  work : " + LocalDateTime.now().format(DateTimeFormatter.ISO_LOCAL_DATE_TIME));
    }, 5, 5, TimeUnit.SECONDS);
    TimeUnit.SECONDS.sleep(15);
    scheduledFuture.cancel(true);
    System.out.println("cancel : " + LocalDateTime.now().format(DateTimeFormatter.ISO_LOCAL_DATE_TIME));
    TimeUnit.SECONDS.sleep(30);
}

作用满意预期，成功撤销了。

 start : 2022-12-10T19:24:31.508
  work : 2022-12-10T19:24:36.538
  work : 2022-12-10T19:24:41.539
  work : 2022-12-10T19:24:46.541
cancel : 2022-12-10T19:24:46.541 //成功撤销

问题：cancel 里都做了什么呢？

看源码可知，其内有两层核心逻辑：

测验撤销正在履行的使命
防止使命再被守时履行

public boolean cancel(boolean mayInterruptIfRunning) {
    // 1. 先调用父类FutureTask#cancel来撤销使命。
    boolean cancelled = super.cancel(mayInterruptIfRunning);
    //2. 核心逻辑是从队列中删去该使命。
    if (cancelled && removeOnCancel && heapIndex >= 0)
        remove(this);
    return cancelled;
}

至此，关于运用线程池来履行|撤销续期使命，看起来现已没啥问题了；美丽的心境应该是这样的。

OK，略微高兴一下就好，还有问题呢，不想划开的，咱不等了。

三、新的考虑

问题：cancel的参数mayInterruptIfRunning 是什么意思？

从父类cancel办法的注释中能够寻觅到答案，假如是 true 的话，即代表测验经过中止的办法来中止使命

If the task has already started, then the mayInterruptIfRunning parameter determines whether the thread executing this task should be interrupted in an attempt to stop the task.
```
@Override
public boolean cancel(boolean mayInterruptIfRunning) {
    if (RESULT_UPDATER.compareAndSet(this, null, CANCELLATION_CAUSE_HOLDER)) {
        if (checkNotifyWaiters()) {
            notifyListeners();
        }
        return true;
    }
    return false;
}
```
问题：那就是说也或许抛出 InterruptedException 了？

假如是抛出 InterruptedException ，示例中，并未看到程序测验有反常中止，也未看到有反常日志信息。

问题：怎样有点玄学了，还能不是interrupt机制？

在使命内测验捕获一下看看：

public static void testExceptionCatch() throws InterruptedException {
    ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
    ScheduledFuture<?> scheduledFuture = null;
    System.out.println(" start : " + LocalDateTime.now().format(DateTimeFormatter.ISO_LOCAL_DATE_TIME));
    try {
        scheduledFuture = scheduledExecutorService.scheduleAtFixedRate(() -> {
            System.out.println("  work : " + LocalDateTime.now().format(DateTimeFormatter.ISO_LOCAL_DATE_TIME));
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            //throw new RuntimeException("");
        }, 5, 5, TimeUnit.SECONDS);
    }catch (Exception exp){
        exp.printStackTrace();
    }
    TimeUnit.SECONDS.sleep(15);
    scheduledFuture.cancel(true);
    System.out.println("cancel : " + LocalDateTime.now().format(DateTimeFormatter.ISO_LOCAL_DATE_TIME));
    TimeUnit.SECONDS.sleep(30);
}

成果中的信息 java.lang.InterruptedException: sleep interrupted 能够明确是使命内的逻辑是可经过中止机制完成的。

start : 2022-12-10T20:10:31.248
  work : 2022-12-10T20:10:36.276
  work : 2022-12-10T20:10:41.272
  work : 2022-12-10T20:10:46.277
cancel : 2022-12-10T20:10:46.277
java.lang.InterruptedException: sleep interrupted
        at java.lang.Thread.sleep(Native Method)
        at java.lang.Thread.sleep(Thread.java:340)
        at java.util.concurrent.TimeUnit.sleep(TimeUnit.java:386)
        at com.wushiyii.lock.ScheduleTest.lambda$testExceptionCatch$1(ScheduleTest.java:39)
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
        at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308)
        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)

问题：之前实例中撤销使命时，外部也无反常信息，线程池内部留着这个反常干嘛了呢？

直接抛出反常试试看

public static void testExceptionCatch() throws InterruptedException {
    ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
    ScheduledFuture<?> scheduledFuture = null;
    System.out.println(" start : " + LocalDateTime.now().format(DateTimeFormatter.ISO_LOCAL_DATE_TIME));
    try {
        scheduledFuture = scheduledExecutorService.scheduleAtFixedRate(() -> {
            System.out.println("  work : " + LocalDateTime.now().format(DateTimeFormatter.ISO_LOCAL_DATE_TIME));
            throw new RuntimeException("just throw ");
            //throw new RuntimeException("");
        }, 5, 5, TimeUnit.SECONDS);
    }catch (Exception exp){
        exp.printStackTrace();
    }
    TimeUnit.SECONDS.sleep(15);
    scheduledFuture.cancel(true);
    System.out.println("cancel : " + LocalDateTime.now().format(DateTimeFormatter.ISO_LOCAL_DATE_TIME));
    TimeUnit.SECONDS.sleep(30);
}

仔细观察能看出，成果变的有意思了，work只履行了一次，前文中的履行成果中work都履行了3次，这里却只履行了一次。

 start : 2022-12-10T20:16:53.285
  work : 2022-12-10T20:16:58.307
cancel : 2022-12-10T20:17:08.305

问题：使命内抛出反常能导致守时使命失去守时履行的才能？

是的，运用scheduleAtFixedRate有以下几个状况有必要留意：
1. 使命逻辑中未捕获的反常能导致本该守时履行的使命，后续不再履行。
2. 使命逻辑中未捕获的反常不会外抛，外部感知不到。
3. 使命逻辑中的反常，需在使命逻辑内捕获并记录，否则无处可知。
看起来守时使命的运用的确是不能为所欲为的，毕竟大美也总是会说：

问题：那还有什么留意事项？

给线程池指定的线程数要合理，不要无限制的提交使命，也不要每提交一个使命就new一个线程池。还有…

老板亲情提示：面都坨了，快点吃，否则不好吃了。

嗯，【自省】也不能饿肚子，该吃饭了。

假如您还知道一些需要留意的奥妙机制，欢迎留言讨论；咱们下一篇再聊。

四、最终说一句

我是石页兄，假如这篇文章对您有协助，或许有所启示的话，欢迎重视笔者的微信大众号【架构染色】进行沟通和学习。您的支撑是我坚持写作最大的动力。

欢迎点击链接扫马儿重视、沟通。

声明：本站所有文章，如无特殊说明或标注，均为本站原创发布。任何个人或组织，在未征得本站同意时，禁止复制、盗用、采集、发布本站内容到任何网站、书籍等各类媒体平台。如若本站内容侵犯了原著者的合法权益，可联系我们进行处理。

【自省】线程池里的定时任务跑的可欢了，可咋停掉特定的任务？

一、布景

二、理还乱?

三、新的考虑

四、最终说一句

近期文章

近期评论