文章目录

    • 5.5.共享模型之工具
      • 5.5.1. 线程池
      • 5.5.2. J.U.C
        • 5.5.2.1. AQS 原理
        • 5.5.2.2. ReentrantLock 原理
        • 5.5.2.3. 读写锁
        • 5.5.2.4.Semaphore
        • 5.5.2.5.CountdownLatch
        • 5.5.2.6.CyclicBarrier
        • 5.5.2.7.线程安全集合类概述
        • 5.5.2.8.ConcurrentHashMap
        • 5.5.2.9.BlockingQueue
        • 5.5.2.10. ConcurrentLinkedQueue
        • 5.5.2.10. CopyOnWriteArrayList
  • 总结

5.5.共享模型之工具

5.5.1. 线程池

1、自定义线程池（TODO手写）

  • 步骤1：自定义拒绝策略接口

    @FunctionalInterface // 拒绝策略，任务定义为泛型，因为不一定是 Runable，也有可能是 Callable
interface RejectPolicy<T> {void reject(BlockingQueue<T> queue, T task);
}

  • 步骤2：自定义任务队列

    @Slf4j(topic = "c.BlockingQueue")
class BlockingQueue<T> {// 1. 任务队列private Deque<T> queue = new ArrayDeque<>();// 2. 锁private ReentrantLock lock = new ReentrantLock();// 3. 生产者条件变量private Condition fullWaitSet = lock.newCondition();// 4. 消费者条件变量private Condition emptyWaitSet = lock.newCondition();// 5. 容量private int capcity;public BlockingQueue(int capcity) {this.capcity = capcity;}// 带超时阻塞获取public T poll(long timeout, TimeUnit unit) {lock.lock();try {// 将 timeout 统一转换为 纳秒long nanos = unit.toNanos(timeout);while (queue.isEmpty()) {try {// 返回值是剩余时间if (nanos <= 0) {return null;}nanos = emptyWaitSet.awaitNanos(nanos);} catch (InterruptedException e) {e.printStackTrace();}}T t = queue.removeFirst();fullWaitSet.signal();return t;} finally {lock.unlock();}}// 阻塞获取public T take() {lock.lock();try {while (queue.isEmpty()) {try {emptyWaitSet.await();} catch (InterruptedException e) {e.printStackTrace();}}T t = queue.removeFirst();fullWaitSet.signal();return t;} finally {lock.unlock();}}// 阻塞添加public void put(T task) {lock.lock();try {while (queue.size() == capcity) {try {log.debug("等待加入任务队列 {} ...", task);fullWaitSet.await();} catch (InterruptedException e) {e.printStackTrace();}}log.debug("加入任务队列 {}", task);queue.addLast(task);emptyWaitSet.signal();} finally {lock.unlock();}}// 带超时时间阻塞添加public boolean offer(T task, long timeout, TimeUnit timeUnit) {lock.lock();try {long nanos = timeUnit.toNanos(timeout);while (queue.size() == capcity) {try {if(nanos <= 0) {return false;}log.debug("等待加入任务队列 {} ...", task);nanos = fullWaitSet.awaitNanos(nanos);} catch (InterruptedException e) {e.printStackTrace();}}log.debug("加入任务队列 {}", task);queue.addLast(task);emptyWaitSet.signal();return true;} finally {lock.unlock();}}public int size() {lock.lock();try {return queue.size();} finally {lock.unlock();}}public void tryPut(RejectPolicy<T> rejectPolicy, T task) {lock.lock();try {// 判断队列是否满if(queue.size() == capcity) {rejectPolicy.reject(this, task);} else {  // 有空闲log.debug("加入任务队列 {}", task);queue.addLast(task);emptyWaitSet.signal();}} finally {lock.unlock();}}
}

  • 步骤3：自定义线程池

    @Slf4j(topic = "c.ThreadPool")
class ThreadPool {// 任务队列private BlockingQueue<Runnable> taskQueue;// 线程集合，非线程安全private HashSet<Worker> workers = new HashSet<>();// 核心线程数private int coreSize;// 获取任务时的超时时间private long timeout;private TimeUnit timeUnit;// 构建线程池的时候，就应当确定拒绝策略private RejectPolicy<Runnable> rejectPolicy;// 执行任务public void execute(Runnable task) {// 当任务数没有超过 coreSize 时，直接交给 worker 对象执行// 如果任务数超过 coreSize 时，加入任务队列暂存，执行拒绝策略// 这段代码保护起来线程安全synchronized (workers) {if(workers.size() < coreSize) {Worker worker = new Worker(task);log.debug("新增 worker{}, {}", worker, task);workers.add(worker);worker.start();} else {//                taskQueue.put(task);// 将拒绝策略全体下放，下放给调用者，也就是函数式接口，不写死，策略模式// 1) 死等// 2) 带超时等待// 3) 让调用者放弃任务执行// 4) 让调用者抛出异常// 5) 让调用者自己执行任务taskQueue.tryPut(rejectPolicy, task);}}}public ThreadPool(int coreSize, long timeout, TimeUnit timeUnit, int queueCapcity, RejectPolicy<Runnable> rejectPolicy) {this.coreSize = coreSize;this.timeout = timeout;this.timeUnit = timeUnit;this.taskQueue = new BlockingQueue<>(queueCapcity);this.rejectPolicy = rejectPolicy;}// 线程对象封装class Worker extends Thread{private Runnable task;// 初始化时的任务对象public Worker(Runnable task) {this.task = task;}@Overridepublic void run() {// 执行任务// 1) 当 task 不为空，执行任务// 2) 当 task 执行完毕，再接着从任务队列获取任务并执行
//            while(task != null || (task = taskQueue.take()) != null) {// 短路或的用法while(task != null || (task = taskQueue.poll(timeout, timeUnit)) != null) {try {log.debug("正在执行...{}", task);task.run();} catch (Exception e) {e.printStackTrace();} finally {// 执行完，任务没用了task = null;}}synchronized (workers) {log.debug("worker 被移除{}", this);workers.remove(this);}}}
}

  • 步骤4：测试

    @Slf4j(topic = "c.TestPool")
