JUC并发包基本使用

一、简介

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

　　JUC 常用工具类：

　　Semaphore - 信号量

　　ReentrantLock - 可重入锁。之前有做过简介使用，详见 https://www.cnblogs.com/eric-fang/p/8991208.html

　　ReadWriteLock - 读写锁

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

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

　　CyclicBarrier - 栅栏。在某一条件达成之前，所有线程均阻塞等待

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

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

　　Executor、ExecutorService、Future ： 之前有做过简介使用，详见 https://www.cnblogs.com/eric-fang/p/9004020.html

二、使用

　　2.1、信号量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.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.3、栅栏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();
    }
}