服务器之家：专注于服务器技术及软件下载分享
分类导航

PHP教程|ASP.NET教程|JAVA教程|ASP教程|

服务器之家 - 编程语言 - JAVA教程 - Java 线程池详解

Java 线程池详解

2020-04-13 11:28赵杰A-124 JAVA教程

本文给大家总结了java中的线程池的相关问题，非常的详细也很实用，有需要的小伙伴可以参考下。

系统启动一个线程的成本是比较高的，因为它涉及到与操作系统的交互，使用线程池的好处是提高性能，当系统中包含大量并发的线程时，会导致系统性能剧烈下降，甚至导致JVM崩溃，而线程池的最大线程数参数可以控制系统中并发线程数不超过次数。

一、Executors 工厂类用来产生线程池，该工厂类包含以下几个静态工厂方法来创建对应的线程池。创建的线程池是一个ExecutorService对象，使用该对象的submit方法或者是execute方法执行相应的Runnable或者是Callable任务。线程池本身在不再需要的时候调用shutdown()方法停止线程池，调用该方法后，该线程池将不再允许任务添加进来，但是会直到已添加的所有任务执行完成后才死亡。

1、newCachedThreadPool()，创建一个具有缓存功能的线程池，提交到该线程池的任务（Runnable或Callable对象）创建的线程，如果执行完成，会被缓存到CachedThreadPool中，供后面需要执行的任务使用。

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
 
public class CacheThreadPool {
  static class Task implements Runnable {
    @Override
    public void run() {
      System.out.println(this + " " + Thread.currentThread().getName() + " AllStackTraces map size: "
          + Thread.currentThread().getAllStackTraces().size());
    }
  }
 
  public static void main(String[] args) {
    ExecutorService cacheThreadPool = Executors.newCachedThreadPool();
    
    //先添加三个任务到线程池
    for(int i = 0 ; i < 3; i++) {
      cacheThreadPool.execute(new Task());
    }
    
    //等三个线程执行完成后，再次添加三个任务到线程池
    try {
      Thread.sleep(3000);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    
    for(int i = 0 ; i < 3; i++) {
      cacheThreadPool.execute(new Task());
    }
  }
 
}

执行结果如下：

?
1
2
3
4
5
6
CacheThreadPool$Task@2d312eb9 pool-1-thread-1 AllStackTraces map size: 7
CacheThreadPool$Task@59522b86 pool-1-thread-3 AllStackTraces map size: 7
CacheThreadPool$Task@73dbb89f pool-1-thread-2 AllStackTraces map size: 7
CacheThreadPool$Task@5795cedc pool-1-thread-3 AllStackTraces map size: 7
CacheThreadPool$Task@256d5600 pool-1-thread-1 AllStackTraces map size: 7
CacheThreadPool$Task@7d1c5894 pool-1-thread-2 AllStackTraces map size: 7

线程池中的线程对象进行了缓存，当有新任务执行时进行了复用。但是如果有特别多的并发时，缓存线程池还是会创建很多个线程对象。

2、newFixedThreadPool(int nThreads) 创建一个指定线程个数，线程可复用的线程池。

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
 
public class FixedThreadPool {
  static class Task implements Runnable {
    @Override
    public void run() {
      System.out.println(this + " " + Thread.currentThread().getName() + " AllStackTraces map size: "
          + Thread.currentThread().getAllStackTraces().size());
    }
  }
 
