程式師世界 >> 編程語言 >> JAVA編程 >> 關於JAVA >> 談談java的concurrent用法

談談java的concurrent用法

編輯：關於JAVA

談談java的concurrent用法。本站提示廣大學習愛好者：（談談java的concurrent用法）文章只能為提供參考，不一定能成為您想要的結果。以下是談談java的concurrent用法正文

我們都曉得，在JDK1.5之前，Java中要停止營業並發時，平日須要有法式員自力完成代碼完成，固然也有一些開源的框架供給了這些功效，然則這些仍然沒有JDK自帶的功效應用起來便利。而當針對高質量Java多線程並發法式設計時,為避免逝世蹦等景象的湧現，好比應用java之前的wait()、notify()和synchronized等，往往須要斟酌機能、逝世鎖、公正性、資本治理和若何防止線程平安性方面帶來的傷害等諸多身分，常常會采取一些較為龐雜的平安戰略，減輕了法式員的開辟累贅.萬幸的是，在JDK1.5湧現以後，Sun年夜神（Doug Lea）終究為我們這些不幸的小法式員推出了java.util.concurrent對象包以簡化並發完成。開辟者們借助於此，將有用的削減競爭前提（race conditions）和逝世鎖線程。concurrent包很好的處理了這些成績，為我們供給了更適用的並發法式模子。

Executor                  ：詳細Runnable義務的履行者。
ExecutorService           ：一個線程池治理者，其完成類有多種，我會引見一部門。我們能把Runnable,Callable提交到池中讓其調劑。
Semaphore                 ：一個計數旌旗燈號量
ReentrantLock             ：一個可重入的互斥鎖定 Lock，功效相似synchronized，但要壯大的多。
Future                    ：是與Runnable,Callable停止交互的接口，好比一個線程履行停止後取前往的成果等等，還供給了cancel終止線程。
BlockingQueue             ：壅塞隊列。
CompletionService         : ExecutorService的擴大，可以取得線程履行成果的
CountDownLatch            ：一個同步幫助類，在完成一組正在其他線程中履行的操作之前，它許可一個或多個線程一向期待。
CyclicBarrier             ：一個同步幫助類，它許可一組線程相互期待，直到達到某個公共樊籬點
Future                    ：Future 表現異步盤算的成果。
ScheduledExecutorService ：一個 ExecutorService，可支配在給定的延遲後運轉或按期履行的敕令。

接上去一一引見

Executors重要辦法解釋

newFixedThreadPool（固定年夜小線程池）

創立一個可重用固定線程聚集的線程池，以同享的無界隊列方法來運轉這些線程（只要要要求的過去，就會在一個隊列裡期待履行）。假如在封閉前的履行時代因為掉敗而招致任何線程終止，那末一個新線程將取代它履行後續的義務（假如須要）。

newCachedThreadPool（無界限程池，可以停止主動線程收受接管）

創立一個可依據須要創立新線程的線程池，然則在之前結構的線程可用時將重用它們。關於履行許多短時間異步義務的法式而言，這些線程池平日可進步法式機能。挪用 execute 將重用之前結構的線程（假如線程可用）。假如現有線程沒有可用的，則創立一個新線程並添加到池中。終止並從緩存中移除那些已有 60 秒鐘未被應用的線程。是以，長時光堅持余暇的線程池不會應用任何資本。留意，可使用 ThreadPoolExecutor 結構辦法創立具有相似屬性但細節分歧（例如超時參數）的線程池。

newSingleThreadExecutor（單個後台線程）

創立一個應用單個 worker 線程的 Executor，以無界隊列方法來運轉該線程。（留意，假如由於在封閉前的履行時代湧現掉敗而終止了此單個線程，那末假如須要，一個新線程將取代它履行後續的義務）。可包管次序地履行各個義務，而且在隨意率性給定的時光不會有多個線程是運動的。與其他等效的 newFixedThreadPool(1) 分歧，可包管無需從新設置裝備擺設此辦法所前往的履行法式便可應用其他的線程。

