2. 程式人生 > >馬士兵老師高併發之6大執行緒池

馬士兵老師高併發之6大執行緒池

阿新 • • 發佈：2018-12-10

Executor

執行器，這是一個介面，內部維護了一個方法execute它負責執行一項任務。引數為Runnable，方法的具體實現由我們自己來執行。如下面的程式碼，我們既可以使用單純的方法呼叫也可以新啟一個新的執行緒去執行Runnable的run方法。

import java.util.concurrent.Executor;

public class T01_MyExecutor implements Executor {

	public static void main(String[] args) {
		new T01_MyExecutor().execute(()->System.out.println("hello executor"));
	}

	@Override
	public void execute(Runnable command) {
		//new Thread(command).run();
		command.run();
	}

}

ExecutorService

代表著啟動一系列的執行緒為使用者提供服務（本質上也是一個執行器），比如說Java8的官方文件就舉了一個網路接受連線池的例子（程式碼如下）。在這裡ExecutorService就代表著一個的執行緒池對外提供接受網路請求的服務。同時它也是一系列執行緒池的介面比如說

RorkJoinPool、ScheduledThreadPoolExecutor,、ThreadPoolExecutor等。同時它可以提交Callable與Runnable的物件返回一個未來的執行結果物件Future。這裡順便說一下，Callable是一個增強版的Runnable，它的call方法可以丟擲異常可以有返回值。其中它的返回值放在了Future物件中，我們可以使用Future物件的get方法來獲得返回值。

class NetworkService implements Runnable {
   private final ServerSocket serverSocket;
   private final ExecutorService pool;

   public NetworkService(int port, int poolSize)
       throws IOException {
     serverSocket = new ServerSocket(port);
     pool = Executors.newFixedThreadPool(poolSize);
   }

   public void run() { // run the service
     try {
       for (;;) {
         pool.execute(new Handler(serverSocket.accept()));
       }
     } catch (IOException ex) {
       pool.shutdown();
     }
   }
 }

 class Handler implements Runnable {
   private final Socket socket;
   Handler(Socket socket) { this.socket = socket; }
   public void run() {
     // read and service request on socket
   }
 }

除了以上方法來建立一個ExecutorService還可以使用Executors這個工具類來建立它，在這裡我們可以把Executors理解為就像utils，collections的工具類。

Future將來的結果

Future常與Callable聯合使用，Future可以獲得Callable執行後的返回值。如果想新建一個執行緒執行一個這個Callable中的call方法而且獲得返回值的話我們可以使用以下的思路。

方案一：new Thread(new FutureTask(一個實現了Callable的類的物件)).start();使用FutureTask來接收任務的返回值。

方案二：new一個執行緒池然後然後提交Callable的實現的物件。使用Future來獲得Callable的返回值。具體實現如下：

/**
 * 認識future
 */
package yxxy.c_026;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit;

public class T06_Future {
	public static void main(String[] args) throws InterruptedException, ExecutionException {
		
		FutureTask<Integer> task = new FutureTask<>(()->{
			TimeUnit.MILLISECONDS.sleep(500);
			return 1000;
		}); //new Callable () { Integer call();}
		
		new Thread(task).start();
		
		System.out.println(task.get()); //阻塞
		
		//*******************************
		ExecutorService service = Executors.newFixedThreadPool(5);
		Future<Integer> f = service.submit(()->{
			TimeUnit.MILLISECONDS.sleep(500);
			return 1;
		});
		System.out.println(f.get());
		System.out.println(f.isDone());
		
	}
}

6大執行緒池的介紹

FixedThreadPool

一個固定大小的執行緒池執行以下程式得到相應的結果：