public class TestPool {public static void main(String[] args) {ThreadPool threadPool = new ThreadPool(1,1000, TimeUnit.MILLISECONDS, 1, (queue, task)->{// 1. 死等
//            queue.put(task);// 2) 带超时等待
//            queue.offer(task, 1500, TimeUnit.MILLISECONDS);// 3) 让调用者放弃任务执行
//            log.debug("放弃{}", task);// 4) 让调用者抛出异常，让剩余的任务不执行了
//            throw new RuntimeException("任务执行失败 " + task);// 5) 让调用者自己执行任务，主线程将run方法当做普通方法执行task.run();});for (int i = 0; i < 4; i++) {int j = i;threadPool.execute(() -> {try {Thread.sleep(1000L);} catch (InterruptedException e) {e.printStackTrace();}log.debug("{}", j);});}}
}

2、ThreadPoolExecutor

  • 1）线程池状态

    ThreadPoolExecutor 使用 int 的高 3 位来表示线程池状态，低 29 位表示线程数量

    从数字上比较，TERMINATED > TIDYING > STOP > SHUTDOWN > RUNNING（有符号位）
    这些信息存储在一个原子变量 ctl 中，目的是将线程池状态与线程个数合二为一，这样就可以用一次 cas 原子操作进行赋值，减少 CAS 操作

  • 2）构造方法

    • corePoolSize 核心线程数目 (最多保留的线程数)
    • maximumPoolSize 最大线程数目
    • keepAliveTime 生存时间 - 针对救急线程
    • unit 时间单位 - 针对救急线程
    • workQueue 阻塞队列
    • threadFactory 线程工厂 - 可以为线程创建时起个好名字
    • handler 拒绝策略

  • 工作方式：

  • 线程池中刚开始没有线程，当一个任务提交给线程池后，线程池会创建一个新线程来执行任务。

  • 当线程数达到 corePoolSize 并没有线程空闲，这时再加入任务，新加的任务会被加入workQueue 队列排队，直到有空闲的线程。

  • 如果队列选择了有界队列，那么任务超过了队列大小时，会创建 maximumPoolSize - corePoolSize 数目的线程来救急。

  • 如果线程到达 maximumPoolSize 仍然有新任务这时会执行拒绝策略。拒绝策略 jdk 提供了 4 种实现，其它著名框架也提供了实现：

    • AbortPolicy 让调用者抛出 RejectedExecutionException 异常，这是默认策略
    • CallerRunsPolicy 让调用者运行任务
    • DiscardPolicy 放弃本次任务
    • DiscardOldestPolicy 放弃队列中最早的任务，本任务取而代之
    • Dubbo 的实现，在抛出 RejectedExecutionException 异常之前会记录日志，并 dump 线程栈信息，方便定位问题
    • Netty （网络框架）的实现，是创建一个新线程来执行任务
    • ActiveMQ 的实现，带超时等待（60s）尝试放入队列，类似我们之前自定义的拒绝策略
    • PinPoint （链路追踪的框架）的实现，它使用了一个拒绝策略链，会逐一尝试策略链中每种拒绝策略

  • 当高峰过去后，超过 corePoolSize 的救急线程如果一段时间没有任务做，需要结束节省资源，这个时间由 keepAliveTime 和 unit 来控制。

    根据这个构造方法，JDK Executors 类（工具类）中提供了众多工厂方法来创建各种用途的线程池

3）newFixedThreadPool

特点：

  • 核心线程数 == 最大线程数（没有救急线程被创建），因此也无需超时时间
  • 阻塞队列是无界的，可以放任意数量的任务
  • 评价：适用于任务量已知，相对耗时的任务
@Slf4j(topic = "c.TestThreadPool")
public class TestThreadPool {public static void main(String[] args) throws InterruptedException {ExecutorService threadPool = Executors.newFixedThreadPool(2);for (int i = 0; i < 3; i++) {int j = i;threadPool.execute(() -> {try {Thread.sleep(1000L);} catch (InterruptedException e) {e.printStackTrace();}log.debug("{}", j);});}}
}


t.setDaemon(false); 线程池中的线程是非守护线程，所以并不会结束。

@Slf4j(topic = "c.TestThreadPool")
public class TestThreadPool {public static void main(String[] args) throws InterruptedException {// 通过工厂，自己定义线程的名字ExecutorService threadPool = Executors.newFixedThreadPool(2, new ThreadFactory() {private AtomicInteger t = new AtomicInteger(1);@Overridepublic Thread newThread(Runnable r) {return new Thread(r, "my_pool_thread_" + t.getAndIncrement());}});for (int i = 0; i < 3; i++) {int j = i;threadPool.execute(() -> {try {Thread.sleep(1000L);} catch (InterruptedException e) {e.printStackTrace();}log.debug("{}", j);});}}
}

4）newCachedThreadPool

特点：

  • 核心线程数是 0， 最大线程数是 Integer.MAX_VALUE，救急线程的空闲生存时间是 60s，意味着

    • 全部都是救急线程（60s 后可以回收）
    • 救急线程可以无限创建
  • 队列采用了 SynchronousQueue 实现特点是，它没有容量，没有线程来取是放不进去的（一手交钱、一手交货，放的线程因为没有人来取从而阻塞住，像是两个线程之间交换任务的队列）
  • 评价：整个线程池表现为线程数会根据任务量不断增长，没有上限，当任务执行完毕，空闲 1分钟后释放线程。 适合任务数比较密集，但每个任务执行时间较短的情况
@Slf4j(topic = "c.TestSynchronousQueue")
public class TestSynchronousQueue {public static void main(String[] args) {SynchronousQueue<Integer> integers = new SynchronousQueue<>();new Thread(() -> {try {log.debug("putting {} ", 1);integers.put(1);log.debug("{} putted...", 1);log.debug("putting...{} ", 2);integers.put(2);log.debug("{} putted...", 2);} catch (InterruptedException e) {e.printStackTrace();}},"t1").start();sleep(1);new Thread(() -> {try {log.debug("taking {}", 1);integers.take();} catch (InterruptedException e) {e.printStackTrace();}},"t2").start();sleep(1);new Thread(() -> {try {log.debug("taking {}", 2);integers.take();} catch (InterruptedException e) {e.printStackTrace();}},"t3").start();}
}

5）newSingleThreadExecutor

使用场景：
希望多个任务排队执行（串行效果）。线程数固定为 1，任务数多于 1 时，会放入无界队列排队。任务执行完毕，这唯一的线程也不会被释放。

@Slf4j(topic = "c.TestExecutors")
public class TestExecutors {public static void main(String[] args) throws InterruptedException {// test1();test2();}public static void test2() {ExecutorService pool = Executors.newSingleThreadExecutor();pool.execute(() -> {log.debug("1");int i = 1 / 0;});pool.execute(() -> {log.debug("2");});pool.execute(() -> {log.debug("3");});}private static void test1() {ExecutorService pool = Executors.newFixedThreadPool(2, new ThreadFactory() {private AtomicInteger t = new AtomicInteger(1);@Overridepublic Thread newThread(Runnable r) {return new Thread(r, "mypool_t" + t.getAndIncrement());}});pool.execute(() -> {log.debug("1");});pool.execute(() -> {log.debug("2");});pool.execute(() -> {log.debug("3");});}
}

区别：

  • 自己创建一个单线程串行执行任务，如果任务执行失败而终止那么没有任何补救措施，而线程池还会新建一个线程，保证池的正常工作。
  • Executors.newSingleThreadExecutor() 线程个数始终为1，不能修改
    • FinalizableDelegatedExecutorService 应用的是装饰器模式，只对外暴露了 ExecutorService 接口，因此不能调用 ThreadPoolExecutor 中特有的方法
  • Executors.newFixedThreadPool(1) 初始时为1，以后还可以修改
    • 对外暴露的是 ThreadPoolExecutor 对象，可以强转后调用 setCorePoolSize 等方法进行修改
      TODO 异步编排

6）提交任务

@Slf4j(topic = "c.TestSubmit")
public class TestSubmit {public static void main(String[] args) throws ExecutionException, InterruptedException {ExecutorService pool = Executors.newFixedThreadPool(1);method1(pool);}private static void method1(ExecutorService pool) throws InterruptedException, ExecutionException {Future<String> future = pool.submit(() -> {log.debug("running");Thread.sleep(1000);return "ok";});// 主线程中执行的方法,在这里阻塞住，等待结果的返回log.debug("{}", future.get());}private static void method3(ExecutorService pool) throws InterruptedException, ExecutionException {String result = pool.invokeAny(Arrays.asList(() -> {log.debug("begin 1");Thread.sleep(1000);log.debug("end 1");return "1";},() -> {log.debug("begin 2");Thread.sleep(500);log.debug("end 2");return "2";},() -> {log.debug("begin 3");Thread.sleep(2000);log.debug("end 3");return "3";}));log.debug("{}", result);}private static void method2(ExecutorService pool) throws InterruptedException {// 接收任务的集合，并返回执行结果的集合List<Future<String>> futures = pool.invokeAll(Arrays.asList(() -> {log.debug("begin");Thread.sleep(1000);return "1";},() -> {log.debug("begin");Thread.sleep(500);return "2";},() -> {log.debug("begin");Thread.sleep(2000);return "3";}));futures.forEach( f ->  {try {log.debug("{}", f.get());} catch (InterruptedException | ExecutionException e) {e.printStackTrace();}});}
}

method1(pool); 执行结果：

method2(pool); 执行结果：

method3(pool); 执行结果：

7）关闭线程池

  • shutdown

  • shutdownNow

  • 其他方法
@Slf4j(topic = "c.TestShutDown")
public class TestShutDown {public static void main(String[] args) throws ExecutionException, InterruptedException {ExecutorService pool = Executors.newFixedThreadPool(2);Future<Integer> result1 = pool.submit(() -> {log.debug("task 1 running...");Thread.sleep(1000);log.debug("task 1 finish...");return 1;});Future<Integer> result2 = pool.submit(() -> {log.debug("task 2 running...");Thread.sleep(1000);log.debug("task 2 finish...");return 2;});Future<Integer> result3 = pool.submit(() -> {log.debug("task 3 running...");Thread.sleep(1000);log.debug("task 3 finish...");return 3;});log.debug("shutdown");pool.shutdown();
//        pool.awaitTermination(3, TimeUnit.SECONDS);
//        List<Runnable> runnables = pool.shutdownNow();log.debug("other.... {}" , runnables);}
}

pool.shutdown(); 结果演示：

pool.shutdownNow(); 结果演示：

8）模式之 Worker Thread （异步模式之工作线程）

  • 定义
    让有限的工作线程（Worker Thread）来轮流异步处理无限多的任务。也可以将其归类为分工模式，它的典型实现就是线程池，也体现了经典设计模式中的享元模式（重用对象）。

    例如，海底捞的服务员（线程），轮流处理每位客人的点餐（任务），如果为每位客人都配一名专属的服务员，那么成本就太高了（对比另一种多线程设计模式：Thread-Per-Message）

    注意，不同任务类型应该使用不同的线程池，这样能够避免饥饿，并能提升效率。

    例如，如果一个餐馆的工人既要招呼客人（任务类型A），又要到后厨做菜（任务类型B）显然效率不咋地，分成服务员（线程池A）与厨师（线程池B）更为合理，当然你能想到更细致的分工

  • 饥饿
    固定大小线程池会有饥饿现象（线程数量不足导致的饥饿，解决方法是线程池划分，不同的任务类型使用不同的线程池）

    • 两个工人是同一个线程池中的两个线程
    • 他们要做的事情是：为客人点餐和到后厨做菜，这是两个阶段的工作
      • 1）客人点餐：必须先点完餐，等菜做好，上菜，在此期间处理点餐的工人必须等待
      • 2）后厨做菜：没啥说的，做就是了
    • 比如工人A 处理了点餐任务，接下来它要等着 工人B 把菜做好，然后上菜，他俩也配合的蛮好
    • 但现在同时来了两个客人，这个时候工人A 和工人B 都去处理点餐了，这时没人做饭了，饥饿。解决方法可以增加线程池的大小，不过不是根本解决方案，还是前面提到的，不同的任务类型，采用不同的线程池，例如：
    @Slf4j(topic = "c.TestDeadLock")
public class TestStarvation {static final List<String> MENU = Arrays.asList("地三鲜", "宫保鸡丁", "辣子鸡丁", "烤鸡翅");static Random RANDOM = new Random();static String cooking() {return MENU.get(RANDOM.nextInt(MENU.size()));}public static void main(String[] args) {// 分了两类线程池ExecutorService waiterPool = Executors.newFixedThreadPool(1);ExecutorService cookPool = Executors.newFixedThreadPool(1);waiterPool.execute(() -> {log.debug("处理点餐...");Future<String> f = cookPool.submit(() -> {log.debug("做菜");return cooking();});try {// 阻塞等待log.debug("上菜: {}", f.get());} catch (InterruptedException | ExecutionException e) {e.printStackTrace();}});waiterPool.execute(() -> {log.debug("处理点餐...");Future<String> f = cookPool.submit(() -> {log.debug("做菜");return cooking();});try {log.debug("上菜: {}", f.get());} catch (InterruptedException | ExecutionException e) {e.printStackTrace();}});}
}


    ExecutorService pool = Executors.newFixedThreadPool(2); 饥饿现象：

    @Slf4j(topic = "c.TestDeadLock")
public class TestStarvation {static final List<String> MENU = Arrays.asList("地三鲜", "宫保鸡丁", "辣子鸡丁", "烤鸡翅");static Random RANDOM = new Random();static String cooking() {return MENU.get(RANDOM.nextInt(MENU.size()));}public static void main(String[] args) {ExecutorService pool = Executors.newFixedThreadPool(2);pool.execute(() -> {log.debug("处理点餐...");Future<String> f = pool.submit(() -> {log.debug("做菜");return cooking();});try {// 阻塞等待log.debug("上菜: {}", f.get());} catch (InterruptedException | ExecutionException e) {e.printStackTrace();}});pool.execute(() -> {log.debug("处理点餐...");Future<String> f = pool.submit(() -> {log.debug("做菜");return cooking();});try {log.debug("上菜: {}", f.get());} catch (InterruptedException | ExecutionException e) {e.printStackTrace();}});}
}

  • 创建多少线程合适
    过小会导致程序不能充分地利用系统资源、容易导致饥饿。过大会导致更多的线程上下文切换，占用更多内存。

    • CPU 密集型运算数据分析运算（）
      通常采用 cpu 核数 + 1 能够实现最优的 CPU 利用率，+1 是保证当线程由于页缺失故障（操作系统）或其它原因导致暂停时，额外的这个线程就能顶上去，保证 CPU 时钟周期不被浪费

    • I/O 密集型运算
      CPU 不总是处于繁忙状态，例如，当你执行业务计算时，这时候会使用 CPU 资源，但当你执行 I/O 操作时、远程 RPC 调用时，包括进行数据库操作时，这时候 CPU 就闲下来了，你可以利用多线程提高它的利用率。

      经验公式如下
      线程数 = 核数 * 期望 CPU 利用率 * 总时间(CPU计算时间+等待时间) / CPU 计算时间
      例如 4 核 CPU 计算时间是 50% ，其它等待时间是 50%，期望 cpu 被 100% 利用，套用公式
      4 * 100% * 100% / 50% = 8
      例如 4 核 CPU 计算时间是 10% ，其它等待时间是 90%，期望 cpu 被 100% 利用，套用公式
      4 * 100% * 100% / 10% = 40

  • 自定义线程池（见上）

9）任务调度线程池
希望某个任务被反复的被执行。在『任务调度线程池』功能加入之前，可以使用 java.util.Timer 来实现定时功能，Timer 的优点在于简单易用，但由于所有任务都是由同一个线程来调度，因此所有任务都是串行执行的，同一时间只能有一个任务在执行，前一个
任务的延迟或异常都将会影响到之后的任务。

输出：

使用 ScheduledExecutorService 改写（延时执行任务）：

输出：

scheduleAtFixedRate 例子（固定速率执行任务）：

输出：

scheduleAtFixedRate 例子（任务执行时间超过了间隔时间）：

输出分析：一开始，延时 1s，接下来，由于任务执行时间 > 间隔时间，间隔被『撑』到了 2s（任务的执行时间较长，影响到了执行的间隔）

scheduleWithFixedDelay 例子（单独延时的执行）：

输出分析：一开始，延时 1s，scheduleWithFixedDelay 的间隔是 上一个任务结束 <-> 延时 <-> 下一个任务开始 所以间隔都是 3s

评价 整个线程池表现为：线程数固定，任务数多于线程数时，会放入无界队列排队。任务执行完毕，这些线程也不会被释放。用来执行延迟或反复执行的任务

10）正确处理执行任务异常

  • 方法1：主动捉异常

    输出：
  • 方法2：使用 Future（Submit 方法的返回值，callable 配合 Future，有结果返回结果，有异常返回异常）

    输出：

11）应用之定时任务

如何让每周四 18:00:00 定时执行任务？

public class TestSchedule {// 如何让每周四 18:00:00 定时执行任务？public static void main(String[] args) {//  获取当前时间，线程安全，Java 8 新增的日期LocalDateTime now = LocalDateTime.now();System.out.println(now);// 获取本周四时间LocalDateTime time = now.withHour(18).withMinute(0).withSecond(0).withNano(0).with(DayOfWeek.THURSDAY);// 如果 当前时间 > 本周周四，必须找到下周周四if(now.compareTo(time) > 0) {time = time.plusWeeks(1);}System.out.println(time);// initailDelay 代表当前时间和周四的时间差// period 一周的间隔时间long initailDelay = Duration.between(now, time).toMillis();long period = 1000 * 60 * 60 * 24 * 7;ScheduledExecutorService pool = Executors.newScheduledThreadPool(1);pool.scheduleAtFixedRate(() -> {System.out.println("running...");}, initailDelay, period, TimeUnit.MILLISECONDS);}
}

12）Tomcat 线程池
Tomcat 在哪里用到了线程池呢？

  • LimitLatch 用来限流，可以控制最大连接个数（防止太多的连接将服务器压垮），类似 J.U.C 中的 Semaphore 后面再讲
  • Acceptor （死循环的线程）只负责【接收新的 socket 连接】
  • Poller （死循环的线程）只负责监听 socket channel 是否有【可读的 I/O 事件】
  • 一旦可读，封装一个任务对象（socketProcessor），提交给 Executor 线程池处理
  • Executor 线程池中的工作线程最终负责【处理请求】

Tomcat 线程池扩展了 ThreadPoolExecutor，行为稍有不同

  • 如果总线程数达到 maximumPoolSize

    • 1）这时不会立刻抛 RejectedExecutionException 异常
    • 2）而是再次尝试将任务放入队列，如果还失败，才抛出 RejectedExecutionException 异常

源码 tomcat-7.0.42

TaskQueue.java

Connector 配置

Executor 线程配置

3、Fork/Join
1）概念
Fork/Join 是 JDK 1.7 加入的新的线程池实现，它体现的是一种分治思想，适用于能够进行任务拆分的 cpu 密集型运算

所谓的任务拆分，是将一个大任务拆分为算法上相同的小任务，直至不能拆分可以直接求解。跟递归相关的一些计算，如归并排序、斐波那契数列、都可以用分治思想进行求解

Fork/Join 在分治的基础上加入了多线程，可以把每个任务的分解和合并交给不同的线程来完成，进一步提升了运算效率

Fork/Join 默认会创建与 cpu 核心数大小相同的线程池

2）使用
提交给 Fork/Join 线程池的任务需要继承 RecursiveTask（有返回值）或 RecursiveAction（没有返回值），例如下面定义了一个对 1~n 之间的整数求和的任务

public class TestForkJoin {public static void main(String[] args) {ForkJoinPool pool = new ForkJoinPool(4);System.out.println(pool.invoke(new AddTask1(5)));
//        System.out.println(pool.invoke(new AddTask3(1, 5)));}
}@Slf4j(topic = "c.AddTask")
class AddTask1 extends RecursiveTask<Integer> {int n;public AddTask1(int n) {this.n = n;}@Overridepublic String toString() {return "{" + n + '}';}// invoke 调用的方法@Overrideprotected Integer compute() {// 终止条件if (n == 1) {log.debug("join() {}", n);return n;}AddTask1 t1 = new AddTask1(n - 1);t1.fork();log.debug("fork() {} + {}", n, t1);int result = n + t1.join();log.debug("join() {} + {} = {}", n, t1, result);return result;}
}@Slf4j(topic = "c.AddTask")
class AddTask2 extends RecursiveTask<Integer> {int begin;int end;public AddTask2(int begin, int end) {this.begin = begin;this.end = end;}@Overridepublic String toString() {return "{" + begin + "," + end + '}';}@Overrideprotected Integer compute() {if (begin == end) {log.debug("join() {}", begin);return begin;}if (end - begin == 1) {log.debug("join() {} + {} = {}", begin, end, end + begin);return end + begin;}int mid = (end + begin) / 2;AddTask2 t1 = new AddTask2(begin, mid - 1);t1.fork();AddTask2 t2 = new AddTask2(mid + 1, end);t2.fork();log.debug("fork() {} + {} + {} = ?", mid, t1, t2);int result = mid + t1.join() + t2.join();log.debug("join() {} + {} + {} = {}", mid, t1, t2, result);return result;}
}


用图来表示

改进

public class TestForkJoin {public static void main(String[] args) {ForkJoinPool pool = new ForkJoinPool(4);
//        System.out.println(pool.invoke(new AddTask1(5)));System.out.println(pool.invoke(new AddTask3(1, 5)));}
}@Slf4j(topic = "c.AddTask")
class AddTask3 extends RecursiveTask<Integer> {int begin;int end;public AddTask3(int begin, int end) {this.begin = begin;this.end = end;}@Overridepublic String toString() {return "{" + begin + "," + end + '}';}@Overrideprotected Integer compute() {if (begin == end) {log.debug("join() {}", begin);return begin;}if (end - begin == 1) {log.debug("join() {} + {} = {}", begin, end, end + begin);return end + begin;}int mid = (end + begin) / 2;AddTask3 t1 = new AddTask3(begin, mid);t1.fork();AddTask3 t2 = new AddTask3(mid + 1, end);t2.fork();log.debug("fork() {} + {} = ?", t1, t2);int result = t1.join() + t2.join();log.debug("join() {} + {} = {}", t1, t2, result);return result;}
}


用图来表示

5.5.2. J.U.C

5.5.2.1. AQS 原理

1）概述
全称是 AbstractQueuedSynchronizer，是阻塞式锁和相关的同步器工具的框架。先明白AQS是一个接口，规范，这个接口定义了一系列规则，而是否要设定公平锁与否由实现它的类来决定。
特点：

  • 用 state 属性来表示资源的状态（分独占模式和共享模式），子类需要定义如何维护这个状态，控制如何获取锁和释放锁。

    • 1）getState - 获取 state 状态
    • 2）setState - 设置 state 状态
    • 3）compareAndSetState - cas 机制设置 state 状态，保证赋值时候的原子性
    • 4）独占模式是只有一个线程能够访问资源，而共享模式可以允许多个线程访问资源
  • 提供了基于 FIFO 的等待队列，类似于 Monitor 的 EntryList（C++实现，而AQS纯Java实现）
  • 条件变量来实现等待、唤醒机制，支持多个条件变量，类似于 Monitor 的 WaitSet

子类主要实现这样一些方法（默认抛出 UnsupportedOperationException）

  • tryAcquire
  • tryRelease
  • tryAcquireShared
  • tryReleaseShared
  • isHeldExclusively

获取锁的姿势。

释放锁的姿势。

2）实现不可重入锁

  • 自定义同步器

    // 独占锁  同步器类
class MySync extends AbstractQueuedSynchronizer {@Overrideprotected boolean tryAcquire(int arg) {if(compareAndSetState(0, 1)) {// 加上了锁，并设置 owner 为当前线程setExclusiveOwnerThread(Thread.currentThread());return true;}return false;}@Overrideprotected boolean tryRelease(int arg) {// 表示没有线程占用，以下两行代码存在顺序setExclusiveOwnerThread(null);// private volatile int state;  防止指令重排序，写前读后，将 setExclusiveOwnerThread(null);// 放在前面，有写屏障，共享变量写入主存，确保前面的都写到主存setState(0);return true;}@Override // 是否持有独占锁protected boolean isHeldExclusively() {return getState() == 1;}public Condition newCondition() {return new ConditionObject();}
}

  • 自定义锁
    有了自定义同步器，很容易复用 AQS ，实现一个功能完备的自定义锁

// 自定义锁（不可重入锁）
class MyLock implements Lock {private MySync sync = new MySync();@Override // 加锁（不成功会进入等待队列）acquire 尝试多次，数字都没有用上public void lock() {sync.acquire(1);}@Override // 加锁，可打断，等待过程中打断public void lockInterruptibly() throws InterruptedException {sync.acquireInterruptibly(1);}@Override // 尝试加锁（一次）public boolean tryLock() {return sync.tryAcquire(1);}@Override // 尝试加锁，带超时public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {return sync.tryAcquireNanos(1, unit.toNanos(time));}@Override // 解锁public void unlock() {sync.release(1);}@Override // 创建条件变量public Condition newCondition() {return sync.newCondition();}
}

public final void acquire(int arg) {// 尝试加锁不成功，将线程放入队列if (!tryAcquire(arg) &&acquireQueued(addWaiter(Node.EXCLUSIVE), arg))selfInterrupt();
}

测试一下：

@Slf4j(topic = "c.TestAqs")
public class TestAqs {public static void main(String[] args) {MyLock lock = new MyLock();new Thread(() -> {lock.lock();try {log.debug("locking...");sleep(1);} finally {log.debug("unlocking...");lock.unlock();}},"t1").start();new Thread(() -> {lock.lock();try {log.debug("locking...");} finally {log.debug("unlocking...");lock.unlock();}},"t2").start();}
}


不可重入测试
如果改为下面代码，会发现自己也会被挡住（只会打印一次 locking）

3）心得TODO

5.5.2.2. ReentrantLock 原理

1）非公平锁实现原理

  • 加锁解锁流程
    先从构造器开始看，默认为非公平锁实现



    NonfairSync 继承自 AQS
    没有竞争时

    第一个竞争出现时：


    Thread-1 执行了
    1） CAS 尝试将 state 由 0 改为 1，结果失败
    2）进入 tryAcquire 逻辑，这时 state 已经是1，结果仍然失败
    3）接下来进入 addWaiter 逻辑，构造 Node 队列

    • 图中黄色三角表示该 Node 的 waitStatus 状态，其中 0 为默认正常状态
    • Node 的创建是懒惰的
    • 其中第一个 Node 称为 Dummy（哑元）或哨兵，用来占位，并不关联线程。

    当前线程进入 acquireQueued 逻辑

    1）acquireQueued 会在一个死循环中不断尝试获得锁，失败后进入 park 阻塞
    2）如果自己是紧邻着 head（排第二位），那么再次 tryAcquire 尝试获取锁，当然这时 state 仍为 1，失败
    3）进入 shouldParkAfterFailedAcquire 逻辑，将前驱 node，即 head 的 waitStatus 改为 -1，这次返回 false



    4）shouldParkAfterFailedAcquire 执行完毕回到 acquireQueued ，再次 tryAcquire 尝试获取锁，当然这时state 仍为 1，失败
    5） 当再次进入 shouldParkAfterFailedAcquire 时，这时因为其前驱 node 的 waitStatus 已经是 -1，这次返回 true
    6）进入 parkAndCheckInterrupt， Thread-1 park（灰色表示）

    再次有多个线程经历上述过程竞争失败，变成这个样子：

    Thread-0 释放锁，进入 tryRelease 流程，如果成功
    1）设置 exclusiveOwnerThread 为 null
    2）state = 0

    可重入的情况有可能 tryRelease 为 false


    当前队列不为 null，并且 head 的 waitStatus = -1，进入 unparkSuccessor 流程。
    找到队列中离 head 最近的一个 Node（没取消的），unpark 恢复其运行，本例中即为 Thread-1。（队列内是公平的，不公平体现在入队列的时候就有机会取获得锁）

    回到 Thread-1 的 acquireQueued 流程。


    如果加锁成功（没有竞争），会设置
    1）exclusiveOwnerThread 为 Thread-1，state = 1
    2）head 指向刚刚 Thread-1 所在的 Node，该 Node 清空 Thread
    3）原本的 head 因为从链表断开，而可被垃圾回收

    如果这时候有其它线程来竞争（非公平的体现），例如这时有 Thread-4 来了

    如果不巧又被 Thread-4 占了先
    1）Thread-4 被设置为 exclusiveOwnerThread，state = 1
    2）Thread-1 再次进入 acquireQueued 流程，获取锁失败，重新进入 park 阻塞

  • 加锁源码（TOSEE 文档中的源码部分P46）

  • 解锁源码（TOSEE 文档中的源码部分P49）

2、可重入原理

3、可打断原理
1）不可打断模式
在此模式下，即使它被打断，仍会驻留在 AQS 队列中，一直要等到获得锁后方能得知自己被打断了。


2）可打断模式

4、公平锁实现原理


5、条件变量实现原理
每个条件变量其实就对应着一个等待队列，其实现类是 ConditionObject。

1）await 流程
开始 Thread-0 持有锁，调用 await，进入 ConditionObject 的 addConditionWaiter 流程。
创建新的 Node 状态为 -2（Node.CONDITION），关联 Thread-0，加入等待队列尾部。



接下来进入 AQS 的 fullyRelease 流程，释放同步器上的锁。




unpark AQS 队列中的下一个节点，竞争锁，假设没有其他竞争线程，那么 Thread-1 竞争成功。

park 阻塞 Thread-0。

2）signal 流程
假设 Thread-1 （锁的持有者）要来唤醒 Thread-0


进入 ConditionObject 的 doSignal 流程，取得等待队列中第一个 Node，即 Thread-0 所在 Node。



执行 transferForSignal 流程，将该 Node 加入 AQS 队列尾部，将 Thread-0 的 waitStatus 改为 0，Thread-3 的 waitStatus 改为 -1

Thread-1 释放锁，进入 unlock 流程，略。
源码（TOSEE 文档中的源码部分P57）

5.5.2.3. 读写锁

1、 ReentrantReadWriteLock
当读操作远远高于写操作时，这时候使用 读写锁 让 读-读 可以并发，提高性能。 类似于数据库中的 select ...from ... lock in share mode
提供一个 数据容器类 内部分别使用读锁保护数据的 read() 方法，写锁保护数据的 write() 方法。
演示：

@Slf4j(topic = "c.DataContainer")
class DataContainer {private Object data;private ReentrantReadWriteLock rw = new ReentrantReadWriteLock();private ReentrantReadWriteLock.ReadLock r = rw.readLock();private ReentrantReadWriteLock.WriteLock w = rw.writeLock();public Object read() {log.debug("获取读锁...");r.lock();try {log.debug("读取");sleep(1);return data;} finally {log.debug("释放读锁...");r.unlock();}}public void write() {log.debug("获取写锁...");w.lock();try {log.debug("写入");sleep(1);} finally {log.debug("释放写锁...");w.unlock();}}
}

测试 读锁-读锁 可以并发（加读锁的原因是因为为了让其他线程来并发读。防止其他线程来写）

@Slf4j(topic = "c.TestReadWriteLock")
public class TestReadWriteLock {public static void main(String[] args) throws InterruptedException {DataContainer dataContainer = new DataContainer();new Thread(() -> {dataContainer.read();}, "t1").start();new Thread(() -> {dataContainer.read();}, "t2").start();}
}

输出结果，从这里可以看到 Thread-0 锁定期间，Thread-1 的读操作不受影响。

测试 读锁-写锁 相互阻塞。

@Slf4j(topic = "c.TestReadWriteLock")
public class TestReadWriteLock {public static void main(String[] args) throws InterruptedException {DataContainer dataContainer = new DataContainer();new Thread(() -> {dataContainer.read();try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}}, "t1").start();Thread.sleep(10);new Thread(() -> {dataContainer.write();}, "t2").start();}
}

输出结果：

写锁-写锁 也是相互阻塞的，这里就不测试了。

注意事项

  • 读锁不支持条件变量
  • 重入时升级不支持：即持有读锁的情况下去获取写锁，会导致获取写锁永久等待
  • 重入时降级支持：即持有写锁的情况下去获取读锁

2、应用之缓存
1）缓存更新策略

更新时，是先清缓存还是先更新数据库
先清缓存（导致数据库与缓存不一致）

先更新数据库（做不到强一致，但能最终一致）

补充一种情况，假设查询线程 A 查询数据时恰好缓存数据由于时间到期失效，或是第一次查询。这种情况的出现几率非常小，见 facebook 论文

2）读写锁实现一致性缓存

public class TestGenericDao {public static void main(String[] args) {// 缓存的优化GenericDao dao = new GenericDaoCached();System.out.println("============> 查询");String sql = "select * from emp where empno = ?";int empno = 7369;Emp emp = dao.queryOne(Emp.class, sql, empno);System.out.println(emp);emp = dao.queryOne(Emp.class, sql, empno);System.out.println(emp);emp = dao.queryOne(Emp.class, sql, empno);System.out.println(emp);System.out.println("============> 更新");dao.update("update emp set sal = ? where empno = ?", 800, empno);emp = dao.queryOne(Emp.class, sql, empno);System.out.println(emp);}
}class GenericDaoCached extends GenericDao {private GenericDao dao = new GenericDao();// 缓存的 Map (SQL语句和查询结果)//  HashMap 作为缓存非线程安全, 需要保护private Map<SqlPair, Object> map = new HashMap<>();private ReentrantReadWriteLock rw = new ReentrantReadWriteLock();@Overridepublic <T> List<T> queryList(Class<T> beanClass, String sql, Object... args) {return dao.queryList(beanClass, sql, args);}@Overridepublic <T> T queryOne(Class<T> beanClass, String sql, Object... args) {// 先从缓存中找，找到直接返回SqlPair key = new SqlPair(sql, args);// 加读锁, 防止其它线程对缓存更改rw.readLock().lock();try {T value = (T) map.get(key);if(value != null) {return value;}} finally {rw.readLock().unlock();}// 加写锁, 防止其它线程对缓存读取和更改rw.writeLock().lock();try {// 多个线程// get 方法上面部分是可能多个线程依次获得锁进来的, 可能已经向缓存填充了数据T value = (T) map.get(key);// 为防止重复查询数据库, 再次验证，双重检查if(value == null) {// 缓存中没有，查询数据库value = dao.queryOne(beanClass, sql, args);map.put(key, value);}return value;} finally {rw.writeLock().unlock();}}@Overridepublic int update(String sql, Object... args) {// 加锁，原子整体，强一致性（并发能力低）// 加写锁, 防止其它线程对缓存读取和更改rw.writeLock().lock();try {// 先更新库int update = dao.update(sql, args);// 清空缓存map.clear();return update;} finally {rw.writeLock().unlock();}}class SqlPair {private String sql;private Object[] args;public SqlPair(String sql, Object[] args) {this.sql = sql;this.args = args;}@Overridepublic boolean equals(Object o) {if (this == o) {return true;}if (o == null || getClass() != o.getClass()) {return false;}SqlPair sqlPair = (SqlPair) o;return Objects.equals(sql, sqlPair.sql) &&Arrays.equals(args, sqlPair.args);}@Overridepublic int hashCode() {int result = Objects.hash(sql);result = 31 * result + Arrays.hashCode(args);return result;}}}

注意

  • 以上实现体现的是读写锁的应用，保证缓存和数据库的一致性，但有下面的问题没有考虑

    • 1）适合读多写少，如果写操作比较频繁，以上实现性能低
    • 2）没有考虑缓存容量
    • 3）没有考虑缓存过期
    • 4）只适合单机
    • 5）并发性还是低，目前只会用一把锁
    • 6）更新方法太过简单粗暴，清空了所有 key（考虑按类型分区或重新设计 key）
  • 乐观锁实现：用 CAS 去更新

3、 读写锁原理

1）图解流程
读写锁用的是同一个 Sycn 同步器，因此等待队列、state 等也是同一个

t1 w.lock，t2 r.lock