這些辦法前往的都是ExecutorService對象，這個對象可以懂得為就是一個線程池。

這個線程池的功效照樣比擬完美的。可以提交義務submit()可以停止線程池shutdown()。

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class MyExecutor extends Thread {
private int index;
public MyExecutor(int i){
  this.index=i;
}
public void run(){
  try{
   System.out.println("["+this.index+"] start....");
   Thread.sleep((int)(Math.random()*));
   System.out.println("["+this.index+"] end.");
  }
  catch(Exception e){
   e.printStackTrace();
  }
}
public static void main(String args[]){
  ExecutorService service=Executors.newFixedThreadPool();
  for(int i=;i<;i++){
   service.execute(new MyExecutor(i));
   //service.submit(new MyExecutor(i));
  }
  System.out.println("submit finish");
  service.shutdown();
}
}

固然打印了一些信息，然則看的不長短常清楚，這個線程池是若何任務的，我們來將休眠的時光調長10倍。

Thread.sleep((int)(Math.random()*10000));

再來看，會清晰看到只能履行4個線程。當履行完一個線程後，才會又履行一個新的線程，也就是說，我們將一切的線程提交後，線程池會期待履行完最初shutdown。我們也會發明，提交的線程被放到一個“無界隊列裡”。這是一個有序隊列（BlockingQueue，這個上面會說到）。

別的它應用了Executors的靜態函數生成一個固定的線程池，望文生義，線程池的線程是不會釋放的，即便它是Idle。

這就會發生機能成績，好比假如線程池的年夜小為200，當全體應用終了後，一切的線程會持續留在池中，響應的內存和線程切換（while(true)+sleep輪回）都邑增長。

假如要防止這個成績，就必需直接應用ThreadPoolExecutor()來結構。可以像通用的線程池一樣設置“最年夜線程數”、“最小線程數”和“余暇線程keepAlive的時光”。

這個就是線程池根本用法。

Semaphore

一個計數旌旗燈號量。從概念上講，旌旗燈號量保護了一個允許聚集。若有需要，在允許可用前會壅塞每個 acquire()，然後再獲得該允許。每一個 release() 添加一個允許，從而能夠釋放一個正在壅塞的獲得者。然則，不應用現實的允許對象，Semaphore 只對可用允許的號碼停止計數，並采用響應的行為。

Semaphore 平日用於限制可以拜訪某些資本（物理或邏輯的）的線程數量。例如，上面的類應用旌旗燈號量掌握對內容池的拜訪：

這裡是一個現實的情形，年夜家列隊上茅廁，茅廁只要兩個地位，來了10小我須要列隊。

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
public class MySemaphore extends Thread {
Semaphore position;
private int id;
public MySemaphore(int i,Semaphore s){
  this.id=i;
  this.position=s;
}
public void run(){
  try{
   if(position.availablePermits()>){
   System.out.println("顧客["+this.id+"]進入茅廁，有空位");
   }
   else{
   System.out.println("顧客["+this.id+"]進入茅廁，沒空位，列隊");
   }
   position.acquire();
   System.out.println("顧客["+this.id+"]取得坑位");
   Thread.sleep((int)(Math.random()*));
   System.out.println("顧客["+this.id+"]應用終了");
   position.release();
  }
  catch(Exception e){
   e.printStackTrace();
  }
}
public static void main(String args[]){
  ExecutorService list=Executors.newCachedThreadPool();
  Semaphore position=new Semaphore();
  for(int i=;i<;i++){
   list.submit(new MySemaphore(i+,position));
  }
  list.shutdown();
  position.acquireUninterruptibly();
  System.out.println("應用終了，須要打掃了");
  position.release();
}
}

ReentrantLock

一個可重入的互斥鎖定 Lock，它具有與應用 synchronized 辦法和語句所拜訪的隱式監督器鎖定雷同的一些根本行動和語義，但功效更壯大。