public class T05_ThreadPool {
	public static void main(String[] args) throws InterruptedException {
		ExecutorService service = Executors.newFixedThreadPool(5); //execute submit
		for (int i = 0; i < 6; i++) {
			service.execute(() -> {
				try {
					TimeUnit.MILLISECONDS.sleep(500);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				System.out.println(Thread.currentThread().getName());
			});
		}
		System.out.println(service);
		
		service.shutdown();
		System.out.println(service.isTerminated());
		System.out.println(service.isShutdown());
		System.out.println(service);
		
		TimeUnit.SECONDS.sleep(5);
		System.out.println(service.isTerminated());
		System.out.println(service.isShutdown());
		System.out.println(service);
	}
}

執行結果

[email protected][Running, pool size = 5, active threads = 5, queued tasks = 1, completed tasks = 0]
false
true
[email protected][Shutting down, pool size = 5, active threads = 5, queued tasks = 1, completed tasks = 0]
pool-1-thread-1
pool-1-thread-2
pool-1-thread-5
pool-1-thread-4
pool-1-thread-3
pool-1-thread-1
true
true
[email protected][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個元素。

CachedPool

CachedPool的主要特點就是如果新來的一個任務需要這個執行緒池來執行的話，如果當前執行緒池沒有閒置的執行緒那麼就新啟動一個執行緒，如果有空閒執行緒那麼就使用其中的一個空閒執行緒。就是這樣的一個有彈性的執行緒池。預設情況下當一個執行緒空閒超過60s那麼就會銷燬，而且執行緒數量最大不能超過int型別的最大值或者是計算機記憶體的大小。以下程式碼展示了這樣的特性：

package yxxy.c_026;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class T08_CachedPool {
	public static void main(String[] args) throws InterruptedException {
		ExecutorService service = Executors.newCachedThreadPool();
		System.out.println(service);
		
		for (int i = 0; i < 2; i++) {
			service.execute(() -> {
				try {
					TimeUnit.MILLISECONDS.sleep(500);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				System.out.println(Thread.currentThread().getName());
			});
		}
		
		System.out.println(service);
		
		TimeUnit.SECONDS.sleep(80);
		
		System.out.println(service);
		
		
	}
}

SingleThreadPool

這個執行緒池中只有一個執行緒，那麼你可能回會問這與單個執行緒有什麼區別呢？: - ) 原因就是它可以被複用！它的使用場景就是當我們需要保證任務執行的先後順序的時候就可以使用它。

ScheduledPool

一個定時執行任務的一個執行緒池它所執行的任務的引數如下：

public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
                                                  long initialDelay,
                                                  long period,
                                                  TimeUnit unit)

initialDelay：在開始多少單位時間的時候執行第一個任務。

Period：每隔多長時間執行下一個任務。

Unit：時間的單位。

它的底層基於DelayedWorkQueue。

以下程式碼展示了已啟動就開始執行的而且步幅為0.5s的執行緒執行方式：

public class T10_ScheduledPool {
	public static void main(String[] args) {
		ScheduledExecutorService service = Executors.newScheduledThreadPool(4);
		service.scheduleAtFixedRate(()->{
			try {
				TimeUnit.MILLISECONDS.sleep(new Random().nextInt(1000));
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			System.out.println(Thread.currentThread().getName());
		}, 0, 500, TimeUnit.MILLISECONDS);
		
	}
}

工作竊取執行緒池，一般情況下CPU是幾核的就會啟動幾個執行緒，每一個執行緒都維護者自己的一個執行佇列的，當某些執行緒將自己佇列中的任務都執行完畢的時候就會去其他執行緒的佇列中竊取任務來執行以此提高效率。它的底層是基於ForkJoinPool的，常常用於任務分配不均勻的場景中。

需要注意的是，這個執行緒池產生的都是daemon的執行緒（後臺執行緒），所以我們需要將主執行緒阻塞來觀察輸出結果。

public class T11_WorkStealingPool {
	public static void main(String[] args) throws IOException {
		ExecutorService service = Executors.newWorkStealingPool();
		System.out.println(Runtime.getRuntime().availableProcessors());

		service.execute(new R(1000));
		service.execute(new R(2000));
		service.execute(new R(2000));
		service.execute(new R(2000)); //daemon
		service.execute(new R(2000));
		
		//由於產生的是精靈執行緒（守護執行緒、後臺執行緒），主執行緒不阻塞的話，看不到輸出
		System.in.read(); 
	}

	static class R implements Runnable {

		int time;

		R(int t) {
			this.time = t;
		}

		@Override
		public void run() {
			
			try {
				TimeUnit.MILLISECONDS.sleep(time);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			
			System.out.println(time  + " " + Thread.currentThread().getName());
			
		}

	}
}

ForkJoinPool

這個執行緒池設計的思想就與MapReduce極其相似，將一個大的任務分解成一個個小的任務當多個執行緒來執行。然後將計算的結果彙總得到最終結果。這也是用到了遞迴的思想。其中它的任務分為兩種一種沒有返回值是RecursiveAction，一種有返回值RecursiveTask。常常用於大量資料的運算以下為示例程式碼：

package yxxy.c_026;

import java.io.IOException;
import java.util.Arrays;
import java.util.Random;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;

public class T12_ForkJoinPool {
	static int[] nums = new int[1000000];
	static final int MAX_NUM = 50000;
	static Random r = new Random();
	
