并发编程Concurrent Programming

 

包名	内容
java.util.concurrent	并发编程工具类
java.util.concurrent.Atomic	原子变量
java.util.concurrent.Lock	锁

JDK1.5 开始大牛Doug Lee为JDK带来了Concurrent包，一个基于AQS和CAS的，崇尚“无锁胜有锁”的并发编程利器。有关AQS和CAS的知识请参考PPT：AbstractQueuedSynchronizer.pptx

这里简单的为大家介绍一下concurrent包下几个开发中经常用到的并发类的用法和使用场景，以供参考。

一、LOCK：锁

1、ReentrantLock  可重入的互斥锁

具有和synchronized近似的语义，但是具有更丰富的功能，比如获取锁时提供超时控制，区分公平非公平锁，利用condition区分不同的等待线程队列以便减少锁释放过程中因为锁竞争带来的性能损失

使用场景：

需要用到互斥锁的地方

构造：

private final ReentrantLock lock = new ReentrantLock();

private final ReentrantLock lock = new ReentrantLock(boolean fair); //控制采用公平还是非公平锁，一般而言，采用默认的非公平锁实现性能更优，减少线程上下文切换

推荐用法：

（1）阻塞

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

（2）立即返回

    

private final ReentrantLock lock = new ReentrantLock();
    // ...
    public void m() {
        if(lock.tryLock()){ //get lock immediately
              try {
                       // ... method body
                   } finally {
                          lock.unlock()
                   }
        }else{// faile to get lock immediately , do something else
        }
     }

（3）超时版本

    

private final ReentrantLock lock = new ReentrantLock();
    // ...
    public void m() {
        if(lock.tryLock() || lock.tryLock(long timeout, TimeUnit unit)){
        //get lock immediately, if failed ,wait to get lock until timeout or get the lock
               try {
                       // ... method body
                   } finally {
                          lock.unlock()
                   }
        }else{// faile to get lock immediately , do something else
        }
    }

(4)Condition

  

  private final ReentrantLock lock = new ReentrantLock();
    Condition not_empty = lock.newCondition();
    not_empty.wait(); //thread wait for not_empty signal()
    not_empty.signal();// invoke the thread which is waiting for signal of not_empty
    not_empty.signalAll();// invoke all  the thread which is waiting for signal of not_empty

2、ReentrantReadWriteLock 读写锁

实现了读写锁语义，读锁是共享锁，写锁是互斥锁，写锁可降级为读锁，读锁不可升级为写锁。

使用场景：读多写少的场景，可以提高并发读的性能

ReentrantReadWriteLock有以下几个特性：

(1)公平性

非公平锁（默认） 这个和独占锁的非公平性一样，由于读线程之间没有锁竞争，所以读操作没有公平性和非公平性，写操作时，由于写操作可能立即获取到锁，所以会推迟一个或多个读操作或者写操作。因此非公平锁的吞吐量要高于公平锁。

公平锁 利用AQS的CLH队列，释放当前保持的锁（读锁或者写锁）时，优先为等待时间最长的那个写线程分配写入锁，当然前提是写线程的等待时间要比所有读线程的等待时间要长。同样一个线程持有写入锁或者有一个写线程已经在等待了，那么试图获取公平锁的（非重入）所有线程（包括读写线程）都将被阻塞，直到最先的写线程释放锁。如果读线程的等待时间比写线程的等待时间还有长，那么一旦上一个写线程释放锁，这一组读线程将获取锁。

(2)重入性

读写锁允许读线程和写线程按照请求锁的顺序重新获取读取锁或者写入锁。当然了只有写线程释放了锁，读线程才能获取重入锁。

写线程获取写入锁后可以再次获取读取锁，但是读线程获取读取锁后却不能获取写入锁。

另外读写锁最多支持65535个递归写入锁和65535个递归读取锁。

(3)锁降级

写线程获取写入锁后可以获取读取锁，然后释放写入锁，这样就从写入锁变成了读取锁，从而实现锁降级的特性。

(4)锁升级

读取锁是不能直接升级为写入锁的。因为获取一个写入锁需要释放所有读取锁，所以如果有两个读取锁视图获取写入锁而都不释放读取锁时就会发生死锁。

(5)锁获取中断

读取锁和写入锁都支持获取锁期间被中断。这个和独占锁一致。

(6)条件变量

写入锁提供了条件变量(Condition)的支持，这个和独占锁一致，但是读取锁却不允许获取条件变量，将得到一个UnsupportedOperationException异常。

(7)重入数