ReentrantLock 將由比來勝利取得鎖定，而且還沒有釋放該鎖定的線程所具有。當鎖定沒有被另外一個線程所具有時，挪用 lock 的線程將勝利獲得該鎖定並前往。假如以後線程曾經具有該鎖定，此辦法將立刻前往。可使用 isHeldByCurrentThread() 和 getHoldCount() 辦法來檢討此情形能否產生。

此類的結構辦法接收一個可選的公正參數。

當設置為 true時，在多個線程的爭用下，這些鎖定偏向於將拜訪權授與期待時光最長的線程。不然此鎖定將沒法包管任何特定拜訪次序。

與采取默許設置（應用不公正鎖定）比擬，應用公正鎖定的法式在很多線程拜訪時表示為很低的整體吞吐量（即速度很慢，經常極端慢），然則在取得鎖定和包管鎖定分派的平衡性時差別較小。

不外要留意的是，公正鎖定不克不及包管線程調劑的公正性。是以，應用公正鎖定的浩瀚線程中的一員能夠取得多倍的勝利機遇，這類情形產生在其他運動線程沒有被處置而且今朝並未持有鎖准時。
還要留意的是，不決時的 tryLock 辦法並沒有應用公正設置。由於即便其他線程正在期待，只需該鎖定是可用的，此辦法便可以取得勝利。

建議老是立刻理論，應用 try 塊來挪用 lock，在之前/以後的結構中，最典范的代碼以下：

class X {
  private final ReentrantLock lock = new ReentrantLock();
  // ...
  public void m() {
   lock.lock(); // block until condition holds
   try {
    // ... method body
   } finally {
    lock.unlock()
   }
  }
}

我的例子：

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.ReentrantLock;
public class MyReentrantLock extends Thread{
TestReentrantLock lock;
private int id;
public MyReentrantLock(int i,TestReentrantLock test){
  this.id=i;
  this.lock=test;
}
public void run(){
  lock.print(id);
}
public static void main(String args[]){
  ExecutorService service=Executors.newCachedThreadPool();
  TestReentrantLock lock=new TestReentrantLock();
  for(int i=;i<;i++){
   service.submit(new MyReentrantLock(i,lock));
  }
  service.shutdown();
}
}
class TestReentrantLock{
private ReentrantLock lock=new ReentrantLock();
public void print(int str){
  try{
   lock.lock();
   System.out.println(str+"取得");
   Thread.sleep((int)(Math.random()*));
  }
  catch(Exception e){
   e.printStackTrace();
  }
  finally{
   System.out.println(str+"釋放");
   lock.unlock();
  }
}
}

BlockingQueue

支撐兩個附加操作的 Queue，這兩個操作是：檢索元素時期待隊列變成非空，和存儲元素時期待空間變得可用。

BlockingQueue 不接收 null 元素。試圖 add、put 或 offer 一個 null 元素時，某些完成會拋出 NullPointerException。null 被用作指導 poll 操作掉敗的警惕值。

BlockingQueue 可所以限制容量的。它在隨意率性給准時間都可以有一個 remainingCapacity，超越此容量，便沒法無壅塞地 put 額定的元素。

沒有任何外部容量束縛的 BlockingQueue 老是申報 Integer.MAX_VALUE 的殘剩容量。

BlockingQueue 完成重要用於臨盆者-應用者隊列，但它別的還支撐 Collection 接口。是以，舉例來講，應用 remove(x) 從隊列中移除隨意率性一個元素是有能夠的。

但是，這類操作平日不會有用履行，只能有籌劃地偶然應用，好比在撤消列隊信息時。

BlockingQueue 完成是線程平安的。一切列隊辦法都可使用外部鎖定或其他情勢的並發掌握來主動到達它們的目標。

但是，年夜量的 Collection 操作（addAll、containsAll、retainAll 和 removeAll）沒有需要主動履行，除非在完成中特殊解釋。

