java高并发系列 - 第19天：JUC中的Executor框架详解1，再不了解你就out啦

3. 介绍ExecutorService

4. 介绍线程池ThreadPoolExecutor及案例

5. 介绍定时器ScheduledExecutorService及案例

6. 介绍Excecutors类的使用

7. 介绍Future接口

8. 介绍Callable接口

9. 介绍FutureTask的使用

10. 获取异步任务的执行结果的几种方法

Executors框架介绍

Executors框架是Doug Lea的神作，通过这个框架，可以很容易的使用线程池高效地处理并行任务。

Excecutor框架主要包含3部分的内容：

1. 任务相关的：包含被执行的任务要实现的接口：Runnable接口或Callable接口

2. 任务的执行相关的：包含任务执行机制的核心接口Executor，以及继承自Executor的ExecutorService接口。Executor框架中有两个关键的类实现了ExecutorService接口（ThreadPoolExecutor和ScheduleThreadPoolExecutor）

3. 异步计算结果相关的：包含接口Future和实现Future接口的FutureTask类

Executors框架包括：

• Executor

• ExecutorService

• ThreadPoolExecutor

• Executors

• Future

• Callable

• FutureTask

• CompletableFuture

• CompletionService

• ExecutorCompletionService

下面我们来一个个介绍其用途和使用方法。

Executor接口

---

Executor接口中定义了方法execute(Runable able)接口，该方法接受一个Runable实例，他来执行一个任务，任务即实现一个Runable接口的类。

ExecutorService接口

---

ExecutorService继承于Executor接口，他提供了更为丰富的线程实现方法，比如ExecutorService提供关闭自己的方法，以及为跟踪一个或多个异步任务执行状况而生成Future的方法。

ExecutorService有三种状态：运行、关闭、终止。创建后便进入运行状态，当调用了shutdown()方法时，便进入了关闭状态，此时意味着ExecutorService不再接受新的任务，但是他还是会执行已经提交的任务，当所有已经提交了的任务执行完后，便达到终止状态。如果不调用shutdown方法，ExecutorService方法会一直运行下去，系统一般不会主动关闭。

ThreadPoolExecutor类

---

线程池类，实现了ExecutorService接口中所有方法，该类也是我们经常要用到的，非常重要，关于此类有详细的介绍，可以移步：[玩转java中的线程池]

ScheduleThreadPoolExecutor定时器

---

ScheduleThreadPoolExecutor继承自ScheduleThreadPoolExecutor，他主要用来延迟执行任务，或者定时执行任务。功能和Timer类似，但是ScheduleThreadPoolExecutor更强大、更灵活一些。Timer后台是单个线程，而ScheduleThreadPoolExecutor可以在创建的时候指定多个线程。

常用方法介绍：

schedule:延迟执行任务1次

使用ScheduleThreadPoolExecutor的schedule方法，看一下这个方法的声明：

public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit)

3个参数：

command：需要执行的任务

delay：需要延迟的时间

unit：参数2的时间单位，是个枚举，可以是天、小时、分钟、秒、毫秒、纳秒等

示例代码：

package com.itsoku.chat18;

import java.util.concurrent.ExecutionException;

import java.util.concurrent.Executors;

import java.util.concurrent.ScheduledExecutorService;

import java.util.concurrent.TimeUnit;

/**

* 跟着阿里p7学并发，微信公众号：javacode2018

*/

public class Demo1 {

public static void main(String[] args) throws ExecutionException, InterruptedException {

System.out.println(System.currentTimeMillis());

ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(10);

scheduledExecutorService.schedule(() -> {

System.out.println(System.currentTimeMillis() + “开始执行”);

//模拟任务耗时

try {

TimeUnit.SECONDS.sleep(3);

} catch (InterruptedException e) {

e.printStackTrace();

}

System.out.println(System.currentTimeMillis() + “执行结束”);

}, 2, TimeUnit.SECONDS);

}

}

输出：

1564575180457

1564575185525开始执行

1564575188530执行结束

scheduleAtFixedRate:固定的频率执行任务

使用ScheduleThreadPoolExecutor的scheduleAtFixedRate方法，该方法设置了执行周期，下一次执行时间相当于是上一次的执行时间加上period，任务每次执行完毕之后才会计算下次的执行时间。

看一下这个方法的声明：

public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,

long initialDelay,

long period,