	static {
		for(int i=0; i<nums.length; i++) {
			nums[i] = r.nextInt(100);
		}
		
		System.out.println(Arrays.stream(nums).sum()); //stream api 
	}
	
	/*
	static class AddTask extends RecursiveAction { 
		
		int start, end;
		
		AddTask(int s, int e) {
			start = s;
			end = e;
		}

		@Override
		protected void compute() {
			
			if(end-start <= MAX_NUM) {
				long sum = 0L;
				for(int i=start; i<end; i++) sum += nums[i];
				System.out.println("from:" + start + " to:" + end + " = " + sum);
			} else {
			
				int middle = start + (end-start)/2;
				
				AddTask subTask1 = new AddTask(start, middle);
				AddTask subTask2 = new AddTask(middle, end);
				subTask1.fork();
				subTask2.fork();
			}
			
			
		}
		
	}
	*/
	
	static class AddTask extends RecursiveTask<Long> { 
		
		int start, end;
		
		AddTask(int s, int e) {
			start = s;
			end = e;
		}

		@Override
		protected Long compute() {
			
			if(end-start <= MAX_NUM) {
				long sum = 0L;
				for(int i=start; i<end; i++) sum += nums[i];
				return sum;
			} 
			
			int middle = start + (end-start)/2;
			
			AddTask subTask1 = new AddTask(start, middle);
			AddTask subTask2 = new AddTask(middle, end);
			subTask1.fork();
			subTask2.fork();
			
			return subTask1.join() + subTask2.join();
		}
		
	}
	