读取锁和写入锁的数量最大分别只能是65535（包括重入数）。

推荐用法：

（1）读写控制

 

class RWDictionary {
    private final Map<String, Data> m = new TreeMap<String, Data>();
    private final ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();
    private final Lock r = rwl.readLock();
    private final Lock w = rwl.writeLock();
  
    public Data get(String key) {
      r.lock();
      try { return m.get(key); }
      finally { r.unlock(); }
    }
    public String[] allKeys() {
      r.lock();
      try { return m.keySet().toArray(); }
      finally { r.unlock(); }
    }
    public Data put(String key, Data value) {
      w.lock();
      try { return m.put(key, value); }
      finally { w.unlock(); }
    }
    public void clear() {
      w.lock();
      try { m.clear(); }
      finally { w.unlock(); }
    }
  }

（2）锁降级

  

class CachedData {
    Object data;
    volatile boolean cacheValid;
    final ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();
  
    void processCachedData() {
      rwl.readLock().lock();
      if (!cacheValid) {
        // Must release read lock before acquiring write lock
        rwl.readLock().unlock();
        rwl.writeLock().lock();
        try {
          // Recheck state because another thread might have
          // acquired write lock and changed state before we did.
          if (!cacheValid) {
            data = ...
            cacheValid = true;
          }
          // Downgrade by acquiring read lock before releasing write lock
          rwl.readLock().lock();
        } finally {
          rwl.writeLock().unlock(); // Unlock write, still hold read
        }
      }
  
      try {
        use(data);
      } finally {
        rwl.readLock().unlock();
      }
    }
  }

二、Atomic：原子变量

1、AtomicInteger 原子整形变量

   提供了整形变量的一系列原子操作指令，CompareAndSet，incrementAndGet，decrementAndGet， getAndUpdate等等

  使用场景：线程安全的计数器等

  推荐用法：

  AtomicInteger ai=new AtomicInteger(0);

  ai.getAndIncrement();

  ai.compareAndSet(1,2); //if ai is 1 then set value to 2

2、AtomicReference原子引用

    原子引用变量，用法和和AtomicInteger类似

    推荐用法：

    

// 创建两个Person对象，它们的id分别是101和102。
Person p1 = new Person(101);
Person p2 = new Person(102);

// 新建AtomicReference对象，初始化它的值为p1对象
AtomicReference ar = new AtomicReference(p1);

// 通过CAS设置ar。如果ar的值为p1的话，则将其设置为p2。 
ar.compareAndSet(p1, p2);

3、AtomicStampedReference 带版本号的原子引用

    解决原子操作的ABA问题

推荐用法：

    AtomicStampedReference atomicStampedRef = new AtomicStampedReference(100, 0);
    //当value=100,且stampe = 0时compareAndSet操作返回成功
    atomicStampedRef.compareAndSet(100, 101, atomicStampedRef.getStamp(), atomicStampedRef.getStamp() + 1);

三、并发集合类

1、LinkedBlockingQueue 线程安全的FIFO queue, 链表实现

    利用ReentrantLock和Condition实现的一个线程安全的FIFO queue

    推荐用法：

    

BlockingQueue<T> blockingQueue = new LinkedBlockingQueue<T>()// Size: Integer#MAX_VALUE
    BlockingQueue<T> blockingQueue = new LinkedBlockingQueue<T>(int capacity);
    BlockingQueue<T> blockingQueue = new LinkedBlockingQueue<T>(Collection collection); //用集合初始化，Size: Integer#MAX_VALUE
    blockingQueue.clear();//Atomically removes all of the elements from this queue, fully lock
    blockingQueue.contains(Object o);//判断o在不在queue中，fully lock
    blockingQueue.drainTo(Collection c);//Removes all available elements from this queue and adds them to collection
    //非阻塞操作
    blockingQueue.add(E e);//add element , if no space ,throw exception
    blockingQueue.offer(E e); //add element, if no space, return false
    blockingQueue.offer(E e, long timeout, TimeUnit unit); //add element with timeout
    blockingQueue.poll();//Retrieves and removes the head of this queue, or returns null if this queue is empty.
    //阻塞操作
    blockingQueue.take();//Retrieves and removes the head of this queue, waiting if necessary until an element becomes available.
    blockingQueue.put();