  public static void main(String[] args) {
    ExecutorService fixedThreadPool = Executors.newFixedThreadPool(3);
 
    // 先添加三个任务到线程池
    for (int i = 0; i < 5; i++) {
      fixedThreadPool.execute(new Task());
    }
 
    // 等三个线程执行完成后，再次添加三个任务到线程池
    try {
      Thread.sleep(3);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
 
    for (int i = 0; i < 3; i++) {
      fixedThreadPool.execute(new Task());
    }
  }
 
}

执行结果：

?
1
2
3
4
5
6
7
8
FixedThreadPool$Task@7045c12d pool-1-thread-2 AllStackTraces map size: 7
FixedThreadPool$Task@50fa0bef pool-1-thread-2 AllStackTraces map size: 7
FixedThreadPool$Task@ccb1870 pool-1-thread-2 AllStackTraces map size: 7
FixedThreadPool$Task@7392b4e3 pool-1-thread-1 AllStackTraces map size: 7
FixedThreadPool$Task@5bdeff18 pool-1-thread-2 AllStackTraces map size: 7
FixedThreadPool$Task@7d5554e1 pool-1-thread-1 AllStackTraces map size: 7
FixedThreadPool$Task@24468092 pool-1-thread-3 AllStackTraces map size: 7
FixedThreadPool$Task@fa7b978 pool-1-thread-2 AllStackTraces map size: 7

3、newSingleThreadExecutor(),创建一个只有单线程的线程池，相当于调用newFixedThreadPool(1)

4、newSheduledThreadPool(int corePoolSize),创建指定线程数的线程池，它可以在指定延迟后执行线程。也可以以某一周期重复执行某一线程，知道调用shutdown()关闭线程池。

示例如下：

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
 
public class ScheduledThreadPool {
  static class Task implements Runnable {
    @Override
    public void run() {
      System.out.println("time " + System.currentTimeMillis() + " " + Thread.currentThread().getName() + " AllStackTraces map size: "
          + Thread.currentThread().getAllStackTraces().size());
    }
  }
 
  public static void main(String[] args) {
    ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(3);
    
    scheduledExecutorService.schedule(new Task(), 3, TimeUnit.SECONDS);
    
    scheduledExecutorService.scheduleAtFixedRate(new Task(), 3, 5, TimeUnit.SECONDS);
  
    try {
      Thread.sleep(30 * 1000);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    scheduledExecutorService.shutdown();
  }
 
}

运行结果如下：

?
1
2
3
4
5
6
7
time 1458921795240 pool-1-thread-1 AllStackTraces map size: 6
time 1458921795241 pool-1-thread-2 AllStackTraces map size: 6
time 1458921800240 pool-1-thread-1 AllStackTraces map size: 7
time 1458921805240 pool-1-thread-1 AllStackTraces map size: 7
time 1458921810240 pool-1-thread-1 AllStackTraces map size: 7
time 1458921815240 pool-1-thread-1 AllStackTraces map size: 7
time 1458921820240 pool-1-thread-1 AllStackTraces map size: 7

由运行时间可看出，任务是按照5秒的周期执行的。

5、newSingleThreadScheduledExecutor() 创建一个只有一个线程的线程池，同调用newScheduledThreadPool(1)。

二、ForkJoinPool和ForkJoinTask

ForkJoinPool是ExecutorService的实现类，支持将一个任务划分为多个小任务并行计算，在把多个小任务的计算结果合并成总的计算结果。它有两个构造函数

ForkJoinPool(int parallelism)创建一个包含parallelism个并行线程的ForkJoinPool。

ForkJoinPool(),以Runtime.availableProcessors()方法返回值作为parallelism参数来创建ForkJoinPool。

ForkJoinTask 代表一个可以并行，合并的任务。它是实现了Future<T>接口的抽象类，它有两个抽象子类，代表无返回值任务的RecuriveAction和有返回值的RecursiveTask。可根据具体需求继承这两个抽象类实现自己的对象，然后调用ForkJoinPool的submit 方法执行。

RecuriveAction 示例如下，实现并行输出0-300的数字。

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveAction;
import java.util.concurrent.TimeUnit;
 
public class ActionForkJoinTask {
  static class PrintTask extends RecursiveAction {
    private static final int THRESHOLD = 50;
    private int start;
    private int end;
 
    public PrintTask(int start, int end) {
      this.start = start;
      this.end = end;
    }
 
