Java 之 JUC

1. JUC 简介

在 Java 5.0 提供了 java.util.concurrent(简称JUC)包，在此包中增加了在并发编程中很常用的工具类，用于定义类似于线程的自定义子系统，包括线程池，异步 IO 和轻量级任务框架；还提供了设计用于多线程上下文中的 Collection 实现等；

2. volatile 关键字

volatile 关键字：当多个线程进行操作共享数据时，可以保证内存中的数据是可见的；相较于 synchronized 是一种较为轻量级的同步策略；
volatile 不具备”互斥性”;
volatile 不能保证变量的”原子性”;

// 使用 volatile 之前
public class Main{

    public static void main(String[] args){
        ThreadDemo td = new ThreadDemo();
        new Thread(td).start();
        while(true){
            if(td.isFlag()){
                System.out.println("########");
                break;
            }
        }
    }
}
class ThreadDemo implements Runnable{
    private boolean flag = false;
    public void run(){
        try{
            // 该线程 sleep(200), 导致了程序无法执行成功
            Thread.sleep(200);
        }catch(InterruptedException e){
            e.printStackTrace();
        }
        flag = true;
        Sytem.out.println("flag="+isFlag());
    }
    public boolean isFlag(){
        return flag;
    }
    public void setFlag(boolean flag){
        this.flag = flag;
    }
}

图中可以看出，这个程序并不会自己停止，处于堵塞状态

2.1 解决方式一: 同步锁

public class Main{    
    public static void main(String[] args) {

        ThreadDemo td = new ThreadDemo();
        new Thread(td).start();
        while(true){
            //加入了synchronized
            synchronized (td){
                if(td.isFlag()){
                    System.out.println("########");
                    break;
                }
            }
        }
    }
}

2.2 解决方式二: volatile 关键字

class ThreadDemo implements Runnable{
    //使用了volatile 关键字
    private volatile boolean flag = false;

    @Override
    public void run(){
        try{ 
            Thread.sleep(200);
        }catch(InterruptedException e){
            e.printStackTrace();
        }

        flag = true;

        System.out.println("flag="+isFlag());
    }

    public boolean isFlag(){
        return flag;
    }

    public void setFlag(boolean flag){
        this.flag = flag;
    }
}

3.i++ 的原子性问题

//`i++`的操作实际上分为三个步骤: "读-改-写";
i = i++;

//执行步骤:
int t = i;
i = i + 1;
i = t;

原子性: 就是”i++”的”读-改-写”是不可分割的三个步骤;
原子变量: JDK1.5 以后,java.util.concurrent.atomic包下,提供了常用的原子变量;
- 原子变量中的值,使用 volatile 修饰,保证了内存可见性;
- CAS(Compare-And-Swap) 算法保证数据的原子性;

public class TestAtomicDemo {
    public static void main(String[] args){

        AtomicDemo ad = new AtomicDemo();

        for(int i=0; i < 20; i++){
            new Thread(ad).start();
        }
    }
}

class AtomicDemo implements Runnable{
    private int serialNumber = 0;

    @Override
    public void run(){
        try{
            Thread.sleep(200);
        }catch(InterruptedException e){
        }
        System.out.println(Thread.currentThread().getName() + ":" + getSerialNumber());
    }
    public int getSerialNumber(){
        return serialNumber++;
    }
}

// 改进: 使用原子变量
class AtomicDemo implements Runnable{
    private AtomicInteger serialNumber = new AtomicInteger();
    public void run(){
        try{
            Thread.sleep(200);
        }catch(InterruptedException e){
        }
        System.out.println(Thread.currentThread().getName()+":"+getSerialNumber());
    }
    public int getSerialNumber(){
        // 自增运算
        return serialNumber.getAndIncrement();
    }
}

这就解决了。

3.1 CAS 算法

CAS(Compare-And-Swap) 算法是硬件对于并发的支持,针对多处理器操作而设计的处理器中的一种特殊指令,用于管理对共享数据的并发访问;
CAS 是一种无锁的非阻塞算法的实现;
CAS 包含了三个操作数:
- 需要读写的内存值: V
- 进行比较的预估值: A
- 拟写入的更新值: B
- 当且仅当 V == A 时, V = B, 否则,将不做任何操作

