同步器是一些使線程能夠等待另一個線程的對象，允許它們協調動作。最常用的同步器是CountDownLatch和Semaphore，不常用的是Barrier和Exchanger
隊列同步器AbstractQueuedSynchronizer是用來構建鎖或者其他同步組件的基礎框架，它內部使用了一個volatiole修飾的int類型的成員變量state來表示同步狀態，通過內置的FIFO隊列來完成資源獲取線程的排隊工作。
同步器的主要使用方式是繼承，子類通過繼承同步器并實現它的抽象方法來管理同步狀態，在抽象方法的實現過程中免不了要對同步狀態進行修改，這時就需要使用同步器來提供的3個方法（getState()、setState(intnewState)/和compareAndSetState(intexpect,intupdate)）來進行操作，因為他們能夠保證狀態的改變是安全的。子類推薦被定義為自定義同步組件的靜態內部類，同步器自身沒有實現任何同步接口，它僅僅是定義了若干同步狀態獲取個釋放的方法來供自定義同步組件使用，同步器既可以獨占式的獲取同步狀態，也可以支持共享式的獲取同步狀態，這樣就可以方便實現不同類型的同步組件（ReentrantLock、ReadWriteLock、和CountDownLatch等）。
同步器是實現鎖的關鍵，在鎖的實現中聚合同步器，利用同步器實現鎖的語義。他們二者直接的關系就是：鎖是面向使用者的，它定義了使用者與鎖交互的接口，隱藏了實現的細節；同步器則是面向鎖的實現者，它簡化了鎖的實現方式，屏蔽了同步狀態管理、線程的排隊、等待與喚醒等底層操作。鎖和同步器很好的隔離了使用者與實現者所需關注的領域。
同步器的設計是基于模版方法模式實現的，使用者需要繼承同步器并重寫這頂的方法，隨后將同步器組合在自定義同步組件的實現中，并調用同步器提供的模版方法，而這些模版方法將會調用使用者重寫的方法。
同步器提供的模版方法基本上分為3類：獨占式獲取鎖與釋放同步狀態、共享式獲取與釋放同步狀態和查詢同步隊列中的等待線程情況。自定義同步組件將使用同步器提供的模版方法來實現自己的同步語義。倒計數器鎖存器是一次性障礙，允許一個或者多個線程等待一個或者多個其它線程來做某些事情。CountDownLatch的唯一構造器帶一個int類型的參數，這個int參數是指允許所有在等待線程被處理之前，必須在鎖存器上調用countDown方法的次數。
EG：
java" id="highlighter_687825">
|
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
|
package hb.java.thread;import java.util.concurrent.CountDownLatch;/** * * @author hb * CountDownLatch最重要的方法是countDown()和await()，前者主要是倒數一次，后者是等待倒數到0，如果沒有到達0 * ，就只有阻塞等待了。 *JAVA同步器之 * CountDownLatch（不能循環使用，如果需要循環使用可以考慮使用CyclicBarrier） 兩種比較常規用法： 1：new * CountDownLatch(1);所有的線程在開始工作前需要做一些準備工作，當所有的線程都準備到位后再統一執行時有用 2：new * CountDownLatch(THREAD_COUNT);當所有的線程都執行完畢后，等待這些線程的其他線程才開始繼續執行時有用 */public class CountDownLatchTest { private static final int THREAD_COUNT = 10; // 在調用startSingal.countDown()之前調用了startSingal.await()的線程一律等待，直到startSingal.countDown()的調用 private static final CountDownLatch startSingal = new CountDownLatch(1); // 在finishedSingal的初始化記數量通過調用finishedSingal.countDown()減少為0時調用了finishedSingal.await()的線程一直阻塞 private static final CountDownLatch finishedSingal = new CountDownLatch( THREAD_COUNT); public static void main(String[] args) throws InterruptedException { for (int i = 0; i < THREAD_COUNT; i++) { new Thread("Task " + i) { public void run() { System.out.println(Thread.currentThread().getName() + " prepared!!"); try { startSingal.await(); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println(Thread.currentThread().getName() + " finished!!"); finishedSingal.countDown(); } ; } .start(); } Thread.sleep(1000); startSingal.countDown(); // 所有的線程被喚醒，同時開始工作.countDown 方法的線程等到計數到達零時才繼續 finishedSingal.await(); // 等待所有的線程完成!! System.out.println("All task are finished!!"); }} |
|
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
70
71
72
73
74
75
76
|
package hb.java.thread;import java.util.concurrent.BrokenBarrierException;import java.util.concurrent.CyclicBarrier;/** * * JAVA同步器之Barrier(能夠循環使用，當計數器增加到Barrier的初始化計數器之后馬上會被置為0為下一次循環使用做準備) * Barrier能夠為指定的一個或多個（一般為多個）線程設置一道屏障，只有當所有的線程都到達該屏障后才能一起沖過該屏障繼續其他任務 一般可以new * CyclicBarrier(ThreadCount)來進行初始化，也可以new * CyclicBarrier(ThreadCount,RunableAction)當初始化數量的線程都調用 * 了await()方法后觸發RunableAction線程，也可以通過初始化一個new * CyclicBarrier(ThreadCount+1)的Barrier在前置線程未執行完成時一直阻塞一個或多個 * 后續線程，這一點類似于CountDownLatch */public class BarrierTest { private static final int THREAD_COUNT = 10; private static final CyclicBarrier barrier = new CyclicBarrier( THREAD_COUNT + 1, new Runnable() { @Override public void run() { System.out.println("All task are prepared or finished!!"); } } ); public static void main(String[] args) throws InterruptedException, BrokenBarrierException { for (int i = 0; i < THREAD_COUNT; i++) { new Thread("Task " + i) { public void run() { try { System.out.println(Thread.currentThread().getName() + " prepared!!"); barrier.await(); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } catch (BrokenBarrierException e) { // TODO Auto-generated catch block e.printStackTrace(); } // do something System.out.println(Thread.currentThread().getName() + " finished!!"); } ; } .start(); } barrier.await(); // --------------開始準備循環使用-------------- for (int i = 0; i < THREAD_COUNT; i++) { new Thread("Task " + i) { public void run() { // do something System.out.println(Thread.currentThread().getName() + " finished!!"); try { barrier.await(); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } catch (BrokenBarrierException e) { // TODO Auto-generated catch block e.printStackTrace(); } } ; } .start(); } barrier.await(); }} |
|
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
70
71
72
73
74
75
76
77
|
package hb.java.thread;import java.util.ArrayList;import java.util.List;import java.util.concurrent.Exchanger;public class ExchangerTest { final static Exchanger<List<String>> exchanger = new Exchanger<List<String>>(); public static void main(String[] args) { new Producer("Producer", exchanger).start(); new Consumer("Consumer", exchanger).start(); } static class Producer extends Thread { private Exchanger<List<String>> exchanger; /** * */ public Producer(String threadName, Exchanger<List<String>> exchanger) { super(threadName); this.exchanger = exchanger; } /* * (non-Javadoc) * * @see java.lang.Thread#run() */ @Override public void run() { List<String> products = new ArrayList<String>(); for (int i = 0; i < 10; i++) { products.add("product " + i); } try { List<String> results = exchanger.exchange(products); System.out.println("get results from consumer"); for (String s : results) { System.out.println(s); } } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } } } static class Consumer extends Thread { private Exchanger<List<String>> exchanger; /** * */ public Consumer(String threadName, Exchanger<List<String>> exchanger) { super(threadName); this.exchanger = exchanger; } /* * (non-Javadoc) * * @see java.lang.Thread#run() */ @Override public void run() { List<String> products = new ArrayList<String>(); for (int i = 0; i < 10; i++) { products.add("consumed " + i); } try { List<String> results = exchanger.exchange(products); System.out.println("got products from produces"); for (String s : results) { System.out.println(s); } } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } } }} |
總結
以上就是本文關于java并發編程之同步器代碼示例的全部內容，希望對大家有所幫助。如有不足之處，歡迎留言指出。
原文鏈接：http://blog.csdn.net/huangbiao86/article/details/8089224