    @Override
    protected void compute() {
      if (end - start < THRESHOLD) {
        for(int i = start; i < end; i++) {
          System.out.println(Thread.currentThread().getName() + " " + i);
        }
      } else {
        int middle = (start + end) / 2;
        PrintTask left = new PrintTask(start, middle);
        PrintTask right = new PrintTask(middle, end);
        left.fork();
        right.fork();
      }
    }
 
  }
 
  public static void main(String[] args) {
    ForkJoinPool pool = new ForkJoinPool();
    
    pool.submit(new PrintTask(0, 300));
    try {
      pool.awaitTermination(2, TimeUnit.SECONDS);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    
    pool.shutdown();
  }
 
}

在拆分小任务后，调用任务的fork（）方法，加入到ForkJoinPool中并行执行。

RecursiveTask示例，实现并行计算100个整数求和。拆分为每20个数求和后获取结果，在最后合并为最后的结果。

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
import java.util.Random;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;
import java.util.concurrent.RecursiveTask;
 
public class TaskForkJoinTask {
  static class CalTask extends RecursiveTask<Integer> {
    private static final int THRESHOLD = 20;
 
    private int arr[];
    private int start;
    private int end;
 
    public CalTask(int[] arr, int start, int end) {
      this.arr = arr;
      this.start = start;
      this.end = end;
    }
 
    @Override
    protected Integer compute() {
      int sum = 0;
 
      if (end - start < THRESHOLD) {
        for (int i = start; i < end; i++) {
          sum += arr[i];
        }
        System.out.println(Thread.currentThread().getName() + " sum:" + sum);
        return sum;
      } else {
        int middle = (start + end) / 2;
        CalTask left = new CalTask(arr, start, middle);
        CalTask right = new CalTask(arr, middle, end);
 
        left.fork();
        right.fork();
 
        return left.join() + right.join();
      }
    }
 
  }
 
  public static void main(String[] args) {
    int arr[] = new int[100];
    Random random = new Random();
    int total = 0;
 
    for (int i = 0; i < arr.length; i++) {
      int tmp = random.nextInt(20);
      total += (arr[i] = tmp);
    }
    System.out.println("total " + total);
 
    ForkJoinPool pool = new ForkJoinPool(4);
 
    Future<Integer> future = pool.submit(new CalTask(arr, 0, arr.length));
    try {
      System.out.println("cal result: " + future.get());
    } catch (InterruptedException e) {
      e.printStackTrace();
    } catch (ExecutionException e) {
      e.printStackTrace();
    }
    pool.shutdown();
  }
 
}

执行结果如下：

?
1
2
3
4
5
6
7
8
9
10
total 912
ForkJoinPool-1-worker-2 sum:82
ForkJoinPool-1-worker-2 sum:123
ForkJoinPool-1-worker-2 sum:144
ForkJoinPool-1-worker-3 sum:119
ForkJoinPool-1-worker-2 sum:106
ForkJoinPool-1-worker-2 sum:128
ForkJoinPool-1-worker-2 sum:121
ForkJoinPool-1-worker-3 sum:89
cal result: 912

子任务执行完后，调用任务的join（）方法获取子任务执行结果，再相加获得最后的结果。

  • Java
  • 线程池

    • 延伸 · 阅读

    精彩推荐
    最近更新
    编辑推荐
    阅读排行
    热门标签
    501 　 2000 　 1433 　 泛型 　 堆栈类 　 new 　 newInstance 　 Hibernate 　 volatile 　 集合 　 spring 　 consul 　 critical 　 嵌套循环 　 注册服务 　 Java编程 　 TIMER 　 算法 　 Java 　 遍历 　 设计模式 　 观察者模式 　 延迟加载技术 　 double 　 小数 　 JAVA8 　 写入文件 　 回车 　 制表符 　 排列组合 　
    © 2019-2020 服务器之家 版权所有 www.zzvips.com 关于我们联系我们版权申明网站地图