TimeUnit unit);

4个参数：

command：表示要执行的任务

initialDelay：表示延迟多久执行第一次

period：连续执行之间的时间间隔

unit：参数2和参数3的时间单位，是个枚举，可以是天、小时、分钟、秒、毫秒、纳秒等

假设系统调用scheduleAtFixedRate的时间是T1，那么执行时间如下：

第1次：T1+initialDelay

第2次：T1+initialDelay+period

第3次：T1+initialDelay+2*period

第n次：T1+initialDelay+(n-1)*period

示例代码：

package com.itsoku.chat18;

import java.sql.Time;

import java.util.concurrent.*;

import java.util.concurrent.atomic.AtomicInteger;

/**

* 跟着阿里p7学并发，微信公众号：javacode2018

*/

public class Demo2 {

public static void main(String[] args) throws ExecutionException, InterruptedException {

System.out.println(System.currentTimeMillis());

//任务执行计数器

AtomicInteger count = new AtomicInteger(1);

ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(10);

scheduledExecutorService.scheduleAtFixedRate(() -> {

int currCount = count.getAndIncrement();

System.out.println(Thread.currentThread().getName());

System.out.println(System.currentTimeMillis() + “第” + currCount + “次” + “开始执行”);

try {

TimeUnit.SECONDS.sleep(2);

} catch (InterruptedException e) {

e.printStackTrace();

}

System.out.println(System.currentTimeMillis() + “第” + currCount + “次” + “执行结束”);

}, 1, 1, TimeUnit.SECONDS);

}

}

前面6次输出结果：

1564576404181

pool-1-thread-1

1564576405247第1次开始执行

1564576407251第1次执行结束

pool-1-thread-1

1564576407251第2次开始执行

1564576409252第2次执行结束

pool-1-thread-2

1564576409252第3次开始执行

1564576411255第3次执行结束

pool-1-thread-1

1564576411256第4次开始执行

1564576413260第4次执行结束

pool-1-thread-3

1564576413260第5次开始执行

1564576415265第5次执行结束

pool-1-thread-2

1564576415266第6次开始执行

1564576417269第6次执行结束

代码中设置的任务第一次执行时间是系统启动之后延迟一秒执行。后面每次时间间隔1秒，从输出中可以看出系统启动之后过了1秒任务第一次执行（1、3行输出），输出的结果中可以看到任务第一次执行结束时间和第二次的结束时间一样，为什么会这样？前面有介绍，任务当前执行完毕之后会计算下次执行时间，下次执行时间为上次执行的开始时间+period，第一次开始执行时间是1564576405247，加1秒为1564576406247，这个时间小于第一次结束的时间了，说明小于系统当前时间了，会立即执行。

scheduleWithFixedDelay:固定的间隔执行任务

使用ScheduleThreadPoolExecutor的scheduleWithFixedDelay方法，该方法设置了执行周期，与scheduleAtFixedRate方法不同的是，下一次执行时间是上一次任务执行完的系统时间加上period，因而具体执行时间不是固定的，但周期是固定的，是采用相对固定的延迟来执行任务。看一下这个方法的声明：

public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,

long initialDelay,

long delay,

TimeUnit unit);

4个参数：

command：表示要执行的任务

initialDelay：表示延迟多久执行第一次

period：表示下次执行时间和上次执行结束时间之间的间隔时间

unit：参数2和参数3的时间单位，是个枚举，可以是天、小时、分钟、秒、毫秒、纳秒等

假设系统调用scheduleAtFixedRate的时间是T1，那么执行时间如下：

第1次：T1+initialDelay，执行结束时间：E1

第2次：E1+period，执行结束时间：E2

第3次：E2+period，执行结束时间：E3

第4次：E3+period，执行结束时间：E4

第n次：上次执行结束时间+period

示例代码：

package com.itsoku.chat18;

import java.util.concurrent.*;

import java.util.concurrent.atomic.AtomicInteger;

/**

* 跟着阿里p7学并发，微信公众号：javacode2018

*/

