zoukankan      html  css  js  c++  java

  • 马士兵老师高并发编程之6大线程池

    Executor

    执行器，这是一个接口，内部维护了一个方法execute它负责执行一项任务。参数为Runnable，方法的具体实现由我们自己来执行。如下面的代码，我们既可以使用单纯的方法调用也可以新启一个新的线程去执行Runnable的run方法。

    1. import java.util.concurrent.Executor;
    2.  
    3. public class T01_MyExecutor implements Executor {
    4.  
    5. 	public static void main(String[] args) {
    6. 		new T01_MyExecutor().execute(()->System.out.println("hello executor"));
    7. 	}
    8.  
    9. 	@Override
    10. 	public void execute(Runnable command) {
    11. 		//new Thread(command).run();
    12. 		command.run();
    13. 	}
    14.  
    15. }

    ExecutorService

    代表着启动一系列的线程为用户提供服务（本质上也是一个执行器），比如说Java8的官方文档就举了一个网络接受连接池的例子（代码如下）。在这里ExecutorService就代表着一个的线程池对外提供接受网络请求的服务。同时它也是一系列线程池的接口比如说

    RorkJoinPool、ScheduledThreadPoolExecutor,、ThreadPoolExecutor等。同时它可以提交Callable与Runnable的对象返回一个未来的执行结果对象Future。这里顺便说一下，Callable是一个增强版的Runnable，它的call方法可以抛出异常可以有返回值。其中它的返回值放在了Future对象中，我们可以使用Future对象的get方法来获得返回值。

    1. class NetworkService implements Runnable {
    2.    private final ServerSocket serverSocket;
    3.    private final ExecutorService pool;
    4.  
    5.    public NetworkService(int port, int poolSize)
    6.        throws IOException {
    7.      serverSocket = new ServerSocket(port);
    8.      pool = Executors.newFixedThreadPool(poolSize);
    9.    }
    10.  
    11.    public void run() { // run the service
    12.      try {
    13.        for (;;) {
    14.          pool.execute(new Handler(serverSocket.accept()));
    15.        }
    16.      } catch (IOException ex) {
    17.        pool.shutdown();
    18.      }
    19.    }
    20.  }
    21.  
    22.  class Handler implements Runnable {
    23.    private final Socket socket;
    24.    Handler(Socket socket) { this.socket = socket; }
    25.    public void run() {
    26.      // read and service request on socket
    27.    }
    28.  }

    除了以上方法来创建一个ExecutorService还可以使用Executors这个工具类来创建它，在这里我们可以把Executors理解为就像utils，collections的工具类。

    Future将来的结果

    Future常与Callable联合使用，Future可以获得Callable执行后的返回值。如果想新建一个线程执行一个这个Callable中的call方法而且获得返回值的话我们可以使用以下的思路。

    方案一：new Thread(new FutureTask(一个实现了Callable的类的对象)).start();使用FutureTask来接收任务的返回值。

    方案二：new一个线程池然后然后提交Callable的实现的对象。使用Future来获得Callable的返回值。具体实现如下：

    1. /**
    2.  * 认识future
    3.  */
    4. package yxxy.c_026;
    5.  
    6. import java.util.concurrent.ExecutionException;
    7. import java.util.concurrent.ExecutorService;
    8. import java.util.concurrent.Executors;
    9. import java.util.concurrent.Future;
    10. import java.util.concurrent.FutureTask;
    11. import java.util.concurrent.TimeUnit;
    12.  
    13. public class T06_Future {
    14. 	public static void main(String[] args) throws InterruptedException, ExecutionException {
    15. 		
    16. 		FutureTask<Integer> task = new FutureTask<>(()->{
    17. 			TimeUnit.MILLISECONDS.sleep(500);
    18. 			return 1000;
    19. 		}); //new Callable () { Integer call();}
    20. 		
    21. 		new Thread(task).start();
    22. 		
    23. 		System.out.println(task.get()); //阻塞
    24. 		
    25. 		//*******************************
    26. 		ExecutorService service = Executors.newFixedThreadPool(5);
    27. 		Future<Integer> f = service.submit(()->{
    28. 			TimeUnit.MILLISECONDS.sleep(500);
    29. 			return 1;
    30. 		});
    31. 		System.out.println(f.get());
    32. 		System.out.println(f.isDone());
    33. 		
    34. 	}
    35. }

    6大线程池的介绍

    FixedThreadPool

    一个固定大小的线程池运行以下程序得到相应的结果：

    1. public class T05_ThreadPool {
    2. 	public static void main(String[] args) throws InterruptedException {
    3. 		ExecutorService service = Executors.newFixedThreadPool(5); //execute submit
    4. 		for (int i = 0; i < 6; i++) {
    5. 			service.execute(() -> {
    6. 				try {
    7. 					TimeUnit.MILLISECONDS.sleep(500);
    8. 				} catch (InterruptedException e) {
    9. 					e.printStackTrace();
    10. 				}
    11. 				System.out.println(Thread.currentThread().getName());
    12. 			});
    13. 		}
    14. 		System.out.println(service);
    15. 		
    16. 		service.shutdown();
    17. 		System.out.println(service.isTerminated());
    18. 		System.out.println(service.isShutdown());
    19. 		System.out.println(service);
    20. 		
    21. 		TimeUnit.SECONDS.sleep(5);
    22. 		System.out.println(service.isTerminated());
    23. 		System.out.println(service.isShutdown());
    24. 		System.out.println(service);
    25. 	}
    26. }

    运行结果

    1. java.util.concurrent.ThreadPoolExecutor@1b28cdfa[Running, pool size = 5, active threads = 5, queued tasks = 1, completed tasks = 0]
    2. false
    3. true
    4. java.util.concurrent.ThreadPoolExecutor@1b28cdfa[Shutting down, pool size = 5, active threads = 5, queued tasks = 1, completed tasks = 0]
    5. pool-1-thread-1
    6. pool-1-thread-2
    7. pool-1-thread-5
    8. pool-1-thread-4
    9. pool-1-thread-3
    10. pool-1-thread-1
    11. true
    12. true
    13. java.util.concurrent.ThreadPoolExecutor@1b28cdfa[Terminated, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 6]
    • 整个程序new了一个5个线程的线程池，使用for循环向这个线程池抛了5个任务。它的执行原则是哪一个线程空闲就由哪个线程来执行这个任务。所以我们看到的线程池的线程序号是不固定的乱序的，但是它有个规则就是先执行完任务的线程会在新线程到来时优先分配到任务。
    • 线程池shutdown之后程序不会立刻停止而是要等待的所有线程都执行完毕之后再停止服务，所以我们看到的就是Runningà Shutting downà Terminated
    • 线程池的任务大体上分为两类，等待就绪队列与已完成任务的队列。通过输出结果我们可以看出在开始有5个正在执行的任务1个任务驻留在就绪队列等待执行，在执行结束后我们的已执行队列中就会有6个元素。

    CachedThreadPool

    CachedPool的主要特点就是如果新来的一个任务需要这个线程池来执行的话，如果当前线程池没有闲置的线程那么就新启动一个线程，如果有空闲线程那么就使用其中的一个空闲线程。就是这样的一个有弹性的线程池。默认情况下当一个线程空闲超过60s那么就会销毁，而且线程数量最大不能超过int类型的最大值或者是计算机内存的大小。以下代码展示了这样的特性：

    1. package yxxy.c_026;
    2.  
    3. import java.util.concurrent.ExecutorService;
    4. import java.util.concurrent.Executors;
    5. import java.util.concurrent.TimeUnit;
    6.  
    7. public class T08_CachedPool {
    8. 	public static void main(String[] args) throws InterruptedException {
    9. 		ExecutorService service = Executors.newCachedThreadPool();
    10. 		System.out.println(service);
    11. 		
    12. 		for (int i = 0; i < 2; i++) {
    13. 			service.execute(() -> {
    14. 				try {
    15. 					TimeUnit.MILLISECONDS.sleep(500);
    16. 				} catch (InterruptedException e) {
    17. 					e.printStackTrace();
    18. 				}
    19. 				System.out.println(Thread.currentThread().getName());
    20. 			});
    21. 		}
    22. 		
    23. 		System.out.println(service);
    24. 		
    25. 		TimeUnit.SECONDS.sleep(80);
    26. 		
    27. 		System.out.println(service);
    28. 		
    29. 		
    30. 	}
    31. }

    SingleThreadPool

    这个线程池中只有一个线程，那么你可能回会问这与单个线程有什么区别呢？: - ) 原因就是它可以被复用！它的使用场景就是当我们需要保证任务执行的先后顺序的时候就可以使用它。

    ScheduledThreadPool

    一个定时执行任务的一个线程池它所执行的任务的参数如下：

    1. public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
    2.                                                   long initialDelay,
    3.                                                   long period,
    4.                                                   TimeUnit unit)

    initialDelay：在开始多少单位时间的时候执行第一个任务。

    Period：每隔多长时间执行下一个任务。

    Unit：时间的单位。

    它的底层基于DelayedWorkQueue。

    以下代码展示了已启动就开始执行的而且步幅为0.5s的线程执行方式：

    1. public class T10_ScheduledPool {
    2. 	public static void main(String[] args) {
    3. 		ScheduledExecutorService service = Executors.newScheduledThreadPool(4);
    4. 		service.scheduleAtFixedRate(()->{
    5. 			try {
    6. 				TimeUnit.MILLISECONDS.sleep(new Random().nextInt(1000));
    7. 			} catch (InterruptedException e) {
    8. 				e.printStackTrace();
    9. 			}
    10. 			System.out.println(Thread.currentThread().getName());
    11. 		}, 0, 500, TimeUnit.MILLISECONDS);
    12. 		
    13. 	}
    14. }

    工作窃取线程池，一般情况下CPU是几核的就会启动几个线程，每一个线程都维护者自己的一个执行队列的，当某些线程将自己队列中的任务都执行完毕的时候就会去其他线程的队列中窃取任务来执行以此提高效率。它的底层是基于ForkJoinPool的，常常用于任务分配不均匀的场景中。

    需要注意的是，这个线程池产生的都是daemon的线程（后台线程），所以我们需要将主线程阻塞来观察输出结果。

    1. public class T11_WorkStealingPool {
    2. 	public static void main(String[] args) throws IOException {
    3. 		ExecutorService service = Executors.newWorkStealingPool();
    4. 		System.out.println(Runtime.getRuntime().availableProcessors());
    5.  
    6. 		service.execute(new R(1000));
    7. 		service.execute(new R(2000));
    8. 		service.execute(new R(2000));
    9. 		service.execute(new R(2000)); //daemon
    10. 		service.execute(new R(2000));
    11. 		
    12. 		//由于产生的是精灵线程（守护线程、后台线程），主线程不阻塞的话，看不到输出
    13. 		System.in.read(); 
    14. 	}
    15.  
    16. 	static class R implements Runnable {
    17.  
    18. 		int time;
    19.  
    20. 		R(int t) {
    21. 			this.time = t;
    22. 		}
    23.  
    24. 		@Override
    25. 		public void run() {
    26. 			
    27. 			try {
    28. 				TimeUnit.MILLISECONDS.sleep(time);
    29. 			} catch (InterruptedException e) {
    30. 				e.printStackTrace();
    31. 			}
    32. 			
    33. 			System.out.println(time  + " " + Thread.currentThread().getName());
    34. 			
    35. 		}
    36.  
    37. 	}
    38. }

    ForkJoinPool

    这个线程池设计的思想就与MapReduce极其相似，将一个大的任务分解成一个个小的任务当多个线程来执行。然后将计算的结果汇总得到最终结果。这也是用到了递归的思想。其中它的任务分为两种一种没有返回值是RecursiveAction，一种有返回值RecursiveTask。常常用于大量数据的运算以下为示例代码：

    1. package yxxy.c_026;
    2.  
    3. import java.io.IOException;
    4. import java.util.Arrays;
    5. import java.util.Random;
    6. import java.util.concurrent.ForkJoinPool;
    7. import java.util.concurrent.RecursiveTask;
    8.  
    9. public class T12_ForkJoinPool {
    10. 	static int[] nums = new int[1000000];
    11. 	static final int MAX_NUM = 50000;
    12. 	static Random r = new Random();
    13. 	
    14. 	static {
    15. 		for(int i=0; i<nums.length; i++) {
    16. 			nums[i] = r.nextInt(100);
    17. 		}
    18. 		
    19. 		System.out.println(Arrays.stream(nums).sum()); //stream api 
    20. 	}
    21. 	
    22. 	/*
    23. 	static class AddTask extends RecursiveAction { 
    24. 		
    25. 		int start, end;
    26. 		
    27. 		AddTask(int s, int e) {
    28. 			start = s;
    29. 			end = e;
    30. 		}
    32. 		@Override
    33. 		protected void compute() {
    34. 			
    35. 			if(end-start <= MAX_NUM) {
    36. 				long sum = 0L;
    37. 				for(int i=start; i<end; i++) sum += nums[i];
    38. 				System.out.println("from:" + start + " to:" + end + " = " + sum);
    39. 			} else {
    40. 			
    41. 				int middle = start + (end-start)/2;
    42. 				
    43. 				AddTask subTask1 = new AddTask(start, middle);
    44. 				AddTask subTask2 = new AddTask(middle, end);
    45. 				subTask1.fork();
    46. 				subTask2.fork();
    47. 			}
    48. 			
    49. 			
    50. 		}
    51. 		
    52. 	}
    53. 	*/
    54. 	
    55. 	static class AddTask extends RecursiveTask<Long> { 
    56. 		
    57. 		int start, end;
    58. 		
    59. 		AddTask(int s, int e) {
    60. 			start = s;
    61. 			end = e;
    62. 		}
    63.  
    64. 		@Override
    65. 		protected Long compute() {
    66. 			
    67. 			if(end-start <= MAX_NUM) {
    68. 				long sum = 0L;
    69. 				for(int i=start; i<end; i++) sum += nums[i];
    70. 				return sum;
    71. 			} 
    72. 			
    73. 			int middle = start + (end-start)/2;
    74. 			
    75. 			AddTask subTask1 = new AddTask(start, middle);
    76. 			AddTask subTask2 = new AddTask(middle, end);
    77. 			subTask1.fork();
    78. 			subTask2.fork();
    79. 			
    80. 			return subTask1.join() + subTask2.join();
    81. 		}
    82. 		
    83. 	}
    84. 	
    85. 	public static void main(String[] args) throws IOException {
    86. 		ForkJoinPool fjp = new ForkJoinPool();
    87. 		AddTask task = new AddTask(0, nums.length);
    88. 		fjp.execute(task);
    89. 		long result = task.join();
    90. 		System.out.println(result);
    91. 		
    92. 		//System.in.read();
    93. 		
    94. 	}
    95. }

    事实上利用普通的线程池也可以完成大量数据的并行运算代码如下：

    1. /**
    2.  * 线程池的概念
    3.  * nasa
    4.  */
    5. package yxxy.c_026;
    6.  
    7. import java.util.ArrayList;
    8. import java.util.List;
    9. import java.util.concurrent.Callable;
    10. import java.util.concurrent.ExecutionException;
    11. import java.util.concurrent.ExecutorService;
    12. import java.util.concurrent.Executors;
    13. import java.util.concurrent.Future;
    14.  
    15. public class T07_ParallelComputing {
    16. 	public static void main(String[] args) throws InterruptedException, ExecutionException {
    17. 		long start = System.currentTimeMillis();
    18. 		List<Integer> results = getPrime(1, 200000); 
    19. 		long end = System.currentTimeMillis();
    20. 		System.out.println(end - start);
    21. 		
    22. 		final int cpuCoreNum = 4;
    23. 		
    24. 		ExecutorService service = Executors.newFixedThreadPool(cpuCoreNum);
    25. 		
    26. 		MyTask t1 = new MyTask(1, 80000); //1-5 5-10 10-15 15-20
    27. 		MyTask t2 = new MyTask(80001, 130000);
    28. 		MyTask t3 = new MyTask(130001, 170000);
    29. 		MyTask t4 = new MyTask(170001, 200000);
    30. 		
    31. 		Future<List<Integer>> f1 = service.submit(t1);
    32. 		Future<List<Integer>> f2 = service.submit(t2);
    33. 		Future<List<Integer>> f3 = service.submit(t3);
    34. 		Future<List<Integer>> f4 = service.submit(t4);
    35. 		
    36. 		start = System.currentTimeMillis();
    37. 		f1.get();
    38. 		f2.get();
    39. 		f3.get();
    40. 		f4.get();
    41. 		end = System.currentTimeMillis();
    42. 		System.out.println(end - start);
    43. 	}
    44. 	
    45. 	static class MyTask implements Callable<List<Integer>> {
    46. 		int startPos, endPos;
    47. 		
    48. 		MyTask(int s, int e) {
    49. 			this.startPos = s;
    50. 			this.endPos = e;
    51. 		}
    52. 		
    53. 		@Override
    54. 		public List<Integer> call() throws Exception {
    55. 			List<Integer> r = getPrime(startPos, endPos);
    56. 			return r;
    57. 		}
    58. 		
    59. 	}
    60. 	
    61. 	static boolean isPrime(int num) {
    62. 		for(int i=2; i<=num/2; i++) {
    63. 			if(num % i == 0) return false;
    64. 		}
    65. 		return true;
    66. 	}
    67. 	
    68. 	static List<Integer> getPrime(int start, int end) {
    69. 		List<Integer> results = new ArrayList<>();
    70. 		for(int i=start; i<=end; i++) {
    71. 			if(isPrime(i)) results.add(i);
    72. 		}
    73. 		
    74. 		return results;
    75. 	}
    76. }

    这是一个质数计算的问题，我们把质数计算划分为不同的数据段是因为越大的质数越难计算，所以直观上计算大量的小数字的质数的时间相当于计算少量的大数字花的时间。这样一来我们就将这个大的任务相对均匀的拆分开来避免了任务分配不均匀造成的等待（也就是时间浪费）。

    线程池的底层实现

    前四种线程池的底层源码如下：

    ChchedThreadPool

     

    FixedThreadPool

    ScheduledPool

    SingleThreadPool

    我们会发现他们都是基于ThreadExecutor。

    而WorkStealingPool与ForkJoinPool的底层都是ForkJoinPool。

     

    最后感谢马士兵老师，一个专心做教育的老师。
  • 相关阅读:
    1648 最大和
    poj2243
    Codevs 2307[SDOI2009]HH的项链
    2597 团伙
    一个JavaWeb项目中使用的部分技术
    Oracle 11g 学习3——表空间操作
    iOS实现抽屉效果
    用shell脚本实现linux系统上wifi模式（STA和soft AP）的转换
    Codeforces Round #243 (Div. 1)——Sereja and Two Sequences
    站点选择配色诀窍
  • 原文地址：https://www.cnblogs.com/jpfss/p/9913351.html
Copyright © 2011-2022 走看看