	public static void main(String[] args) throws IOException {
		ForkJoinPool fjp = new ForkJoinPool();
		AddTask task = new AddTask(0, nums.length);
		fjp.execute(task);
		long result = task.join();
		System.out.println(result);
		
		//System.in.read();
		
	}
}

事實上利用普通的執行緒池也可以完成大量資料的並行運算程式碼如下：

/**
 * 執行緒池的概念
 * nasa
 */
package yxxy.c_026;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class T07_ParallelComputing {
	public static void main(String[] args) throws InterruptedException, ExecutionException {
		long start = System.currentTimeMillis();
		List<Integer> results = getPrime(1, 200000); 
		long end = System.currentTimeMillis();
		System.out.println(end - start);
		
		final int cpuCoreNum = 4;
		
		ExecutorService service = Executors.newFixedThreadPool(cpuCoreNum);
		
		MyTask t1 = new MyTask(1, 80000); //1-5 5-10 10-15 15-20
		MyTask t2 = new MyTask(80001, 130000);
		MyTask t3 = new MyTask(130001, 170000);
		MyTask t4 = new MyTask(170001, 200000);
		
		Future<List<Integer>> f1 = service.submit(t1);
		Future<List<Integer>> f2 = service.submit(t2);
		Future<List<Integer>> f3 = service.submit(t3);
		Future<List<Integer>> f4 = service.submit(t4);
		
		start = System.currentTimeMillis();
		f1.get();
		f2.get();
		f3.get();
		f4.get();
		end = System.currentTimeMillis();
		System.out.println(end - start);
	}
	
	static class MyTask implements Callable<List<Integer>> {
		int startPos, endPos;
		
		MyTask(int s, int e) {
			this.startPos = s;
			this.endPos = e;
		}
		
		@Override
		public List<Integer> call() throws Exception {
			List<Integer> r = getPrime(startPos, endPos);
			return r;
		}
		
	}
	
	static boolean isPrime(int num) {
		for(int i=2; i<=num/2; i++) {
			if(num % i == 0) return false;
		}
		return true;
	}
	
	static List<Integer> getPrime(int start, int end) {
		List<Integer> results = new ArrayList<>();
		for(int i=start; i<=end; i++) {
			if(isPrime(i)) results.add(i);
		}
		
		return results;
	}
}

這是一個質數計算的問題，我們把質數計算劃分為不同的資料段是因為越大的質數越難計算，所以直觀上計算大量的小數字的質數的時間相當於計算少量的大數字花的時間。這樣一來我們就將這個大的任務相對均勻的拆分開來避免了任務分配不均勻造成的等待（也就是時間浪費）。

執行緒池的底層實現

前四種執行緒池的底層原始碼如下：

ChchedThreadPool

FixedThreadPool

ScheduledPool

SingleThreadPool

我們會發現他們都是基於ThreadExecutor。

而WorkStealingPool與ForkJoinPool的底層都是ForkJoinPool。

最後感謝馬士兵老師，一個專心做教育的老師。

相關推薦

馬士兵老師高併發之6大執行緒池

Executor 執行器，這是一個介面，內部維護了一個方法execute它負責執行一項任務。引數為Runnable，方法的具體實現由我們自己來執行。如下面的程式碼，我們既可以使用單純的方法呼叫也可以新啟一個新的執行緒去執行Runnable的run方法。 import ja

馬士兵老師高併發同步容器

手寫固定同步容器 寫一個固定容量同步容器，擁有put和get方法，以及getCount方法，能夠支援2個生產者執行緒以及10個消費者執行緒的阻塞呼叫。 使用wait與notify 思路：使用一個集合來當做生產或者消費的中轉站，然後每當生產或者消費的時刻都判斷集合的容量，

Java多執行緒高併發知識點二：執行緒池和CountDownLatch

執行緒池 池化是在為了複用常用的一種技術解決思路，比如連線池、執行緒池等，執行緒池主要是為了降低執行緒建立和銷燬造成的資源消耗，進而起到解決系統資源，提高系統相應速度的目的。Java中的ThreadPoolExecutor是JDK為我們提供好的執行緒池工具。

高併發第三彈:執行緒安全-原子性

執行緒安全性？ 執行緒安全性主要體現在三個方面：原子性、可見性、有序性 原子性:提供了互斥訪問，同一時刻只能有一個執行緒來對它進行操作 可見性:一個執行緒對主記憶體的修改可以及時的被其他執行緒觀察到 有序性:一個執行緒觀察其他執行緒中的指令執行順序，由於指令重排序的存在，該觀察結果一般雜亂無序。 本章主

高併發第六彈:執行緒封閉

當訪問共享的可變資料時，通常需要使用同步。一種避免使用同步的方式就是不共享資料。如果僅在單執行緒內訪問資料，就不需要同步。這種技術被稱為執行緒封閉。 它其實就是把物件封裝到一個執行緒裡，只有一個執行緒能看到這個物件，那麼這個物件就算不是執行緒安全的，也不會出現任何執行緒安全方面的問題。 二 執行緒封閉技術有

java高併發學習(九)-----多執行緒的團隊協作:同步控制

Java高併發學習(八)-------多執行緒的團隊協作:同步控制   同步控制是併發程式必不可少的重要手段。之前介紹的synchronized關鍵字就是一種最簡單的控制方法。同時，wait()和notify()方法起到了執行緒等待和通知的作用。這些工具對於實現複雜的多

高併發下最全執行緒安全的單例模式幾種實現

為了節約系統資源，有時需要確保系統中某個類只有唯一一個例項，當這個唯一例項建立成功之後，我們無法再建立一個同類型的其他物件，所有的操作都只能基於這個唯一例項。 但是餓漢式單例類不能實現延遲載入，不管將來用不用始終佔據記憶體；懶漢式單例類執行緒安全控制煩瑣，而且效能受影響。可

java併發之如何解決執行緒安全問題

併發（concurrency）一個並不陌生的詞，簡單來說，就是cpu在同一時刻執行多個任務。 而Java併發則由多執行緒實現的。 在jvm的世界裡，執行緒就像不相干的平行空間，序列在虛擬機器中。（當然這是比較籠統的說法，執行緒之間是可以互動的，他們也不一定是序列。） 多執行緒的存在就是壓榨cpu，提高程

什麼是高併發？與多執行緒有什麼區別？你的專案有高併發問題嗎？是如何解決的？

高併發是指較多使用者同時訪問服務。高併發可以由多執行緒實現，但是多執行緒不代表就是高併發。在會計匯有個投票調查專案，是和財政部合作的，會計人員進行填寫完調查結果後，將獲得5個學分。通過簡訊向全國500萬

高併發Java 二 多執行緒基

1. 什麼是執行緒 執行緒是程序內的執行單元 某個程序當中都有若干個執行緒。 執行緒是程序內的執行單元。 使用執行緒的原因是，程序的切換是非常重量級的操作，非常消耗資源。如果使用多程序，那麼併發數相對來說不會很高。而執行緒是更細小的排程單元，更加輕量級，所以執行緒會較為廣

C++11併發學習之六：執行緒池的實現

為什麼要使用執行緒池？       目前的大多數網路伺服器，包括Web伺服器、Email伺服器以及資料庫伺服器等都具有一個共同點，就是單位時間內必須處理數目巨大的連線請求，但處理時間卻相對較短。       傳統多執行緒方案中我們採用的伺服器模型則是一旦接受到請求之後，即建立

菜鳥之路——Java併發（四）執行緒池的使用