public class Demo3 {

public static void main(String[] args) throws ExecutionException, InterruptedException {

System.out.println(System.currentTimeMillis());

//任务执行计数器

AtomicInteger count = new AtomicInteger(1);

ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(10);

scheduledExecutorService.scheduleWithFixedDelay(() -> {

int currCount = count.getAndIncrement();

System.out.println(Thread.currentThread().getName());

System.out.println(System.currentTimeMillis() + “第” + currCount + “次” + “开始执行”);

try {

TimeUnit.SECONDS.sleep(2);

} catch (InterruptedException e) {

e.printStackTrace();

}

System.out.println(System.currentTimeMillis() + “第” + currCount + “次” + “执行结束”);

}, 1, 3, TimeUnit.SECONDS);

}

}

前几次输出如下：

1564578510983

pool-1-thread-1

1564578512087第1次开始执行

1564578514091第1次执行结束

pool-1-thread-1

1564578517096第2次开始执行

1564578519100第2次执行结束

pool-1-thread-2

1564578522103第3次开始执行

1564578524105第3次执行结束

pool-1-thread-1

1564578527106第4次开始执行

1564578529106第4次执行结束

延迟1秒之后执行第1次，后面每次的执行时间和上次执行结束时间间隔3秒。

scheduleAtFixedRate和scheduleWithFixedDelay示例建议多看2遍。

定时任务有异常会怎么样？

示例代码：

package com.itsoku.chat18;

import java.util.concurrent.*;

import java.util.concurrent.atomic.AtomicInteger;

/**

* 跟着阿里p7学并发，微信公众号：javacode2018

*/

public class Demo4 {

public static void main(String[] args) throws ExecutionException, InterruptedException {

System.out.println(System.currentTimeMillis());

//任务执行计数器

AtomicInteger count = new AtomicInteger(1);

ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(10);

ScheduledFuture<?> scheduledFuture = scheduledExecutorService.scheduleWithFixedDelay(() -> {

int currCount = count.getAndIncrement();

System.out.println(Thread.currentThread().getName());

System.out.println(System.currentTimeMillis() + “第” + currCount + “次” + “开始执行”);

System.out.println(10 / 0);

System.out.println(System.currentTimeMillis() + “第” + currCount + “次” + “执行结束”);

}, 1, 1, TimeUnit.SECONDS);

TimeUnit.SECONDS.sleep(5);

System.out.println(scheduledFuture.isCancelled());

System.out.println(scheduledFuture.isDone());

}

}

系统输出如下内容就再也没有输出了：

1564578848143

pool-1-thread-1

1564578849226第1次开始执行

false

true

先说补充点知识：schedule、scheduleAtFixedRate、scheduleWithFixedDelay这几个方法有个返回值ScheduledFuture，通过ScheduledFuture可以对执行的任务做一些操作，如判断任务是否被取消、是否执行完成。

再回到上面代码，任务中有个10/0的操作，会触发异常，发生异常之后没有任何现象，被ScheduledExecutorService内部给吞掉了，然后这个任务再也不会执行了，scheduledFuture.isDone()输出true，表示这个任务已经结束了，再也不会被执行了。所以如果程序有异常，开发者自己注意处理一下，不然跑着跑着发现任务怎么不跑了，也没有异常输出。

取消定时任务的执行

可能任务执行一会，想取消执行，可以调用ScheduledFuture的cancel方法，参数表示是否给任务发送中断信号。

package com.itsoku.chat18;

import java.util.concurrent.*;

import java.util.concurrent.atomic.AtomicInteger;

/**

* 跟着阿里p7学并发，微信公众号：javacode2018

*/

public class Demo5 {

public static void main(String[] args) throws ExecutionException, InterruptedException {

System.out.println(System.currentTimeMillis());

//任务执行计数器

AtomicInteger count = new AtomicInteger(1);

ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);

ScheduledFuture<?> scheduledFuture = scheduledExecutorService.scheduleWithFixedDelay(() -> {

int currCount = count.getAndIncrement();

System.out.println(Thread.currentThread().getName());

System.out.println(System.currentTimeMillis() + “第” + currCount + “次” + “开始执行”);

try {

TimeUnit.SECONDS.sleep(2);

} catch (InterruptedException e) {

e.printStackTrace();

}

System.out.println(System.currentTimeMillis() + “第” + currCount + “次” + “执行结束”);

}, 1, 1, TimeUnit.SECONDS);

TimeUnit.SECONDS.sleep(5);

scheduledFuture.cancel(false);

TimeUnit.SECONDS.sleep(1);

System.out.println(“任务是否被取消：”+scheduledFuture.isCancelled());

System.out.println(“任务是否已完成：”+scheduledFuture.isDone());

}

}

