摘要：它能阻塞一組線程直到某個事件發生。與閉鎖的區別所有線程必須同時到達柵欄位置，才能繼續執行。閉鎖用于等待事件，而柵欄用于等待其它線程。閉鎖一旦進入終止狀態，就不能被重置，它是一次性對象，而柵欄可以重置。
同步容器。它的原理是將狀態封裝起來，并對每個公有方法都實行同步，使得每次只有1個線程能夠訪問容器的狀態。
Vector和HashTable
Collections.synchronizedXXX方法
同步容器的問題
這種方式使得對容器的訪問都串行化，嚴重降低了并發性，如果多個線程來競爭容器的鎖時，吞吐量嚴重降低
對容器的多個方法的復合操作，是線程不安全的，比如一個線程負責刪除，另一個線程負責查詢，有可能出現越界的異常
并發容器。java.util.concurrent包里面的一系列實現
Concurrent開頭系列。以ConcurrentHashMap為例，它的實現原理為分段鎖。默認情況下有16個，每個鎖守護1/16的散列數據，這樣保證了并發量能達到16
分段鎖缺陷在于雖然一般情況下只要一個鎖，但是遇到需要擴容等類似的事情，只能去獲取所有的鎖
ConcurrentHashMap一些問題
需要對整個容器中的內容進行計算的方法，比如size、isEmpty、contains等等。由于并發的存在，在計算的過程中可能已進過期了，它實際上就是個估計值，但是在并發的場景下，需要使用的場景是很少的。
以ConcurrentHashMap的size方法為例:
/**
* Returns the number of key-value mappings in this map. If the
* map contains more than Integer.MAX_VALUE elements, returns
* Integer.MAX_VALUE.
*
* @return the number of key-value mappings in this map
*/
public int size() {
//為了能夠算準數量，會算2次，如果兩次算的不準，就鎖住再算
final Segment[] segments = this.segments;
int size;
boolean overflow; // true if size overflows 32 bits
long sum; // sum of modCounts
long last = 0L; // previous sum
int retries = -1; // 第一輪的計算總數不重試
try {
for (;;) {
if (retries++ == RETRIES_BEFORE_LOCK) {
//RETRIES_BEFORE_LOCK 默認是2
for (int j = 0; j < segments.length; ++j)
ensureSegment(j).lock(); // force creation
}
sum = 0L;
size = 0;
overflow = false;
for (int j = 0; j < segments.length; ++j) {
Segment seg = segmentAt(segments, j);
if (seg != null) {
sum += seg.modCount;
int c = seg.count;
if (c < 0 || (size += c) < 0)
overflow = true;
}
}
//第一次計算的時候
if (sum == last)
break; //如果前后兩次數數一致，就認為已經算好了
last = sum;
}
} finally {
if (retries > RETRIES_BEFORE_LOCK) {
for (int j = 0; j < segments.length; ++j)
segmentAt(segments, j).unlock();
}
}
return overflow ? Integer.MAX_VALUE : size;
}
不能提供線程獨占的功能
CopyOnWrite系列。以CopyOnWriteArrayList為例,只在每次修改的時候，進行加鎖控制，修改會創建并重新發布一個新的容器副本，其它時候由于都是事實上不可變的，也就不會出現線程安全問題
CopyOnWrite的問題
每次修改都復制底層數組，存在開銷，因此使用場景一般是迭代操作遠多于修改操作
CopyOnWriteArrayList的讀寫示例
/**
* Appends the specified element to the end of this list.
*
* @param e element to be appended to this list
* @return true (as specified by {@link Collection#add})
*/
public boolean add(E e) {
final ReentrantLock lock = this.lock;
lock.lock();
try {
Object[] elements = getArray();
int len = elements.length;
Object[] newElements = Arrays.copyOf(elements, len + 1);
newElements[len] = e;
setArray(newElements);
return true;
} finally {
lock.unlock();
}
}
/**
* {@inheritDoc}
*
* @throws IndexOutOfBoundsException {@inheritDoc}
*/
public E get(int index) {
return get(getArray(), index);
}
/**
* Gets the array. Non-private so as to also be accessible
* from CopyOnWriteArraySet class.
*/
final Object[] getArray() {
return array;
}
private E get(Object[] a, int index) {
return (E) a[index];
}
java中的同步工具類
阻塞隊列，BlockingQueue。它提供了put和take方法，在隊列不滿足各自條件時將產生阻塞
BlockingQueue使用示例,生產者-消費者
public static void main(String[] args) throws Exception {
BlockingQueue queue = new ArrayBlockingQueue(1024);
Producer producer = new Producer(queue);
Consumer consumer = new Consumer(queue);
new Thread(producer).start();
new Thread(consumer).start();
}
}
public class Producer implements Runnable{
protected BlockingQueue queue = null;
public Producer(BlockingQueue queue) {
this.queue = queue;
}
public void run() {
try {
queue.put("1");
Thread.sleep(1000);
queue.put("2");
Thread.sleep(2000);
queue.put("3");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
public class Consumer implements Runnable{
protected BlockingQueue queue = null;
public Consumer(BlockingQueue queue) {
this.queue = queue;
}
public void run() {
try {
System.out.println(queue.take());
System.out.println("Wait 1 sec");
System.out.println(queue.take());
System.out.println("Wait 2 sec");
System.out.println(queue.take());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
輸出為
1
Wait 1 sec
2
Wait 2 sec
3
閉鎖
CountDownLatch。使多個線程等待一組事件發生，它包含一個計數器，表示需要等待的事件的數量，每發生一個事，就遞減一次，當減為0時，所有事情發生，允許“通行”
CountDownLatch示例：
public class TestHarness{
public long timeTasks(int nThreads,final Runnable task) throws InterruptedException {
final CountDownLatch startGate = new CountDownLatch(1);
final CountDownLatch endGate = new CountDownLatch(nThreads);
for (int i=0;i
啟動門使主線程能夠同時釋放所有的工作線程，結束門使得主線程能夠等待最后一個線程執行完
- FutureTask。Future.get的如果任務執行完成，則立即返回，否則將阻塞直到任務完結，再返回結果或者是拋出異常
信號量，Semaphore 。它管理著一組虛擬的許可，許可的數量可通過構造函數指定，在執行操作時首先獲得許可，并在使用后釋放許可，如果沒有，那么accquire將阻塞直到有許可。
Semaphore示例
public class BoundedHashSet{
private final Set set;
private final Semaphore sem;
public BoundedHashSet(int bound) {
this.set = Collections.synchronizedSet(new HashSet());
this.sem = new Semaphore(bound);
}
public boolean add(T o) throws InterruptedException {
sem.acquire();
boolean wasAdded = false;
try {
wasAdded = set.add(o);
return wasAdded;
}finally {
if (!wasAdded){
sem.release();
}
}
}
public boolean remove(Object o){
boolean wasRemoved = set.remove(o);
if(wasRemoved){
sem.release();
}
return wasRemoved;
}
}
柵欄。它能阻塞一組線程直到某個事件發生。
與閉鎖的區別：
所有線程必須同時到達柵欄位置，才能繼續執行。閉鎖用于等待事件，而柵欄用于等待其它線程。
閉鎖一旦進入終止狀態，就不能被重置，它是一次性對象，而柵欄可以重置
CyclicBarrier。可以使一定數量的參與方反復地在柵欄位置匯集
CyclicBarrier使用示例
public static void main(String[] args) {
//第k步執行完才能執行第k+1步
CyclicBarrier barrier = new CyclicBarrier(3,new StageKPlusOne());
StageK[] stageKs = new StageK[3];
for (int i=0;i<3;i++){
stageKs[i] = new StageK(barrier,"k part "+(i+1));
}
for (int i=0;i<3;i++){
new Thread(stageKs[i]).start();
}
}
class StageKPlusOne implements Runnable{
@Override
public void run() {
System.out.println("stage k over");
System.out.println("stage k+1 start counting");
}
}
class StageK implements Runnable{
private CyclicBarrier barrier;
private String stage;
public StageK(CyclicBarrier barrier, String stage) {
this.barrier = barrier;
this.stage = stage;
}
@Override
public void run() {
System.out.println("stage "+stage+" counting...");
try {
TimeUnit.MILLISECONDS.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("stage "+stage+" count over");
try {
barrier.await();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
}
}
輸出為
stage k part 1 counting...
stage k part 3 counting...
stage k part 2 counting...
stage k part 2 count over
stage k part 3 count over
stage k part 1 count over
stage k over
stage k+1 start counting
Exchanger。它是一種兩方柵欄，各方在柵欄位置交換數據
Exchanger 使用示例：
public static void main(String[] args) {
Exchanger exchanger = new Exchanger();
ExchangerRunnable er1 = new ExchangerRunnable(exchanger,"1");
ExchangerRunnable er2 = new ExchangerRunnable(exchanger,"2");
new Thread(er1).start();
new Thread(er2).start();
}
class ExchangerRunnable implements Runnable{
private Exchanger e;
private Object o;
public ExchangerRunnable(Exchanger e, Object o) {
this.e = e;
this.o = o;
}
@Override
public void run() {
Object pre=o;
try {
o=e.exchange(o);
System.out.println("pre:"+pre+" now:"+o);
} catch (InterruptedException e1) {
e1.printStackTrace();
}
}
}
輸出如下
pre:1 now:2
pre:2 now:1
附錄
案例
文章版權歸作者所有，未經允許請勿轉載,若此文章存在違規行為，您可以聯系管理員刪除。
轉載請注明本文地址：http://specialneedsforspecialkids.com/yun/72396.html