是以，舉例來講，在只添加了 c 中的一些元素後，addAll(c) 有能夠掉敗（拋出一個異常）。

BlockingQueue 本質上不支撐應用任何一種“close”或“shutdown”操作來指導不再添加任何項。

這類功效的需乞降應用有依附於完成的偏向。例如，一種經常使用的戰略是：關於臨盆者，拔出特別的 end-of-stream 或 poison 對象，並依據應用者獲得這些對象的時光來對它們停止說明。

上面的例子演示了這個壅塞隊列的根本功效。

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
public class MyBlockingQueue extends Thread {
public static BlockingQueue<String> queue = new LinkedBlockingQueue<String>();
private int index;
public MyBlockingQueue(int i) {
  this.index = i;
}
public void run() {
  try {
  queue.put(String.valueOf(this.index));
  System.out.println("{" + this.index + "} in queue!");
  } catch (Exception e) {
  e.printStackTrace();
  }
}
public static void main(String args[]) {
  ExecutorService service = Executors.newCachedThreadPool();
  for (int i = ; i < ; i++) {
  service.submit(new MyBlockingQueue(i));
  }
  Thread thread = new Thread() {
  public void run() {
   try {
   while (true) {
    Thread.sleep((int) (Math.random() * ));
    if(MyBlockingQueue.queue.isEmpty())
    break;
    String str = MyBlockingQueue.queue.take();
    System.out.println(str + " has take!");
   }
   } catch (Exception e) {
   e.printStackTrace();
   }
  }
  };
  service.submit(thread);
  service.shutdown();
}
}

---------------------履行成果-----------------
{0} in queue!
{1} in queue!
{2} in queue!
{3} in queue!
0 has take!
{4} in queue!
1 has take!
{6} in queue!
2 has take!
{7} in queue!
3 has take!
{8} in queue!
4 has take!
{5} in queue!
6 has take!
{9} in queue!
7 has take!
8 has take!
5 has take!
9 has take!
-----------------------------------------

CompletionService

將臨盆新的異步義務與應用已完成義務的成果分別開來的辦事。臨盆者 submit 履行的義務。應用者 take 已完成的義務，並依照完成這些義務的次序處置它們的成果。例如，CompletionService 可以用來治理異步 IO ，履行讀操作的義務作為法式或體系的一部門提交，然後，當完成讀操作時，會在法式的分歧部門履行其他操作，履行操作的次序能夠與所要求的次序分歧。

平日，CompletionService 依附於一個零丁的 Executor 來現實履行義務，在這類情形下，CompletionService 盡管理一個外部完成隊列。ExecutorCompletionService 類供給了此辦法的一個完成。

import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class MyCompletionService implements Callable<String> {
private int id;
public MyCompletionService(int i){
  this.id=i;
}
public static void main(String[] args) throws Exception{
  ExecutorService service=Executors.newCachedThreadPool();
  CompletionService<String> completion=new ExecutorCompletionService<String>(service);
  for(int i=;i<;i++){
  completion.submit(new MyCompletionService(i));
  }
  for(int i=;i<;i++){
  System.out.println(completion.take().get());
  }
  service.shutdown();
}
public String call() throws Exception {
  Integer time=(int)(Math.random()*);
  try{
  System.out.println(this.id+" start");
  Thread.sleep(time);
  System.out.println(this.id+" end");
  }
  catch(Exception e){
  e.printStackTrace();
  }
  return this.id+":"+time;
}
}

CountDownLatch

一個同步幫助類，在完成一組正在其他線程中履行的操作之前，它許可一個或多個線程一向期待。

用給定的計數初始化 CountDownLatch。因為挪用了 countDown() 辦法，所以在以後計數達到零之前，await 辦法會一向受壅塞。

以後，會釋放一切期待的線程，await 的一切後續挪用都將立刻前往。這類景象只湧現一次——計數沒法被重置。假如須要重置計數，請斟酌應用 CyclicBarrier。

