JUC并发包基本使用

1、简介

　　传统的Java多线程开发中，wait、notify、synchronized等若是不注意使用的话，很容易引发死锁、脏读问题。Java1.5 版本开始增长 java.util.concurrent 并发编程包，简化了多线程开发难度。添加了不少的多线程操做工具类，可根据实际需求去选择使用。

　　JUC 经常使用工具类：

　　Semaphore - 信号量

　　ReentrantLock - 可重入锁。以前有作过简介使用，详见 http://www.javashuo.com/article/p-tklmsesy-kk.html

　　ReadWriteLock - 读写锁

　　BlockingQueue - 阻塞队列。详见 https://www.cnblogs.com/eric-fang/p/8989860.html

　　CountDownLatch - 计数器。在计数器归零后，容许以前阻塞的若干线程继续执行

　　CyclicBarrier - 栅栏。在某一条件达成以前，全部线程均阻塞等待

　　AtomicXXXXXXX - 原子操做类，常见的有：AtomicInteger、AtomicLong、AtomicBoolean。

　　TimeUnit - 时间枚举类，提供一些时间的便捷操做

　　Executor、ExecutorService、Future ： 以前有作过简介使用，详见 http://www.javashuo.com/article/p-wgyamqkp-kh.html

2、使用

　　2.一、信号量Semaphore 

　　　　通常用于限定同时访问操做的线程数量。例如：有时候能够很好的限制公共资源的使用，例如若是开启几十个线程去读取一些文件，而后读取到的数据须要入库的话，因为数据库的链接资源是稀缺资源，可能远小于读取文件的线程数，这时候能够利用信号量去限制每次并发获取数据库链接资源的线程数。

　　　　以下示例代码，虽然同时有10个线程执行，可是只能容许2个线程的并发执行。

package com.cfang.prebo.thread; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; import lombok.extern.slf4j.Slf4j; @Slf4j public class TestSemaphore2 { private static Semaphore semaphore = new Semaphore(2); private static ExecutorService executorService = Executors.newFixedThreadPool(10); public static void main(String[] args) { for(int i = 0; i < 10; i++) { executorService.execute(new Runnable() { @Override public void run() { try { //申请通行证
 semaphore.acquire(); // 模拟业务逻辑
                        TimeUnit.SECONDS.sleep(2); log.info("{} 处理完成", Thread.currentThread().getName()); //释放通行证
 semaphore.release(); } catch (InterruptedException e) { e.printStackTrace(); } } }); } executorService.shutdown(); } }

　　2.二、计数器CountDownLatch 

 　　　　同步计数器，构造方法传值，用来限定计数器的次数。

　　　　countDown方法每次调用，计数器值减 1。CountDownLatch会一直阻塞着调用await方法的线程，直到计数器值变为0。

package com.cfang.prebo.thread; import java.util.Random; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import lombok.extern.slf4j.Slf4j; @Slf4j public class TestCountDownLatch { private static CountDownLatch countDownLatch = new CountDownLatch(4); private static AtomicInteger integerVal = new AtomicInteger(); public static void main(String[] args) throws Exception{ for(int i = 0; i < 4; i++) { new Thread(new Runnable() { @Override public void run() { //业务处理逻辑
                    try { int size = new Random().nextInt(100); integerVal.getAndAdd(size); TimeUnit.SECONDS.sleep(2); log.info("{} 处理完成，{}", Thread.currentThread().getName(), size); } catch (InterruptedException e) { e.printStackTrace(); } countDownLatch.countDown(); } }, "thread-" + i).start(); } String threadName = Thread.currentThread().getName(); log.info("{} thread waiting...", threadName); countDownLatch.await(); log.info("{} doing, value: {}",threadName, integerVal.get()); } }

　　2.三、栅栏CyclicBarrier 

　　栅栏屏障，构造方法传值来设定一个阈值。线程调用 await 方法的时候，线程就会被阻塞。当阻塞的线程数达到阈值的时候，全部阻塞线程所有放行。可重置重复使用。

package com.cfang.prebo.thread; import java.util.Random; import java.util.concurrent.CyclicBarrier; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import lombok.extern.slf4j.Slf4j; @Slf4j public class TestCyclicBarrier implements Runnable{ private CyclicBarrier barrier = new CyclicBarrier(4, this); private static AtomicInteger integerVal = new AtomicInteger(); public void count() { for(int i = 0; i < 4; i++) { new Thread(new Runnable() { @Override public void run() { //业务处理逻辑
                    try { int size = new Random().nextInt(100); integerVal.getAndAdd(size); TimeUnit.SECONDS.sleep(2); log.info("{} 处理完成，{}", Thread.currentThread().getName(), size); barrier.await(); } catch (Exception e) { e.printStackTrace(); } } }, "thread-" + i).start(); } } @Override public void run() { //业务逻辑处理完成后调用
        log.info("{} 统计完成，{}", Thread.currentThread().getName(), integerVal.get()); } public static void main(String[] args) { TestCyclicBarrier testCyclicBarrier = new TestCyclicBarrier(); testCyclicBarrier.count(); } }