2、ConcurrentHashMap

    ConcurrentHashMap相比于HashMap它是线程安全的，HashTable虽然用synchronized实现了线程安全，但是吞吐量小，而ConcurrentHashMap利用了分段锁等技术提高了并发读写性能。关于ConcurrentHashMap的实现原理请参考http://www.infoq.com/cn/articles/ConcurrentHashMap/和http://www.ibm.com/developerworks/cn/java/java-lo-concurrenthashmap/

    这里简单讲讲用法

    实现原理：

    （1）用分离锁实现多个线程间的更深层次的共享访问。

    （2）用 HashEntery 对象的不变性来降低执行读操作的线程在遍历链表期间对加锁的需求。

    （3）通过对同一个 Volatile 变量的写 / 读访问，协调不同线程间读 / 写操作的内存可见性。

    使用特点：

    （1）ConcurrentHashMap不能以null作为key或者value，这点和HashMap是不一样的

    （2）get操作不需要加锁

    （3）ConcurrentHashMap允许一边更新，一边遍历，所以遍历操作拿到的值可能是旧值，也可能是新值，取决于线程调度情况

    （4）和HashTable不一样的是，遍历操作和更新操作并发情况下，不会抛出ConcurrentModificationException异常

    （5）jdk1.8增加了几个方法search,reduce,reduceKeys等，需要了解的可以参看源码介绍

    推荐用法：

    （1）构造

    //initialCapacity可以指定初始容量大小，实际容量会是大于initialCapacity的最小的一个2的幂次方

    //loadFactor表示负载因子，当负载达到 总容量*loadFactor 时，ConcurrentHashMap需要做resize,这是一个非常损耗性能的操作，因此建议使用时一定设置一下初始容量

    //concurrencyLevel并发级别控制，可并发写的线程数的一个预估值，具体可参看详细实现

    ConcurrentHashMap concurrentHashMap = new ConcurrentHashMap(int initialCapacity,

                             float loadFactor, int concurrencyLevel);

3、CopyOnWriteArrayList

   作者本人介绍：A thread-safe variant of {@link java.util.ArrayList} in which all mutative operations ({@code add}, {@code set}, and so on) are implemented by making a fresh copy of the underlying array. This is ordinarily too costly！

   可以看出其实现原理是更新的时候重新拷贝一份原始数据，也就没有线程不安全的担忧了，但非常消耗内存。不过对于遍历或者读取操作远远大于更新操作的应用来说，性能可能会非常好。

   使用场景：

   遍历或者读取操作远远大于更新操作，还有一个特点，允许存null值

4、ConcurrentLinkedQueue

    和BlockingQueue非常相似，都是线程安全的，区别在于BlockingQueue提供了两个阻塞操作，并且ConcurrentLinkedQueue是一个无界队列

还有一些并发集合类比如BlockingDeque,ConcurrentSkipListMap用到的并不是很多，需要了解的话大家可以参看源码

四、Semaphore, CountDownLatch, CyclicBarrier

1、Semaphore  信号量

  使用场景：资源池等，协调有限资源的并发获取

  推荐用法：

    

class Pool {
    private static final int MAX_AVAILABLE = 100;
    private final Semaphore available = new Semaphore(MAX_AVAILABLE, true);
  
    public Object getItem() throws InterruptedException {
      available.acquire();
      return getNextAvailableItem(); //just for an example
    }
  
    public void putItem(Object x) {
      if (markAsUnused(x))
        available.release();
    }
  }

2、CountDownLatch 倒数计数器

    使用场景：线程阻塞直到满足某些条件才执行，CountDownLatch(1) 可以用作一个“门”，门开的时候多个线程开始运行，CountDownLatch(N) 可以将任务分成多份，分发给多个线程并行执行，全部成功以后主线程唤醒。

    推荐用法：

    (1) 门的用法

   

 class Driver { // ...
       void main() throws InterruptedException {
         CountDownLatch startSignal = new CountDownLatch(1);
         CountDownLatch doneSignal = new CountDownLatch(N);
     
         for (int i = 0; i < N; ++i) // create and start threads
           new Thread(new Worker(startSignal, doneSignal)).start();
     
         doSomethingElse();            // don't let run yet
         startSignal.countDown();      // let all threads proceed
         doSomethingElse();
         doneSignal.await();           // wait for all to finish
       }
     }
     
    class Worker implements Runnable {
       private final CountDownLatch startSignal;
       private final CountDownLatch doneSignal;
       Worker(CountDownLatch startSignal, CountDownLatch doneSignal) {
         this.startSignal = startSignal;
         this.doneSignal = doneSignal;
       }
       public void run() {
         try {
           startSignal.await();
           doWork();
           doneSignal.countDown();
         } catch (InterruptedException ex) {} // return;
       }
     
       void doWork() { ... }
     }

 (2) 任务分片

    

 class Driver2 { // ...
       void main() throws InterruptedException {
         CountDownLatch doneSignal = new CountDownLatch(N);
         Executor e = ...
     
         for (int i = 0; i < N; ++i) // create and start threads
           e.execute(new WorkerRunnable(doneSignal, i));
     
         doneSignal.await();           // wait for all to finish
       }
     }
     
     class WorkerRunnable implements Runnable {
       private final CountDownLatch doneSignal;
       private final int i;
       WorkerRunnable(CountDownLatch doneSignal, int i) {
         this.doneSignal = doneSignal;
         this.i = i;
       }
       public void run() {
         try {
           doWork(i);
           doneSignal.countDown();
         } catch (InterruptedException ex) {} // return;
       }
     
       void doWork() { ... }
     }