输出：

1564579843190

pool-1-thread-1

1564579844255第1次开始执行

1564579846260第1次执行结束

pool-1-thread-1

1564579847263第2次开始执行

任务是否被取消：true

任务是否已完成：true

1564579849267第2次执行结束

输出中可以看到任务被取消成功了。

Executors类

---

Executors类，提供了一系列工厂方法用于创建线程池，返回的线程池都实现了ExecutorService接口。常用的方法有：

newSingleThreadExecutor

public static ExecutorService newSingleThreadExecutor()

public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory)

自我介绍一下，小编13年上海交大毕业，曾经在小公司待过，也去过华为、OPPO等大厂，18年进入阿里一直到现在。

深知大多数Java工程师，想要提升技能，往往是自己摸索成长或者是报班学习，但对于培训机构动则几千的学费，着实压力不小。自己不成体系的自学效果低效又漫长，而且极易碰到天花板技术停滞不前！

因此收集整理了一份《2024年Java开发全套学习资料》，初衷也很简单，就是希望能够帮助到想自学提升又不知道该从何学起的朋友，同时减轻大家的负担。

既有适合小白学习的零基础资料，也有适合3年以上经验的小伙伴深入学习提升的进阶课程，基本涵盖了95%以上Java开发知识点，真正体系化！

由于文件比较大，这里只是将部分目录大纲截图出来，每个节点里面都包含大厂面经、学习笔记、源码讲义、实战项目、讲解视频，并且后续会持续更新

如果你觉得这些内容对你有帮助，可以添加V获取：vip1024b （备注Java）

Kafka实战笔记

关于这份笔记，为了不影响大家的阅读体验，我只能在文章中展示部分的章节内容和核心截图

• Kafka入门
• 为什么选择Kafka
• Karka的安装、管理和配置

• Kafka的集群
• 第一个Kafka程序
• 

afka的生产者

• Kafka的消费者
• 深入理解Kafka
• 可靠的数据传递

• Spring和Kalka的整合
• Sprinboot和Kafka的整合
• Kafka实战之削峰填谷
• 数据管道和流式处理(了解即可)

• Kafka实战之削峰填谷

一个人可以走的很快，但一群人才能走的更远。如果你从事以下工作或对以下感兴趣，欢迎戳这里加入程序员的圈子，让我们一起学习成长！

AI人工智能、Android移动开发、AIGC大模型、C C#、Go语言、Java、Linux运维、云计算、MySQL、PMP、网络安全、Python爬虫、UE5、UI设计、Unity3D、Web前端开发、产品经理、车载开发、大数据、鸿蒙、计算机网络、嵌入式物联网、软件测试、数据结构与算法、音视频开发、Flutter、IOS开发、PHP开发、.NET、安卓逆向、云计算

8541)]

• Kafka入门
• 为什么选择Kafka
• Karka的安装、管理和配置

[外链图片转存中…(img-Q1LDPFjy-1712169968541)]

• Kafka的集群
• 第一个Kafka程序
• [外链图片转存中…(img-VroUJZrY-1712169968542)]

afka的生产者

[外链图片转存中…(img-IaPKzzMx-1712169968542)]

• Kafka的消费者
• 深入理解Kafka
• 可靠的数据传递

[外链图片转存中…(img-aFPt8Ssm-1712169968542)]

[外链图片转存中…(img-XKZyz8lD-1712169968542)]

• Spring和Kalka的整合
• Sprinboot和Kafka的整合
• Kafka实战之削峰填谷
• 数据管道和流式处理(了解即可)

[外链图片转存中…(img-FvCgsinh-1712169968543)]

• Kafka实战之削峰填谷

[外链图片转存中…(img-7E6n7J55-1712169968543)]

一个人可以走的很快，但一群人才能走的更远。如果你从事以下工作或对以下感兴趣，欢迎戳这里加入程序员的圈子，让我们一起学习成长！

AI人工智能、Android移动开发、AIGC大模型、C C#、Go语言、Java、Linux运维、云计算、MySQL、PMP、网络安全、Python爬虫、UE5、UI设计、Unity3D、Web前端开发、产品经理、车载开发、大数据、鸿蒙、计算机网络、嵌入式物联网、软件测试、数据结构与算法、音视频开发、Flutter、IOS开发、PHP开发、.NET、安卓逆向、云计算