java 對線程安全支持有哪些？

時間 2019-12-04

標籤 java 線程安全支持哪些欄目 Java 简体版

原文原文鏈接

同步容器。它的原理是將狀態封裝起來，並對每一個公有方法都實行同步，使得每次只有1個線程可以訪問容器的狀態。html
- Vector和HashTable
- Collections.synchronizedXXX方法
同步容器的問題
1. 這種方式使得對容器的訪問都串行化，嚴重下降了併發性，若是多個線程來競爭容器的鎖時，吞吐量嚴重下降
2. 對容器的多個方法的複合操做，是線程不安全的，好比一個線程負責刪除，另外一個線程負責查詢，有可能出現越界的異常

併發容器。java.util.concurrent包裏面的一系列實現java

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 <tt>Integer.MAX_VALUE</tt> elements, returns
    * <tt>Integer.MAX_VALUE</tt>.
    *
    * @return the number of key-value mappings in this map
    */
   public int size() {
       //爲了可以算準數量，會算2次，若是兩次算的不許，就鎖住再算
       final Segment<K,V>[] 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<K,V> 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 <tt>true</tt> (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方法，在隊列不知足各自條件時將產生阻塞bash

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();
       }
   }
}
複製代碼

輸出爲app

1
Wait 1 sec
2
Wait 2 sec
3
複製代碼

閉鎖ide

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<nThreads;i++){
       Thread t = new Thread(){
           public void run(){
               try {
                   startGate.await();
                   try {
                       task.run();
                   }finally {
                       endGate.countDown();
                   }
               } catch (InterruptedException e) {
                   e.printStackTrace();
               }
           }
       };
       t.start();
   }
   long start = System.nanoTime();
   startGate.countDown();
   endGate.await();
   long end=System.nanoTime();
   return end-start;
   }
}
複製代碼

啓動門使主線程可以同時釋放全部的工做線程，結束門使得主線程可以等待最後一個線程執行完工具

FutureTask。Future.get的若是任務執行完成，則當即返回，不然將阻塞直到任務完結，再返回結果或者是拋出異常

信號量，Semaphore 。它管理着一組虛擬的許可，許可的數量可經過構造函數指定，在執行操做時首先得到許可，並在使用後釋放許可，若是沒有，那麼accquire將阻塞直到有許可。

Semaphore示例

public class BoundedHashSet<T>{
   private final Set<T> set;
   private final Semaphore sem;

   public BoundedHashSet(int bound) {
       this.set = Collections.synchronizedSet(new HashSet<T>());
       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();
        }
    }
}
複製代碼

輸出以下