3、CyclicBarrier  可多次使用的栅栏

    使用场景：多个线程互相等待直到都运行到一个公共的“栅栏”，“栅栏”打开，然后所有线程一起开始执行后续代码，Cyclic表示栅栏可以多次使用

    推荐用法：

    

class Solver {
    final int N;
    final float[][] data;
    final CyclicBarrier barrier;
  
    class Worker implements Runnable {
      int myRow;
      Worker(int row) { myRow = row; }
      public void run() {
        while (!done()) {
          processRow(myRow);
  
          try {
            barrier.await(); //方法返回一个int值，代表当前线程第几个到达“栅栏”， 0表示最后一个
          } catch (InterruptedException ex) {
            return;
          } catch (BrokenBarrierException ex) {
            return;
          }
        }
      }
    }
  
    public Solver(float[][] matrix) {
      data = matrix;
      N = matrix.length;
      Runnable barrierAction =
        new Runnable() { public void run() { mergeRows(...); }};
      barrier = new CyclicBarrier(N, barrierAction); //构造的时候可以指定一个Runnable，当“栅栏”打开时，且所有线程未释放前执行
  
      List<Thread> threads = new ArrayList<Thread>(N);
      for (int i = 0; i < N; i++) {
        Thread thread = new Thread(new Worker(i));
        threads.add(thread);
        thread.start();
      }
  
      // wait until done
      for (Thread thread : threads)
        thread.join();
    }
  }

  在这个例子中，演示了如何多线程的计算一个矩阵的“merge”, 每个worker线程计算一行的merge, barrierAction中在所有worker merge完后进行所有行结果的merge，执行barrierAction的线程是最后一个到达“栅栏的”线程。

  注意：当有一个线程由于执行异常，超时，中断导致过早离开“栅栏点”，其余的线程也将失败，抛出“BrokenBarrierException”或者“InterruptedException”异常，如果线程处于等待期间，别的线程调用了barrier的reset方法，等待线程抛出“BrokenBarrierException”

  

 

五、Executors框架

    Executors框架提供了一些简单易用的API供我们构造各种类型的线程池(ExecutorService)，以适应不同的应用场景。可提交Runnable, Callabe等Task，返回Future

    1、newFixedThreadPool 固定大小的线程池，无界队列

        ExecutorService executorService = Executors.newFixedThreadPool(int N);

        如果某个线程以外终结变得不可用，线程池将新建一个线程替补

        使用场景：线程需要使用某个固定大小的资源，比如连接账号

    2、newCachedThreadPool 无界线程池

       ExecutorService executorService = Executors.newCachedThreadPool();

       当线程池中有空闲线程，使用空闲线程；如果没有，新建一个新的线程

       空闲线程默认1min以后销毁

       使用场景：大量的耗CPU较短的异步任务场景，比如IO任务

    3、newSingleThreadExecutor 一个线程的线程池，无界队列

        ExecutorService executorService = Executors.newSingleThreadExecutor();

        如果线程以外终结变得不可用，线程池将新建一个线程替补

        使用场景：任务队列需串行执行

    4、newScheduledThreadPool  调度线程池

        ExecutorService executorService = Executors.newScheduledThreadPool(int corePoolSize);

        使用场景：需要延迟执行，或者重复调度的任务，submit返回ScheduledFuture 具体可以参见ScheduledExecutorService的API

    如果Executors框架不能满足你的需求，可以使用ThreadPoolExecutor 类，它提供了更为细致的线程池管理，例如可配置线程池核心数，线程空闲超时，线程池队列，拒绝策略，线程工厂等

