线程池

我就是我 2023-07-17 11:28 2阅读 0赞

\#\#线程池知识点大纲\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#  
1.Executor  
2.ExecutorService  
3.Future  
4.Callable  
5.Executors  
6.FixedThreadPool  
7.CachedThreadPool  
8.SingleThreadExceutor  
9.ScheduledThreadPool  
10.ForkJoinPool  
11.ThreadPoolExecutor

\#\#线程池知识点详细\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#\#  
1.Executor  
①Executor  
线程池顶级接口。定义方法，void execute(Runnable)。方法是用于处理任务的一个服务方法。  
②代码实现：  
public class Test\_01\_MyExecutor implements Executor \{  
public static void main(String\[\] args) \{  
new Test\_01\_MyExecutor().execute(new Runnable() \{  
public void run() \{  
System.out.println(Thread.currentThread().getName() + " - test executor");  
\}  
\});  
\}  
@Override  
public void execute(Runnable command) \{  
new Thread(command).start();  
\}  
\}

2.ExecutorService  
①ExecutorService【实现这个接口，代表可以提供线程池能力】  
ExecutorService是Executor的子接口：public interface ExecutorService extends Executor \{\}。  
常见方法 -> void execute(Runnable),Future submit(Callable),Future submit(Runnable),shutdown()  
线程池状态：Running,ShuttingDown,Termitnaed  
Running - 线程池正在执行中。活动状态。  
ShuttingDown - 线程池正在关闭过程中。优雅关闭。一旦进入这个状态，线程池不再接收新的任务，处理所已接收的任务，处理完毕后，关闭线程池。  
Terminated - 线程池已经关闭。  
Future:是submit方法的返回值。代表未来，也就是线程执行结束后的一种结果。如返回值。

②代码实现  
public class Test\_02\_FixedThreadPool \{  
public static void main(String\[\] args) \{  
ExecutorService service = Executors.newFixedThreadPool(5);  
for(int i = 0; i < 6; i++)\{  
service.execute(new Runnable() \{  
public void run() \{  
try \{TimeUnit.MILLISECONDS.sleep(500);\}  
catch (InterruptedException e) \{e.printStackTrace();\}  
//System.out.println(Thread.currentThread().getName() + " - test executor");  
\}  
\});  
\}  
// 优雅关机  
service.shutdown();  
// 是否已经结束， 相当于回收了资源。  
System.out.println(service.isTerminated());  
// 是否已经关闭， 是否调用过shutdown方法  
System.out.println(service.isShutdown());  
System.out.println(service);  
\}  
\}

3.Future  
①Future【固定容量线程池】  
未来结果,代表线程任务执行结束后的结果。获取线程执行结果的方式是通过get方法获取的。  
常用方法：T get() T get(long, TimeUnit)  
get无参阻塞等待线程执行结束，并得到结果。  
get有参阻塞固定时长,等待线程执行结束后的结果,如果在阻塞时长范围内,线程未执行结束,抛出异常。

②代码实现:  
public class Test\_03\_Future \{  
public static void main(String\[\] args) throws InterruptedException, ExecutionException \{  
ExecutorService service = Executors.newFixedThreadPool(1); // 1个线程池  
Future future = service.submit(new Callable() \{  
@Override  
public String call() \{  
try \{TimeUnit.MILLISECONDS.sleep(500);\}  
catch (InterruptedException e) \{e.printStackTrace();\}  
System.out.println(“aaa”);  
return Thread.currentThread().getName() + " - test executor";  
\}  
\});  
System.out.println(future.isDone()); // 查看线程是否结束， 任务是否完成。 call方法是否执行结束  
System.out.println(future.get()); // 获取call方法的返回值。  
try \{ System.out.println(future.get(1L,TimeUnit.SECONDS));\}  
catch (TimeoutException e) \{e.printStackTrace();\}  
if(future.isDone()) \{  
service.shutdown();  
\}  
\}  
\}

4.Callable  
类似Runnable接口。也是可以启动一个线程的接口。  
其中call方法的作用和Runnable中的run方法完全一致。  
和Runnable接口的选择->需要返回值或需要抛出异常时,使用Callable,其他情况可任意择。

5.固定容量线程池应用  
把一个大任务，分成5个小任务执行完合并结果返回。代码如：  
public class Test\_04\_ParallelComputingWithFixedThreadPool \{  
public static void main(String\[\] args) throws InterruptedException, ExecutionException \{  
long start = System.currentTimeMillis();  
ExecutorService service = Executors.newFixedThreadPool(5);  
ComputingTask t1 = new ComputingTask(1, 60000);  
ComputingTask t2 = new ComputingTask(60001, 110000);  
ComputingTask t3 = new ComputingTask(110001, 150000);  
ComputingTask t4 = new ComputingTask(150001, 180000);  
ComputingTask t5 = new ComputingTask(180001, 200000);  
Future<List> f1 = service.submit(t1);  
Future<List> f2 = service.submit(t2);  
Future<List> f3 = service.submit(t3);  
Future<List> f4 = service.submit(t4);  
Future<List> f5 = service.submit(t5);  
start = System.currentTimeMillis();  
f1.get();  
f2.get();  
f3.get();  
f4.get();  
f5.get();  
long end = System.currentTimeMillis();  
System.out.println("parallel computing times : " + (end - start));  
\}  
// 内部任务类  
static class ComputingTask implements Callable<List>\{  
int start, end;  
public ComputingTask(int start, int end)\{this.start = start;this.end = end;\}  
//任务执行逻辑代码  
public List call() throws Exception\{  
List results = new ArrayList<>();  
boolean isPrime = true;  
for(int i = start; i <= end; i++)\{  
for(int j = 1; j < Math.sqrt(i); j++)\{if(i % j == 0)\{isPrime = false;break;\}\}  
if(isPrime)\{results.add(i);\}  
isPrime = true;  
\}  
return results;  
\}  
\}  
\}