CountDownLatch 是一個通用同步對象，它有許多用處。將計數 1 初始化的 CountDownLatch 用作一個簡略的開/關鎖存器，或進口：在經由過程挪用 countDown() 的線程翻開進口前，一切挪用 await 的線程都一向在進口處期待。

用 N 初始化的 CountDownLatch 可使一個線程在 N 個線程完成某項操作之前一向期待，或許使其在某項操作完成 N 次之前一向期待。

CountDownLatch 的一個有效特征是，它不請求挪用 countDown 辦法的線程比及計數達到零時才持續，而在一切線程都能經由過程之前，它只是阻攔任何線程持續經由過程一個 await。

一下的例子是他人寫的，異常抽象。

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class TestCountDownLatch {
public static void main(String[] args) throws InterruptedException {
  // 開端的倒數鎖
  final CountDownLatch begin = new CountDownLatch();
  // 停止的倒數鎖
  final CountDownLatch end = new CountDownLatch();
  // 十名選手
  final ExecutorService exec = Executors.newFixedThreadPool();
  for (int index = ; index < ; index++) {
  final int NO = index + ;
  Runnable run = new Runnable() {
   public void run() {
   try {
    begin.await();//一向壅塞
    Thread.sleep((long) (Math.random() * ));
    System.out.println("No." + NO + " arrived");
   } catch (InterruptedException e) {
   } finally {
    end.countDown();
   }
   }
  };
  exec.submit(run);
  }
  System.out.println("Game Start");
  begin.countDown();
  end.await();
  System.out.println("Game Over");
  exec.shutdown();
}
}

CountDownLatch最主要的辦法是countDown()和await()，前者重要是倒數一次，後者是期待倒數到0，假如沒有達到0，就只要壅塞期待了。

CyclicBarrier

一個同步幫助類，它許可一組線程相互期待，直到達到某個公共樊籬點 (common barrier point)。

在觸及一組固定年夜小的線程的法式中，這些線程必需不時地相互期待，此時 CyclicBarrier 很有效。由於該 barrier 在釋放期待線程後可以重用，所以稱它為輪回的 barrier。

CyclicBarrier 支撐一個可選的 Runnable 敕令，在一組線程中的最初一個線程達到以後（但在釋放一切線程之前），該敕令只在每一個樊籬點運轉一次。若在持續一切介入線程之前更新同享狀況，此樊籬操作很有效。

示例用法：上面是一個在並行分化設計中應用 barrier 的例子，很經典的觀光團例子：

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class TestCyclicBarrier {
 // 徒步須要的時光: Shenzhen, Guangzhou, Shaoguan, Changsha, Wuhan
 private static int[] timeWalk = { , , , , };
 // 自駕游
 private static int[] timeSelf = { , , , , };
 // 旅游年夜巴
 private static int[] timeBus = { , , , , };
 static String now() {
   SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss");
   return sdf.format(new Date()) + ": ";
 }
 static class Tour implements Runnable {
   private int[] times;
   private CyclicBarrier barrier;
   private String tourName;
   public Tour(CyclicBarrier barrier, String tourName, int[] times) {
    this.times = times;
    this.tourName = tourName;
    this.barrier = barrier;
   }
   public void run() {
    try {
     Thread.sleep(times[] * );
     System.out.println(now() + tourName + " Reached Shenzhen");
     barrier.await();
     Thread.sleep(times[] * );
     System.out.println(now() + tourName + " Reached Guangzhou");
     barrier.await();
     Thread.sleep(times[] * );
     System.out.println(now() + tourName + " Reached Shaoguan");
     barrier.await();
     Thread.sleep(times[] * );
     System.out.println(now() + tourName + " Reached Changsha");
     barrier.await();
     Thread.sleep(times[] * );
     System.out.println(now() + tourName + " Reached Wuhan");
     barrier.await();
    } catch (InterruptedException e) {
    } catch (BrokenBarrierException e) {
    }
   }
 }
 public static void main(String[] args) {
   // 三個觀光團
   CyclicBarrier barrier = new CyclicBarrier();
   ExecutorService exec = Executors.newFixedThreadPool();
   exec.submit(new Tour(barrier, "WalkTour", timeWalk));
   exec.submit(new Tour(barrier, "SelfTour", timeSelf));
//當我們把上面的這段代碼正文後，會發明，法式壅塞了，沒法持續運轉下去。
   exec.submit(new Tour(barrier, "BusTour", timeBus));
   exec.shutdown();
 }
}