六、Fork-Join轻量级任务执行框架

    JDK1.7开始提供了一个并行分治开发的利器Fork-Join框架，它通过Work-Stealing思想和双端队列Deque让并行计算得以实现。有关Fork-Join框架的具体实现细节请参看API文档和源码

    特点：

    （1）非常适用于做递归计算的并行化

    （2）建议任务CPU密集型，线程最好不要长时间阻塞（比如阻塞型IO任务），也不要使用synchornized进行线程同步和过多的访问共享变量，否则无法保证性能

    （3）对于特别小的任务，体现不出性能提升，可能还不如单线程性能，因为线程调度和竞争需要损耗性能

    （4）任务执行过程中会将CPU打满，使用时注意不要影响到业务处理

    使用场景：

        递归计算的并行化，如对链表的统一操作，求和，排序等

    推荐用法：

    （1）不返回结果用RecursiveAction

      

  public class Task extends RecursiveAction {
            private static final long serialVersionUID = 1L;
            // These attributes will determine the block of products this task has to
            // process.
            private List<Product> products;
            private int first;
            private int last;
            // store the increment of the price of the products
            private double increment;
            public Task(List<Product> products, int first, int last, double increment) {
                super();
                this.products = products;
                this.first = first;
                this.last = last;
                this.increment = increment;
            }
            @Override
            protected void compute() {
                if (last - first < 10) {
                    updatePrices();
                } else {
                    int middle = (first + last) / 2;
                    System.out.printf("Task: Pending tasks:%s
", getQueuedTaskCount());
                    Task t1 = new Task(products, first, middle + 1, increment);
                    Task t2 = new Task(products, middle + 1, last, increment);
                    invokeAll(t1, t2);
                }
            }
            private void updatePrices() {
                for (int i = first; i < last; i++) {
                    Product product = products.get(i);
                    product.setPrice(product.getPrice()          (1 + increment));
                }
            }
            public static void main(String[] args) {
                ProductListGenerator productListGenerator = new ProductListGenerator();
                List<Product> products = productListGenerator.generate(10000); //get a list to be processed
                Task task = new Task(products, 0, products.size(), 0.2);
                ForkJoinPool pool = new ForkJoinPool();
                pool.execute(task);
                do {
                    System.out.printf("Main: Thread Count: %d
",
                            pool.getActiveThreadCount());
                    System.out.printf("Main: Thread Steal: %d
", pool.getStealCount());
                    System.out.printf("Main: Parallelism: %d
", pool.getParallelism());
                    try {
                        TimeUnit.MILLISECONDS.sleep(5);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                } while (!task.isDone());
                 
                pool.shutdown();
                 
                if(task.isCompletedNormally()) { //检查任务或者子任务是否出现异常
                    System.out.printf("Main: The process has completed normally.
");
                }
                 
                for(Product product : products) {
                    if(product.getPrice() != 12) {
                        System.out.printf("Product %s: %f
",product.getName(),product.getPrice());
                    }
                }
                 
                System.out.println("Main: End of the program.
");
            }
        }

    （2）需返回结果用RecursiveTask

         

public class ForkJoinTest {
            public class Fibonacci extends RecursiveTask<Integer> {
                final int n;
                Fibonacci(int n) { this.n = n; }
                public Integer compute() {
                    if (n <= 1)
                        return n;
                    Fibonacci f1 = new Fibonacci(n - 1);
                    f1.fork();
                    Fibonacci f2 = new Fibonacci(n - 2);
                    return f2.compute() + f1.join();
                }
            }
            @Test
            public void test() throws ExecutionException, InterruptedException {
                ForkJoinPool pool = new ForkJoinPool();
                Fibonacci fibonacci = new Fibonacci(30);
                pool.execute(fibonacci);
                do {
                    System.out.printf("Main: Thread Count: %d
",
                            pool.getActiveThreadCount());
                    System.out.printf("Main: Thread Steal: %d
", pool.getStealCount());
                    System.out.printf("Main: Parallelism: %d
", pool.getParallelism());
                    try {
                        TimeUnit.MILLISECONDS.sleep(5);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                } while (!fibonacci.isDone());
                System.out.println("Main: End of the program.result:{}
" + fibonacci.get());
            }
        }

    异常处理：

        Fork-Join Task中不能抛出已检查的异常，可以抛出一个未受检查的异常，因为compute()方法没有异常声明，当异常出现，程序不会结束执行而是会继续异常的执行完毕，从控制台也无法看到任何异常日志，可以通过isCompletedAbnormally()或者isCompletedNormally()判断任务或者子任务是否发生异常，通过getException()拿到异常。另外，当在任务中catch到异常时可以通过completeExceptionally()方法主动停止任务以及相关任务的执行。