// 模拟CAS 算法
class CompareAndSwap {
    private int value;
    // 获取内存值
    public synchronized int get(){
        return value;
    }
    // 无论更新成功与否,都返回修改之前的内存值
    public synchronized int compareAndSwap(int expectedValue, int newValue){
        // 获取旧值
        int oldValue = value;
        if(oldValue == expectedValue){
            this.value = newValue;
        }
        // 返回修改之前的值
        return oldValue;
    }
    // 判断是否设置成功
    public synchronized boolean compareAndSet(int expectedValue, int newValue){
        return expectedValue == compareAndSwap(expectedValue, newValue);
    }
}

class TestCompareAndSwap{
    public static void main(String[] args){
        final CompareAndSwap cas = new CompareAndSwap();
        for(int i=0; i<10; i++){
            // 创建10个线程,模拟多线程环境
            new Thread(new Runnable(){
                @Override
                public void run(){
                    int expectedValue = cas.get();
                    boolean b = cas.compareAndSet(expectedValue, (int)(Math.random()*100));
                    System.out.println(b);
                }
            }).start();
        }
    }
}

4. 并发容器类

Java 5.0 在 java.util.concurrent 包中提供了多种并发容器类来改进同步容器的性能;

4.1 ConcurrentHashMap

ConcurrentHashMap 同步容器类是 Java5 增加的一个线程安全的哈希表；介于 HashMap 与 Hashtable 之间；内部采用”锁分段”机制替代Hashtable的独占锁,进而提高性能；
此包还提供了设计用于多线程上下文中的Collection实现：ConcurrentHashMap，ConcurrentSkipListMap，ConcurrentSkipListSet，CopyOnWriteArrayList和CopyOnWriteArraySet；
- 当期望许多线程访问一个给定collection时，ConcurrentHashMap通常优于同步的HashMap；
  ConcurrentSkipListMap通常优于同步的TreeMap；
- 当期望的读数和遍历远远大于列表的更新数时，CopyOnWriteArrayList优于同步的ArrayList；

4.2 CountDownLatch(闭锁)

CountDownLatch是一个同步辅助类,在完成一组正在其他线程中执行的操作之前,它允许一个或多个线程一直等待;

public class TestCountDownLatch {
    public static void main(String[] args){
        final CountDownLatch latch = new CountDownLatch(10);
        LatchDemo ld = new LatchDemo(latch);
        long start = System.currentTimeMillis();
        // 创建10个线程
        for(int i=0; i<10; i++){
            new Thread(ld).start();
        }
        try{
            latch.await();
        }catch(InterruptedException e){

        }
        long end = System.currentTimeMillis();
        System.out.println("耗费时间为:"+(end - start));
    }
}

class LatchDemo implements Runnable{
    private CountDownLatch latch;

    // 有参构造器
    public LatchDemo(CountDownLatch latch){
        this.latch = latch;
    }
    @Override
    public void run(){
        synchronized(this){
            try{
                // 打印50000以内的偶数
                for(int i=0; i<50000; i++){
                    if(i % 2 == 0){
                        System.out.println(i);
                    }
                }
            }finally{
                // 线程数量递减
                latch.countDown();
            }
        }
    }
}

5. 创建执行线程的方式三

相较于实现 Runnable 接口的方式,实现 Callable 接口类中的方法可以有返回值,并且可以抛出异常;

 class TestCallable {
    public static void main(String[] args){
        ThreadDemo1 td = new ThreadDemo1();
        // 执行 Callable 方式,需要 FutureTask 实现类的支持
        // FutureTask 实现类用于接收运算结果, FutureTask 是 Future 接口的实现类
        FutureTask<Integer> result = new FutureTask<>(td);
        new Thread(result).start();
        // 接收线程运算后的结果
        try{
            // 只有当 Thread 线程执行完成后,才会打印结果;
            // 因此, FutureTask 也可用于闭锁
            Integer sum = result.get();
            System.out.println(sum);
        }catch(InterruptedException | ExecutionException e){
            e.printStackTrace();
        }
    }
}

class ThreadDemo1 implements Callable<Integer> {

    // 需要实现的方法
    @Override
    public Integer call() throws Exception{
        // 计算 0~100 的和
        int sum = 0;
        for(int i=0; i<=100; i++){
            sum += i;
        }
        return sum;
    }
}

6. 同步锁(Lock)

Lock 接口

Lock 接口, 位于 java.util.concurrent.locks 包中, 使用该接口需要导包.
Lock 接口的出现替代了同步代码块或者同步函数, 因为同步代码块对于锁的操作(获取或释放)是隐式的。
Lock 接口将同步的隐式锁操作变成显式锁操作。同时,更为灵活，可以在一个锁上加上多个监视器。
Lock 接口中的方法:
- lock(): 获取锁
- unlock(): 释放锁, 这个动作是必须要完成的, 所以通常需要定义在 finally 代码块中
格式:

Lock lock = new ReentrantLock(); // Lock 接口的实现类
void show()
{
    try
    {
        lock.lock(); //获取锁
        // 执行代码...
    }
    finally
    {
        lock.unlock(); // 释放锁
    }
}

Condition 接口

Condition 接口的出现替代了 Object 类中的 wait(), notify(), notifyAll()方法,将这些
监视器方法单独进行了封装, 变成 Condition 监视器对象, 可以与任意锁进行组合.
常用方法:
- await(): 让线程处于冻结状态
- signal(): 唤醒一个等待线程
- signalAll(): 唤醒所有等待线程
格式:
Condition c1 = lock.newCondition(); // 新建一个监视器对象

private boolean flag = false;
// 创建一个锁对象
Lock lock = new ReentrantLock();
// 通过已有的锁获取两组监视器, 一组监视生产者, 一组监视消费者
Condition producer_con = lock.newCondition();
Condition consumer_con = lock.newCondition();
void show()
{
    lock.lock(); //获取锁
    try
    {
        while(flag)
                try{producer_con.wait();}catch(InterruptedException e){}
        // 执行代码...
        flag = true;
        consumer_con.signal();
    }
    finally
    {
        lock.unlock(); // 释放锁
    }
}

wait() 和 sleep() 的区别

wait() 可以指定时间也可以不指定时间
sleep() 必须指定时间.
在同步中, 对 CPU 的执行权和锁的处理不同
- wait(): 释放执行权, 释放锁
- sleep(): 释放执行权, 不释放锁

停止线程

run() 方法结束
怎么控制线程的任务结束呢?
- 任务中都会有循环结构, 只要控制住循环就可以结束任务
- 控制循环通常就用定义标记(条件)来完成
如果线程处于冻结状态, 无法读取标记, 如何结束呢?
- 可以使用 interrupt() 方法将线程从冻结状态强制恢复到运行状态, 让线程具备 CPU 的执行资格
- 该强制动作会发生 InterruptedException, 需要处理

言归正传：

// 使用 lock 之前
public class TestLock{
    public static void main(String[] args){
        Ticket ticket = new Ticket();
        new Thread(ticket,"1号窗口").start();
        new Thread(ticket,"2号窗口").start();
        new Thread(ticket,"3号窗口").start();
    }
}

class Ticket implements Runnable{
    private int tick = 100;
    
    public void run(){
        while(true){
            if(tick > 0){
                try{
                    Thread.sleep(200);
                }catch(InterruptedException e){}
                System.out.println(Thread.currentThread().getName()+"完成售票,余票为: "+ --tick);
            }
        }
    }
}

// 使用了 lock
class ticke implements Runnable {

    int tick = 100;
    Lock lock = new ReentrantLock();

    @Override
    public void run() {
            while (true) {
                //上锁
                lock.lock();
                try {
                    if (tick > 0) {
                        try {
                            Thread.sleep(200);
                        } catch (InterruptedException e) {}
                        System.out.println(Thread.currentThread().getName() + "完成售票,余票为: " + --tick);
                    }
                }finally {
                    lock.unlock();
                }
            }
    }
}

7. ReadWriteLock(读写锁)

public class TestReadWriteLock {
    public static void main(String[] args){
        ReadWriteLockDemo rw = new ReadWriteLockDemo();
        // 一个线程进行写
        new Thread(new Runnable(){
            @Override
            public void run(){
                rw.set((int)(Math.random()*100));
            }
        },"Write:").start();

        // 100个线程进行读操作
        for(int i=0; i<100; i++){
            new Thread(new Runnable(){
                @Override
                public void run(){
                    rw.get();
                }
            },"Read:").start();
        }
    }

}

class ReadWriteLockDemo{
    private int number = 0;
    private ReadWriteLock lock = new ReentrantReadWriteLock();
    // 读
    public void get(){
        lock.readLock().lock(); // 上锁
        try{
            System.out.println(Thread.currentThread().getName()+":"+number);
        }finally{
            lock.readLock().unlock(); // 释放锁
        }
    }
    // 写
    public void set(int number){
        lock.writeLock().lock();
        try{
            System.out.println(Thread.currentThread().getName());
            this.number = number;
        }finally{
            lock.writeLock().unlock();
        }
    }
}