CyclicBarrier最主要的屬性就是介入者個數，別的最要辦法是await()。當一切線程都挪用了await()後，就表現這些線程都可以持續履行，不然就會期待。

Future

Future 表現異步盤算的成果。它供給了檢討盤算能否完成的辦法，以期待盤算的完成，並檢索盤算的成果。

盤算完成後只能應用 get 辦法來檢索成果，若有需要，盤算完成前可以壅塞此辦法。撤消則由 cancel 辦法來履行。

還供給了其他辦法，以肯定義務是正常完成照樣被撤消了。一旦盤算完成，就不克不及再撤消盤算。

假如為了可撤消性而應用 Future但又不供給可用的成果，則可以聲明 Future<?> 情勢類型、並前往 null 作為基本義務的成果。

這個我們在後面CompletionService曾經看到了，這個Future的功效，並且這個可以在提交線程的時刻被指定為一個前往對象的。

ScheduledExecutorService

一個 ExecutorService，可支配在給定的延遲後運轉或按期履行的敕令。

schedule 辦法應用各類延遲創立義務，並前往一個可用於撤消或檢討履行的義務對象。scheduleAtFixedRate 和 scheduleWithFixedDelay 辦法創立並履行某些在撤消前一向按期運轉的義務。

用 Executor.execute(java.lang.Runnable) 和 ExecutorService 的 submit 辦法所提交的敕令，經由過程所要求的 0 延遲停止支配。

schedule 辦法中許可湧現 0 和正數延遲（但不是周期），並將這些視為一種立刻履行的要求。

一切的 schedule 辦法都接收絕對延遲和周期作為參數，而不是相對的時光或日期。將以 Date 所表現的相對時光轉換成請求的情勢很輕易。

例如，要支配在某個今後的日期運轉，可使用：schedule(task, date.getTime() - System.currentTimeMillis(), TimeUnit.MILLISECONDS)。

然則要留意，因為收集時光同步協定、時鐘漂移或其他身分的存在，是以絕對延遲的期滿日期不用與啟用義務確當前 Date 符合。

Executors 類為此包中所供給的 ScheduledExecutorService 完成供給了便捷的工場辦法。

一下的例子也是網上比擬風行的。

import static java.util.concurrent.TimeUnit.SECONDS;
import java.util.Date;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
public class TestScheduledThread {
public static void main(String[] args) {
  final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool();
  final Runnable beeper = new Runnable() {
  int count = ;
  public void run() {
   System.out.println(new Date() + " beep " + (++count));
  }
  };
  // 秒鐘後運轉，並每隔秒運轉一次
  final ScheduledFuture beeperHandle = scheduler.scheduleAtFixedRate(beeper, , , SECONDS);
  // 秒鐘後運轉，並每次在前次義務運轉完後期待秒後從新運轉
  final ScheduledFuture beeperHandle = scheduler.scheduleWithFixedDelay(beeper, , , SECONDS);
  // 秒後停止封閉義務，而且封閉Scheduler
  scheduler.schedule(new Runnable() {
  public void run() {
   beeperHandle.cancel(true);
   beeperHandle.cancel(true);
   scheduler.shutdown();
  }
  }, , SECONDS);
}
}

如許我們就把concurrent包下比擬主要的功效都曾經總結完了，願望對我們懂得能有贊助。