JavaSE 多线程：JUC 核心组件介绍

Callable 接口的实现类对象重写 call() 方法，该方法的返回值由泛型参数指定；我们说的'有返回值的任务'，返回值就是这个 call() 方法的返回值，而承载这个任务的是 Callable 接口的实现类对象。

示例代码定义了一个从 1 加到 100 的计算任务，借助 FutureTask 作为 Callable 和 Thread 之间的桥梁，通过 Thread 类创建并启动线程，实现多线程并发执行该 Callable 任务，最终还能通过 FutureTask 获取任务的返回结果。

这样设计的目的是实现解耦合，Thread 类就是实现多线程，和具体的任务剥离开，Thread 不再关心任务是什么样的。

2. ReentrantLock

可重入锁，可以认为是和 synchronized 并列的锁，相比于 synchronized 更加复杂一点。

它是基于 AQS 框架实现的。

AQS：JUC 包中所有同步器（锁、信号量等）的基础骨架，它封装了同步状态的管理（通过一个 volatile int state 变量，reentrantlock 用它记录锁的持有次数）；线程排队等待的逻辑（基于双向链表的 CLH 队列，管理等待获取锁的线程）；线程的阻塞/唤醒（借助 LockSupport 工具类的 park() / unpark() 方法）

ReentrantLock 内部定义了静态内部类 NonfairSync / FairSync，这两个类直接继承自 AQS，并重写了 AQS 的核心方法，从而实现公平锁和非公平锁。

以下是部分源码示例：

/**
 * Sync object for non-fair locks
 */
static final class NonfairSync extends Sync {
    private static final long serialVersionUID = 7316153563782823691L;
    final boolean initialTryLock() {
        Thread current = Thread.currentThread();
        if (compareAndSetState(0, 1)) {
            // first attempt is unguarded
            setExclusiveOwnerThread(current);
            return true;
        } else if (getExclusiveOwnerThread() == current) {
            int c = getState() + 1;
            if (c < 0)
                throw new Error("Maximum lock count exceeded");
            setState(c);
            return true;
        } else
            return false;
    }
    /**
     * Acquire for non-reentrant cases after initialTryLock prescreen
     */
    protected final boolean tryAcquire(int acquires) {
        if (getState() == 0 && compareAndSetState(0, acquires)) {
            setExclusiveOwnerThread(Thread.currentThread());
            return true;
        }
        return false;
    }
}

/**
 * Sync object for fair locks
 */
static final class FairSync extends Sync {
    private static final long serialVersionUID = -3000897897090466540L;
    /**
     * Acquires only if reentrant or queue is empty.
     */
    final boolean initialTryLock() {
        Thread current = Thread.currentThread();
        int c = getState();
        if (c == 0) {
            if (!hasQueuedThreads() && compareAndSetState(0, 1)) {
                setExclusiveOwnerThread(current);
                return true;
            }
        } else if (getExclusiveOwnerThread() == current) {
            if (++c < 0)
                throw new Error("Maximum lock count exceeded");
            setState(c);
            return true;
        }
        return false;
    }
    /**
     * Acquires only if thread is first waiter or empty
     */
    protected final boolean tryAcquire(int acquires) {
        if (getState() == 0 && !hasQueuedPredecessors() && compareAndSetState(0, acquires)) {
            setExclusiveOwnerThread(Thread.currentThread());
            return true;
        }
        return false;
    }
}

ReentrantLock 使用 lock() 和 unlock() 加锁解锁，使用时创建一个锁对象，多个线程只有针对同一个锁对象进行加锁时才能实现互斥。

public class Demo {
    private static int count = 0;
    public static void main(String[] args) throws InterruptedException {
        ReentrantLock locker = new ReentrantLock();
        Thread t1 = new Thread(new Runnable() {
            @Override
            public void run() {
                for (int i = 0; i < 5000; i++) {
                    // 加速
                    locker.lock();
                    try {
                        count++;
                    } finally {
                        // 解锁，放在 finally 块中确保一定会执行
                        locker.unlock();
                    }
                }
            }
        });
        Thread t2 = new Thread(new Runnable() {
            @Override
            public void run() {
                for (int i = 0; i < 5000; i++) {
                    locker.lock();
                    count++;
                    locker.unlock();
                }
            }
        });
        t1.start();
        t2.start();
        t1.join();
        t2.join();
        System.out.println(count);
    }
}

由于它需要调用 unlock 函数手动释放锁，所以需要注意 unlock 不被调用的问题，这里是将 unlock 放在了 finally 块中。

2.1 与 synchronized 的区别

1. synchronized 是关键字，是 JVM 内部通过 C++ 实现的，而 ReentrantLock 是标准库中的类（Java 实现的）
2. synchronized 通过代码块控制加锁解锁，ReentrantLock 需要通过调用 lock/unlock 方法，需要注意 unlock 方法不被调用的问题
3. ReentrantLock 还提供了一个 tryLock() 方法，不会阻塞，加锁成功返回 true，失败返回 false，调用者根据返回值决定接下来做什么。这个方法还可以设置超时时间
4. ReentrantLock 提供了公平锁的实现，在调用构造方法时传入参数 true 即可，不传默认为 false
5. ReentrantLock 中搭配的等待通知机制是 Condition 类，相比 wait 和 notify 来说功能更强大一点（可以支持多个等待队列，有更丰富的等待方式，与 Lock 体系高度统一，可以唤醒特定 Condition 队列中的线程等）

3. CountDownLatch

锁存器

我们知道，使用多线程本身目的是为了将一个大的任务分成若干子任务，使用多线程来执行这些任务从而提高并发能力和资源利用率。

那么如何衡量这些任务是否执行完毕，整个任务是否执行完毕呢？

此时我们用到的就是这个 CountDownLatch 类。

CountDownLatch 通过一个计数器来控制线程的等待与唤醒，计数器初始值表示需要等待的时间数量，当计数器减为 0 时，所有在 await 上等待的线程被唤醒。

同样来举个例子：

public class Demo34 {
    public static void main(String[] args) throws InterruptedException {
        // 总共十个子任务提交到线程池中执行
        CountDownLatch latch = new CountDownLatch(10);
        // 容量为 10 的线程池
        ExecutorService threadPool = Executors.newFixedThreadPool(10);
        for (int i = 0; i < 10; i++) {
            int id = i;
            threadPool.submit(() -> {
                // 每个循环相隔 1 秒
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                // 每个线程打印 id 并输出
                System.out.println("子任务执行：" + id);
                latch.countDown();
            });
        }
        // 等待所有子任务执行完
        latch.await();
        System.out.println("所有任务执行完毕...");
    }
}

原子类和 CountDownLatch 都是为特定并发场景设计的工具，在对应场景下可以提供非常高效和简洁的解决方案，但并不是用来解决所有并发问题的通用工具。