前端带你学后端系列 ②【高并发JUC之篇一】

Ⅰ前置概念 ①并发与并行 ②线程与进程

①并发与并行

咖啡机就是我们的处理器核
接咖啡的人就是我们的任务

②线程与进程

进程是程序资源分配的最小单位，线程是程序执行的最小单位。

比如打开谷歌浏览器，计算机会开启多个进程，如下图

同理，线程是组成各个进程的最小单位。

Ⅱ JUC并发编程的各个组件

Java.util.concurrent（JUC）是 Java 标准库中的一个包，它提供了一组并发编程工具。

本文将介绍很多组件，java8 以下的，不推荐使用，已经不支持了。

① 什么是线程池？

① 线程池的组成

② 线程池的原理

③ 简单的例子demo

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ThreadPoolExample {
    public static void main(String[] args) {
        //1. 创建一个容量为 20 的线程池
        ExecutorService executorService = Executors.newFixedThreadPool(20);
       
       // 3. 提交任务，由线程池中空闲的线程执行
       // 模拟有100个任务
	 	for (int i = 0; i < 100; i++) {
       	    executorService.execute(new Task(i)); 
     	}
     
        // 3. 调用任务完成，关闭线程池
        executorService.shutdown(); 
    }


// 2. 继承【Runnable接口】创建任务
static class Task implements Runnable {
        private int taskNum;

        public Task(int num) {
            this.taskNum = num;
        }

		// 线程池中的线程会调用该方法进行具体任务的执行
        @Override
        public void run() { 
            System.out.println("正在执行task " + taskNum);
            try {
            	// 通过sleep模拟耗时任务
                Thread.sleep(2000); 
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("task " + taskNum + " 执行完毕");
        }
    }
}

简单的例子demo2，使用lambda表达式

import java.util.concurrent.*;

/*
	主要代码：
		1. 创建线程池，指定【核心线程数量】【最大线程数量】等基本参数
		2. 调用execute()方法，使用lambda表达式创建任务
		3. 这个时候，线程池的线程会自动调用你传进去的方法。
*/

public class ThreadPoolExecutorDemo {
    public static void main(String[] args) {
        // 1. 创建一个线程池对象
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                2, 5, 10, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(5), Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy());

        // 2. 使用lambda表达式提交多个任务到线程池中
        for (int i = 1; i <= 100; i++) {
            executor.execute(() -> {
                try {
                    Thread.sleep(2000);
                    System.out.println(Thread.currentThread().getName() + " is running");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }

        // 3. 关闭线程池
        executor.shutdown();
    }
}

// ThreadPoolExecutor 七个参数的解释：
1. corePoolSize 核心线程数量
2. maximumPoolSize 最大线程数
3. keepAliveTime 最大空闲时间(非核心线程在空闲状态下保持存活的最长时间)
4. unit 时间单位
5. BlockingQueue workQueue 任务队列(队列，缓存超出的线程数)
6. ThreadFactory threadFactory 线程工厂(创建线程用)
7. RejectedExecutionHandler handler 饱和处理机制(线程数到达核心线程数量，并且任务队列满了，这个时候的处理机制)

阿里推荐使用的创建线程池的方式

② ExecutorService线程池(Java5出现)–高并发

① 线程类的继承

② ThreadPoolExecutor 详解

// 构造方法
ThreadPoolExecutor(
                  int corePoolSize,  //线程池最小线程数，必须大于0
                  int maximumPoolSize,  //线程池最大线程数，必须大于0和corePoolSize
                  long keepAliveTime, TimeUnit unit,  //设置线程池中线程最大空闲时间，必须大于0
                  BlockingQueue<Runnable> workQueue, //任务队列，只有是excute提交的任务，才会进去
                  ThreadFactory threadFactory（非必传，默认是当前线程）, //创建一个新线程时使用的工厂。默认defaultThreadFactory
                  RejectedExecutionHandler handler（非必传，默认是AbortPolicy）//拒绝策略。当任务队列已经满时的策略
)

// 拒绝策略
static class     ThreadPoolExecutor.AbortPolicy //抛出一个 RejectedExecutionException
static class     ThreadPoolExecutor.CallerRunsPolicy //把任务交给父线程运行
static class     ThreadPoolExecutor.DiscardOldestPolicy //丢弃最旧的未处理请求，接受最新的任务
static class     ThreadPoolExecutor.DiscardPolicy //不接受，且没有任何日志

//获取与设置线程池固定信息

//最小线程数
void setCorePoolSize(int corePoolSize)
int  getCorePoolSize()

//最大线程数
void setMaximumPoolSize(int maximumPoolSize)
int  getMaximumPoolSize()

//线程池空闲时间
void setKeepAliveTime(long time, TimeUnit unit)
long getKeepAliveTime(TimeUnit unit)

//拒绝策略
void    setRejectedExecutionHandler(RejectedExecutionHandler handler)
RejectedExecutionHandler    getRejectedExecutionHandler()

//线程工厂
void    setThreadFactory(ThreadFactory threadFactory)
ThreadFactory    getThreadFactory()

// 获取线程池动态信息
int    getActiveCount() //获取活动线程数
int    getPoolSize() //返回池中当前的线程数
long   getCompletedTaskCount() //获取已经完成的任务的大致总数
int    getLargestPoolSize() //是一个动态变量,是记录线程池曾经达到的最高值
long   getTaskCount() //返回计划执行的任务的大概总数
BlockingQueue<Runnable>   getQueue() //返回未执行的任务队列




// 获取线程池动态信息
int    getActiveCount() //获取活动线程数
int    getPoolSize() //返回池中当前的线程数
long   getCompletedTaskCount() //获取已经完成的任务的大致总数
int    getLargestPoolSize() //是一个动态变量,是记录线程池曾经达到的最高值
long   getTaskCount() //返回计划执行的任务的大概总数
BlockingQueue<Runnable>   getQueue() //返回未执行的任务队列




// 判断线程池的状态
//当调用shutdown()或shutdownNow()方法后返回为true,不能作为判断线程池是否关闭的依据
boolean    isShutdown()

//isTerminated当调用shutdown()方法后，并且所有提交的任务完成后返回为true;
//isTerminated当调用shutdownNow()方法后，成功停止后返回为true;
boolean    isTerminated()

//执行 shutdown 或 shutdownnow 之后,如果正在终止但尚未完成，则返回 true
boolean    isTerminating()

//awaitTermination不会关闭ExecutorService，只是定时检测一下他是否关闭，若关闭则返回true，否则返回false。
//一般情况下会和shutdown方法组合使用。可以作为判断线程池是否关闭的依据
boolean awaitTermination(long timeout, TimeUnit unit)





// 设置任务开始与结束时的工作
protected void  afterExecute(Runnable r, Throwable t) //完成指定Runnable的执行后调用方法。
protected void  beforeExecute(Thread t, Runnable r) //在给定的线程中执行给定的Runnable之前调用方法




// 开启核心线程
int     prestartAllCoreThreads() //启动所有核心线程,返回开启的线程数(返回值不一定等于核心线程数)
boolean prestartCoreThread() //开启一个线程，如果线程池中线程数大于核心线程数，那么方法无效，返回false





// 开启核心线程的回收
protected void  afterExecute(Runnable r, Throwable t) //完成指定Runnable的执行后调用方法。
protected void  beforeExecute(Thread t, Runnable r) //在给定的线程中执行给定的Runnable之前调用方法

③ 线程池工厂类Executors创建线程(了解就行，不让用)

可以创建
① 可重用固定线程数
② 单线程的线程池，单一的工作线程可以保证提交任务的顺序执行
③ 可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程
④ 延迟定时执行任务的线程池

③ Fork/Join框架(java7) --高并发

① 概念

ForkJoinPool是Java中提供了一个线程池，特点是用来执行分治任务。主题思想是将大任务分解为小任务，然后继续将小任务分解，直至能够直接解决为止，然后再依次将任务的结果合并。

② 代码demo

package com.javakk;

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;
import java.util.stream.LongStream;

/**
 * ForkJoinPool求和
 * @author 老K
 */
public class ForkJoinPoolTest {

    private static ForkJoinPool forkJoinPool;

    /**
     * 求和任务类继承RecursiveTask
     * ForkJoinTask一共有3个实现：
     * RecursiveTask：有返回值
     * RecursiveAction：无返回值
     * CountedCompleter：无返回值任务，完成任务后可以触发回调
     */
    private static class SumTask extends RecursiveTask<Long> {
        private long[] numbers;
        private int from;
        private int to;

        public SumTask(long[] numbers, int from, int to) {
            this.numbers = numbers;
            this.from = from;
            this.to = to;
        }

        /**
         * ForkJoin执行任务的核心方法
         * @return
         */
        @Override
        protected Long compute() {
            if (to - from < 10) { // 设置拆分的最细粒度，即阈值，如果满足条件就不再拆分，执行计算任务
                long total = 0;
                for (int i = from; i <= to; i++) {
                    total += numbers[i];
                }
                return total;
            } else { // 否则继续拆分，递归调用
                int middle = (from + to) / 2;
                SumTask taskLeft = new SumTask(numbers, from, middle);
                SumTask taskRight = new SumTask(numbers, middle + 1, to);
                taskLeft.fork();
                taskRight.fork();
                return taskLeft.join() + taskRight.join();
            }
        }
    }

    public static void main(String[] args) {
        // 也可以jdk8提供的通用线程池ForkJoinPool.commonPool
        // 可以在构造函数内指定线程数
        forkJoinPool = new ForkJoinPool();
        long[] numbers = LongStream.rangeClosed(1, 100000000).toArray();
        // 这里可以调用submit方法返回的future，通过future.get获取结果
        Long result = forkJoinPool.invoke(new SumTask(numbers, 0, numbers.length - 1));
        forkJoinPool.shutdown();
        System.out.println("最终结果："+result);
        System.out.println("活跃线程数："+forkJoinPool.getActiveThreadCount());
        System.out.println("窃取任务数："+forkJoinPool.getStealCount());
    }
}

③ 与ThreadPoolExecutor原生线程池的区别

④ Future(java1.5) 与 CompletableFuture(java8) --高并发

① Future的概念

Future 最主要的作用是，比如当做一定运算的时候，运算过程可能比较耗时，有时会去查数据库，或是繁重的计算，比如压缩、加密等，在这种情况下，如果我们一直在原地等待方法返回，显然是不明智的，整体程序的运行效率会大大降低。我们可以把运算的过程放到子线程去执行，再通过 Future 去控制子线程执行的计算过程，最后获取到计算结果。这样一来就可以把整个程序的运行效率提高，是一种异步的思想。

常见Future的使用场景：发送api请求

ExecutorService threadPool = Executors.newFixedThreadPool(5);
// 泛型指定返回的结果类型
Future<String> future = threadPool.submit(new Callable<String>() {
    @Override
    public String call() throws Exception {
         System.out.println("模拟执行复杂任务，睡3秒");
         Thread.sleep(3000);
         System.out.println("模拟复杂结束任务");
        return "ok";
    }
});

// 阻塞等待任务完成
String result = future.get(); 
System.out.println("获取任务结果为: "+result + "  "+Thread.currentThread().getName());
                

② Future详细

public interface Future<V> {
 
 	// 取消任务的执行
    boolean cancel(boolean mayInterruptIfRunning);
    
    // 判断任务是否取消
    boolean isCancelled();
 
    // 判断是否执行完毕
    boolean isDone();
 
    // 任务已经执行完毕了，获取到任务执行的结果
    V get() throws InterruptedException, ExecutionException;
 
    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutExceptio
}

③ CompletableFuture(支持异步回调的功能)

@Test
public void test1() throws ExecutionException, InterruptedException {
   // 1. 创建CompletableFuture 	
   CompletableFuture<Object> future = new CompletableFuture<>();
   // 2. 模拟异步任务
   new Thread(() -> {
       try {
           Thread.sleep(5000);
       } catch (InterruptedException e) {
           throw new RuntimeException(e);
       }
       int i = new Random().nextInt(7);
       System.out.println("随机数" + i);
       if ( i < 3){
           future.complete(3);
       } else {
           future.completeExceptionally(new RuntimeException("太大了"));
       }
   }).start();

   // 3. 添加结束回调器
   future.whenComplete((r,e) -> {
       if (e != null){
           System.out.println("结果： " + r+ "  ex:" + e.getMessage());
       }else {
           System.out.println("结果： " + r);
       }
   });
}

⑤ CountDownLatch(计数器)–高并发

CountDownLatch是通过一个计数器来实现的，计数器的初始值是线程的数量。每当一个线程执行完毕后，计数器的值就减1，当计数器的值为0时，表示所有线程都执行完毕，然后在闭锁上（调用await方法的线程）等待的线程就可以恢复工作了。

例子1

例子2

/**
 * @author: 公众号：java金融
 */
public class TestCountDownLatch1 {
     public static void main(String[] args) throws InterruptedException {
          int count = 3;
          CountDownLatch countDownLatch = new CountDownLatch(count);
          for (int i = 0; i < count; i++) {
               final int index = i;
               new Thread(() -> {
                    try {
                         Thread.sleep(1000 + ThreadLocalRandom.current().nextInt(1000));               
                    } catch (InterruptedException e) {
                         e.printStackTrace();
                    }finally{
                        countDownLatch.countDown();
                    }
               }).start();
          }
          countDownLatch.await();// 主线程在阻塞，当计数器==0，就唤醒主线程往下执行。
          System.out.println("主线程:在所有任务运行完成后，进行结果汇总");
     }
}

⑥ Semaphore(信号量)–高并发

① 概念

Semaphore 通常我们叫它信号量， 可以用来控制同时访问特定资源的线程数量，通过协调各个线程，以保证合理的使用资源。

比如：停车场入口立着的那个显示屏，每有一辆车进入停车场显示屏就会显示剩余车位减1，每有一辆车从停车场出去，显示屏上显示的剩余车辆就会加1，当显示屏上的剩余车位为0时，停车场入口的栏杆就不会再打开，车辆就无法进入停车场了，直到有一辆车从停车场出去为止。

比如：在学生时代都去餐厅打过饭，假如有3个窗口可以打饭，同一时刻也只能有3名同学打饭。第四个人来了之后就必须在外面等着，只要有打饭的同学好了，就可以去相应的窗口了 。

② 使用场景

常用于限流

比如：数据库连接池，同时进行连接的线程有数量限制，连接不能超过一定的数量，当连接达到了限制数量后，后面的线程只能排队等前面的线程释放了数据库连接才能获得数据库连接。

比如：停车场场景，车位数量有限，同时只能容纳多少台车，车位满了之后只有等里面的车离开停车场外面的车才可以进入。

③ 常用api与使用demo

常用api

acquire()  
获取一个令牌，在获取到令牌、或者被其他线程调用中断之前线程一直处于阻塞状态。
?
acquire(int permits)  
获取一个令牌，在获取到令牌、或者被其他线程调用中断、或超时之前线程一直处于阻塞状态。
    
acquireUninterruptibly() 
获取一个令牌，在获取到令牌之前线程一直处于阻塞状态（忽略中断）。
    
tryAcquire()
尝试获得令牌，返回获取令牌成功或失败，不阻塞线程。
?
tryAcquire(long timeout, TimeUnit unit)
尝试获得令牌，在超时时间内循环尝试获取，直到尝试获取成功或超时返回，不阻塞线程。
?
release()
释放一个令牌，唤醒一个获取令牌不成功的阻塞线程。
?
hasQueuedThreads()
等待队列里是否还存在等待线程。
?
getQueueLength()
获取等待队列里阻塞的线程数。
?
drainPermits()
清空令牌把可用令牌数置为0，返回清空令牌的数量。
?
availablePermits()
返回可用的令牌数量。

使用demo

用semaphore 实现停车场提示牌功能。

public class TestCar {
?
    //停车场同时容纳的车辆10
    private  static  Semaphore semaphore=new Semaphore(10);
?
    public static void main(String[] args) {
?
        //模拟100辆车进入停车场
        for(int i=0;i<100;i++){
?
            Thread thread=new Thread(new Runnable() {
                public void run() {
                    try {
                        System.out.println("===="+Thread.currentThread().getName()+"来到停车场");
                        if(semaphore.availablePermits()==0){
                            System.out.println("车位不足，请耐心等待");
                        }
                        semaphore.acquire();//获取令牌尝试进入停车场
                        System.out.println(Thread.currentThread().getName()+"成功进入停车场");
                        Thread.sleep(new Random().nextInt(10000));//模拟车辆在停车场停留的时间
                        System.out.println(Thread.currentThread().getName()+"驶出停车场");
                        semaphore.release();//释放令牌，腾出停车场车位
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            },i+"号车");
?
            thread.start();
?
        }
?
    }
}
?