  轉自：http://www.cnblogs.com/dolphin0520/p/3932921.html   在前面的文章中，我們使用執行緒的時候就去建立一個執行緒，這樣實現起來非常簡便，但是就會有一個問題：   如果併發的執行緒數量很多，並且每個執行緒都是執行一個時間

高效能、高併發TCP伺服器(多執行緒呼叫libevent)

本文講述的TCP伺服器是模仿memcache中的TCP網路處理框架，其中是基於libevent網路庫的。 主執行緒只處理監聽客戶端的連線請求，並將請求平均分配給子執行緒。 子執行緒處理與客戶端的連線以及相關業務。 每個子執行緒有一個“連線”佇列。每個“連線”有一個“反饋”佇列。 先上個流程圖，要上班了

java高併發（九）執行緒封閉

上一節介紹了不可變物件，通過在某些情況下將不會修改的類物件設計成不可變物件，來讓物件在多個執行緒間保證物件是執行緒安全的。歸根到底

java高併發（六）執行緒安全性

執行緒安全性體現在以下三個方面：  原子性：提供了互斥訪問，同一時刻只能有一個執行緒來對它進行操作。 可見性：一個執行緒對主記憶體的修改可以及時的被其他執行緒觀察到。 有序性：一個執行緒觀察其他執行緒中的指令執行順序，由於指令重排序的存在，該觀察結果一般雜亂無序。 原

高併發程式設計學習(2)——執行緒通訊詳解

為獲得良好的閱讀體驗，請訪問原文: 傳送門 前序文章 高併發程式設計學習(1)——併發基礎 - https://www.wmyskxz.com/2019/11/26/gao-bing-fa-bian-cheng-xue-xi-1-bing-fa-ji-chu/ 一、經典的生產者消費者案例 上一篇

Java併發程式設計系列-(6) Java執行緒池

6. 執行緒池 6.1 基本概念 在web開發中，伺服器需要接受並處理請求，所以會為一個請求來分配一個執行緒來進行處理。如果每次請求都新建立一個執行緒的話實現起來非常簡便，但是存在一個問題：如果併發的請求數量非常多，但每個執行緒執行的時間很短，這樣就會頻繁的建立和銷燬執行緒，如此一來會大大降低系統的效率。

JAVA學習筆記（併發程式設計 - 捌）- 執行緒池

文章目錄 執行緒池 執行緒池的好處 執行緒池原理 執行緒池狀態 執行緒池常用方法 使用ThreadPoolExecutor建立執行緒池 執行緒池 執行緒資源必須通過執行緒池提供,不允許在應用中

【架構】Java併發程式設計 - 深入剖析執行緒池

前言 如果我們要使用執行緒的時候就去建立一個，這樣雖然非常簡便，但是就會有一個問題： 如果併發的執行緒數量很多，並且每個執行緒都是執行一個時間很短的任務就結束了，這樣頻繁建立執行緒就會大大降低系統的效率，因為頻繁建立執行緒和銷燬執行緒需要時間。 那麼有沒有一種辦法使得執行緒可以複用

Java 併發程式設計深入學習——執行緒池及其實現原理

Java執行緒池介紹   執行緒池，從字面含義來看，是指管理一組同構工作執行緒的資源池。執行緒池是與工作佇列（work Queue）密切相關的，其中工作佇列中儲存了所有等待執行的任務。工作者執行緒（Work Thread）的任務很簡單：從工作佇列中獲取一個任務，執行任務，然