6.Executors  
工具类型。为Executor线程池提供工具方法。可以快速的提供若干种线程池。  
如：固定容量的，无限容量的，容量为1等各种线程池。  
ExecutorService service = Executors.newCachedThreadPool();  
ExecutorService service = Executors.newSingleThreadExecutor();  
ScheduledExecutorService service = Executors.newScheduledThreadPool(3);  
ExecutorService service = Executors.newFixedThreadPool(5);

7.FixedThreadPool  
容量固定的线程池。活动状态和线程池容量是上限的线程池。所的线程池中，都一个任务队列。  
使用的是BlockingQueue作为任务的载体。  
当任务数量大于线程池容量的时候，没运行的任务保存在任务队列中，当线程空闲的，自动从队列中取出任务执行。  
使用场景： 大多数情况下，使用的线程池，首推荐FixedThreadPool。OS系统和硬件是线程支持上限。不能随意的无限制提供线程池。  
线程池默认的容量上限是Integer.MAX\_VALUE。  
常见的线程池容量： PC - 200。 服务器 - 1000~10000  
queued tasks - 任务队列  
completed tasks - 结束任务队列

8.CachedThreadPool  
①Future【无容量限制的线程池,最大容量默认为Integer.MAX\_VALUE】  
缓存线程池。容量不限（Integer.MAX\_VALUE)自动扩容。  
容量管理策略：如果线程池中的线程数量不满足任务执行，创建新的线程。  
每次新任务无法即时处理的时候，都会创建新的线程。  
当线程池中的线程空闲时长达到一定的临界值（默认60秒)，自动释放线程。  
应用场景： 内部应用或测试应用。

②代码实现:  
public class Test\_05\_CachedThreadPool \{  
public static void main(String\[\] args) \{  
ExecutorService service = Executors.newCachedThreadPool();  
System.out.println(service);  
for(int i = 0; i < 5; i++)\{  
service.execute(new Runnable() \{  
public void run() \{  
try \{TimeUnit.MILLISECONDS.sleep(500);\}  
catch (InterruptedException e) \{e.printStackTrace();\}  
System.out.println(Thread.currentThread().getName() + " - test executor");  
\}  
\});  
\}  
System.out.println(service);  
try \{TimeUnit.SECONDS.sleep(65);\}  
catch (InterruptedException e) \{e.printStackTrace();\}  
System.out.println(service);  
\}  
\}

9.SingleThreadExceutor  
①SingleThreadExceutor【容量为1的线程池,顺序执行】  
单一容量的线程池。  
使用场景：保证任务顺序时使用。如： 游戏大厅中的公共频道聊天。秒杀。

②代码实现:  
public class Test\_06\_SingleThreadExecutor \{  
public static void main(String\[\] args) \{  
ExecutorService service = Executors.newSingleThreadExecutor();  
System.out.println(service);  
for(int i = 0; i < 5; i++)\{  
service.execute(new Runnable() \{  
@Override  
public void run() \{  
try \{TimeUnit.MILLISECONDS.sleep(500);\}  
catch (InterruptedException e) \{e.printStackTrace();\}  
System.out.println(Thread.currentThread().getName() + " - test executor");  
\}  
\});  
\}  
\}  
\}

10.ScheduledThreadPool  
①ScheduledThreadPool【计划任务线程池】  
计划任务线程池。可以根据计划自动执行任务的线程池。  
scheduleAtFixedRate(Runnable, start\_limit, limit, timeunit)  
runnable - 要执行的任务。  
start\_limit - 第一次任务执行的间隔。  
limit - 多次任务执行的间隔。  
timeunit - 多次任务执行间隔的时间单位。  
使用场景： 计划任务时用（DelaydQueue，如：电信行业中的数据整理，没分钟整理，没消失整理，每天整理等。  
②代码实现:  
public class Test\_07\_ScheduledThreadPool \{  
public static void main(String\[\] args) \{  
ScheduledExecutorService service = Executors.newScheduledThreadPool(3);  
System.out.println(service);  
// runnable - 要执行的任务。  
service.scheduleAtFixedRate(new Runnable() \{  
public void run() \{  
try \{TimeUnit.MILLISECONDS.sleep(500);\}  
catch (InterruptedException e) \{e.printStackTrace();\}  
System.out.println(Thread.currentThread().getName());  
\}\}, 0, 300, TimeUnit.MILLISECONDS);  
\}  
\}

11.ForkJoinPool  
①ForkJoinPool【分支合并线程池】  
分支合并线程池（mapduce类似的设计思想。适合用于处理复杂任务。  
初始化线程容量与CPU核心数相关。  
线程池中运行的内容必须是ForkJoinTask的子类型（RecursiveTask,RecursiveAction)。  
ForkJoinPool - 分支合并线程池。 可以递归完成复杂任务。要求可分支合并的任务必须是ForkJoinTask类型的子类型。  
其中提供了分支和合并的能力。  
ForkJoinTask类型提供了两个抽象子类型，RecursiveTask返回结果的分支合并任务,RecursiveAction无返回结果的分支合并任务。  
(Callable/Runnable)compute方法：就是任务的执行逻辑。  
ForkJoinPool没所谓的容量。默认都是1个线程。根据任务自动的分支新的子线程。  
当子线程任务结束后，自动合并。所谓自动是根据fork和join两个方法实现的。  
应用： 主要是做科学计算或天文计算的。数据分析的。

②代码实现:  
public class Test\_08\_ForkJoinPool \{  
final static int\[\] numbers = new int\[1000000\];  
final static int MAX\_SIZE = 50000;  
final static Random r = new Random();  
static\{  
for(int i = 0; i < numbers.length; i++)\{numbers\[i\] = r.nextInt(1000);\}  
\}  
static class AddTask extends RecursiveTask\{ // RecursiveAction  
int begin, end;  
public AddTask(int begin, int end)\{  
this.begin = begin;  
this.end = end;  
\}  
//分支逻辑处理  
protected Long compute()\{  
if((end - begin) < MAX\_SIZE)\{ //不满足条件就继续分支  
long sum = 0L;  
for(int i = begin; i < end; i++)\{sum += numbers\[i\];\}  
return sum;  
\}else\{  
int middle = begin + (end - begin)/2;  
AddTask task1 = new AddTask(begin, middle);  
AddTask task2 = new AddTask(middle, end);  
task1.fork();// 就是用于开启新的任务的。 就是分支工作的。 就是开启一个新的线程任务。  
task2.fork();  
// join - 合并。将任务的结果获取。 这是一个阻塞方法。一定会得到结果数据。  
return task1.join() + task2.join();  
\}  
\}  
\}  
public static void main(String\[\] args) throws InterruptedException, ExecutionException, IOException \{  
long result = 0L;  
for(int i = 0; i < numbers.length; i++)\{result += numbers\[i\];\}  
System.out.println(result);  
ForkJoinPool pool = new ForkJoinPool();  
AddTask task = new AddTask(0, numbers.length);  
Future future = pool.submit(task);  
System.out.println(future.get());  
\}  
\}

12.ThreadPoolExecutor  
①ThreadPoolExecutor【模拟fixedThreadPool】  
线程池底层实现。除ForkJoinPool外，其他常用线程池底层都是使用ThreadPoolExecutor实现的。  
public ThreadPoolExecutor  
(  
int corePoolSize, // 核心容量，创建线程池的时候，默认多少线程。也是线程池保持的最少线程数  
int maximumPoolSize, // 最大容量，线程池最多多少线程  
long keepAliveTime, // 生命周期，0为永久。当线程空闲多久后，自动回收。  
TimeUnit unit, // 生命周期单位，为生命周期提供单位，如：秒，毫秒  
BlockingQueue workQueue // 任务队列，阻塞队列。注意，泛型必须是Runnable  
);  
使用场景： 默认提供的线程池不满足条件时使用。如：初始线程数据4，最大线程数200，线程空闲周期30秒。

②代码实现:  
public class Test\_09\_ThreadPoolExecutor \{  
public static void main(String\[\] args) \{  
// 模拟fixedThreadPool， 核心线程5个，最大容量5个，线程的生命周期无限。  
ExecutorService service = new ThreadPoolExecutor(5, 5, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue());  
for(int i = 0; i < 6; i++)\{  
service.execute(new Runnable() \{  
public void run() \{  
try \{TimeUnit.MILLISECONDS.sleep(500);\}  
catch (InterruptedException e) \{e.printStackTrace();\}  
System.out.println(Thread.currentThread().getName() + " - test executor");  
\}  
\});  
\}  
try \{TimeUnit.SECONDS.sleep(2);\}  
catch (InterruptedException e) \{e.printStackTrace();\}  
service.shutdown();  
System.out.println(service.isTerminated());  
System.out.println(service.isShutdown());  
System.out.println(service);  
\}  
\}

③线程池底层实现。除ForkJoinPool外,其他常用线程池底层都是使用ThreadPoolExecutor实现的。  
(1)SingleThreadExceutor  
ExecutorService service = Executors.newSingleThreadExecutor();

(2)SingleThreadExceutor底层实现
         public static ExecutorService newSingleThreadExecutor() {
          return new FinalizableDelegatedExecutorService(
              new ThreadPoolExecutor(1, 1,
              0L, TimeUnit.MILLISECONDS,
              new LinkedBlockingQueue<Runnable>()));
         }