尚硅谷JUC并发编程与源码分析2022详细笔记

尚硅谷JUC并发编程2022

1 线程基础知识
- 1.1 并发相关的jar包
- 1.2 start()方法源码分析
- 1.3 多线程基础概念
- 1.4 用户线程和守护线程
2 CompletableFuture
- 2.1 Future接口
- 2.2 FutureTask实现类
3. CompletableFuture类

1 线程基础知识 1.1 并发相关的jar包

java.util.concurrent
java.util.concurrent.atomic
java.util.concurrent.locks

JUC包设计开发者

1.2 start()方法源码分析

public static void main(String[] args){
	Thread t1 = new Thread(() -> {},"t1");
	t1.start();
}

//start方法源码
public synchronized void start() {
    
    if (threadStatus != 0)
        throw new IllegalThreadStateException();

    
    group.add(this);

    boolean started = false;
    try {
    	//重点调用了此处的方法
        start0();
        started = true;
    } finally {
        try {
            if (!started) {
                group.threadStartFailed(this);
            }
        } catch (Throwable ignore) {
            
        }
    }
}

private native void start0();//start0是一个native方法

native调用的是本地方法,我们可以通过下载官网的OpenJDK查看其源码

thread.c

java线程是通过start的方法启动执行的,主要内容在native方法start0中
openjdk的写JNI一般是一一对应的,Thread.java对应的就是Thread.c
start0其实是JVM_StartThread.此时查看源码可以看到在jvm.h中找到声明,jvm.cpp中有实现.

jvm.cpp

thread.cpp

在这里调用了操作系统的线程启动.

1.3 多线程基础概念

一把锁

	synchronized

两个并

并发(concurrent): 是在同一实体上的多个事件,是在一台处理器上"同时"处理多个任务,同一时刻,其实只有一个事件在发生.
并行(parallel): 是在不同实体上的多个事件,是在多台处理器上同时处理多个任务,同一时刻,大家真的都在做事情,你做你的,我做我的,但是我们都在做.

并发VS并行

三个程(进程,线程,管程)

通过上面start线程的案例,其实进程线程都来源与操作系统.

进程: 系统中运行的一个应用程序就是一个进程,每一个进程都有它自己的内存空间和系统资源.
线程: 也被称为轻量级进程,在同一进程内基本会有一个或者多个线程,是大多数操作系统进行调度的基本单元.
管程: Monitor(监视器),也就是我们平时说的锁.Monitor其实是一种同步机制,它的义务是保证(同一时间)只有一个线程可以访问被包含的数据和代码.JVM中同步是基于进入和退出监视器对象(Monitor管程对象)来实现的,每个对象实例都会有一个Monitor对象.Monitor对象会和Java对象一同创建并销毁,它的底层是由C++语言来实现的.

1.4 用户线程和守护线程

一般情况下不做特别说明配置,默认都是用户线程.

用户线程(User Thread): 是系统的工作线程,它会完成这个程序需要完成的业务操作.
守护线程(Daemon Thread): 是一种特殊的线程为其他线程服务的,在后台默默的完成一些系统性的服务,比如垃圾回收线程就是最典型的例子.守护线程作为一个服务线程,没有服务对象就没有必要继续运行了,如果用户线程结束了,意味着程序需要完成的业务操作已经结束了.系统可以退出了.所有假设当系统只剩下守护线程的时候,java虚拟机会自动退出.

判断一个线程是否为守护线程

isDaemon
public final boolean isDaemon()测试这个线程是否是守护线程。 
结果 
true如果这个线程是一个守护线程; false否则。 
另请参见： 
setDaemon(boolean)

public class DaemonDemo{
	public static void main(String[] args){
	    Thread t1 = new Thread(() -> {
	        System.out.println(Thread.currentThread().getName()+"t 开始运行，"+(Thread.currentThread().isDaemon() ? "守护线程":"用户线程"));
	        while (true) {
	
	        }
	    }, "t1");
	    //线程的daemon属性为true表示是守护线程，false表示是用户线程
	    //---------------------------------------------
	    t1.setDaemon(true);
	    //-----------------------------------------------
	    t1.start();
	    //3秒钟后主线程再运行
	    try { TimeUnit.SECONDS.sleep(3); } catch (InterruptedException e) { e.printStackTrace(); }
	
	    System.out.println("----------main线程运行完毕");
	}
}
//未加t1.setDaemon(true);，默认是用户线程，他会继续运行，所以灯亮着
//加了t1.setDaemon(true);是守护线程，当用户线程main方法结束后自动退出了

守护线程作为一个服务线程，没有服务对象就没有必要继续运行了，如果用户线程全部结束了，意味着程序需要完成的业务操作已经结束了，系统可退出了。假如当系统只剩下守护线程的时候，java虚拟机会自动退出。
setDaemon(true)方法必须在start()之前设置，否则报IIIegalThreadStateException异常

2 CompletableFuture 2.1 Future接口

Future接口(FutureTask实现类)定义了操作异步任务执行的一些方法,如获取异步任务的执行结果,取消任务的执行,判断任务是否被取消,判断任务是否执行完毕等.
比如主线程让一个子线程取执行任务,子线程可能比较耗时,启动子线程开始执行任务后,主线程就去做其他的事情了,忙其他事情或者先执行完毕,过了一会才去获取子任务的执行结果或者变更的任务状态.
一句话: Future接口可以为主线程开一个分支任务，专门为主线程处理耗时和费力的复杂业务.

2.2 FutureTask实现类

Future是java5新加的一个接口,它提供了一种异步并行计算的功能.
如果主线程需要执行一个很耗时的计算任务,我们就可以通过Future吧这个任务放到异步线程中执行.
主线程进行处理其他任务或者先行接收,再通过Future获取计算结果.
异步多线程任务执行且返回结果,三个特点: 多线程,有返回,同步任务.(班长为老师取买水作为新启动的异步线程任务且买到水有结果返回 )

FutureTask相关继承关系

在源码中可以看到，FutureTask既继承了RunnableFuture接口，也在构造方法中实现了Callable接口（有返回值、可抛出异常）和Runnable接口

public class CompletableFutureDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        FutureTask futureTask = new FutureTask<>(new MyThread());
        Thread thread = new Thread(futureTask,"t1");
        thread.start();
        System.out.println(futureTask.get());//接收call方法的返回值
    }
}

class MyThread implements Callable{

    @Override
    public String call() throws Exception {
        System.out.println("come in call()...");
        return "hello Callable";
    }
}

Futrue + 线程池异步多线程任务配合,能显著的提高城乡的执行效率

//示例1:3个任务,1个main线程处理,大概耗时1140毫秒
public class FutureThreadPoolDemo {
    public static void main(String[] args) {
        m1();
    }
    public static void m1(){
        //3个任务,只有一个main线程执行,耗时多少?
        long startTime = System.currentTimeMillis();
        //暂停毫秒
        try {
            TimeUnit.MILLISECONDS.sleep(500);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        try {
            TimeUnit.MILLISECONDS.sleep(300);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        try {
            TimeUnit.MILLISECONDS.sleep(300);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        long endTime = System.currentTimeMillis();
        System.out.println("相差时间:"+ (endTime - startTime) + "毫秒");
        System.out.println(Thread.currentThread().getName()+"线程t...end");
    }
}

//3个任务3个线程，利用线程池（假如每次new一个Thread，太浪费资源），大概902毫秒。
public class FutureThreadPoolDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        m2();
    }
    public static void m2() throws ExecutionException, InterruptedException {
        //3个任务，目前开启多个异步任务线程来处理，请问耗时多少？
        ExecutorService threadPool = Executors.newFixedThreadPool(3);
        long startTime = System.currentTimeMillis();
        FutureTask futureTask1 = new FutureTask<>(() ->{
            try {
                TimeUnit.MILLISECONDS.sleep(500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "task1 over";
        });
        threadPool.submit(futureTask1);
        FutureTask futureTask2 = new FutureTask<>(() ->{
            try {
                TimeUnit.MILLISECONDS.sleep(300);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "task2 over";
        });
        threadPool.submit(futureTask2);
        System.out.println(futureTask1.get());
        System.out.println(futureTask2.get());
        try {
            TimeUnit.MILLISECONDS.sleep(300);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        long endTime = System.currentTimeMillis();
        System.out.println("相差时间: "+(endTime - startTime) +" 毫秒");
        System.out.println(Thread.currentThread().getName()+"线程t...end");
        threadPool.shutdown();
    }
}

Future缺点

get()方法阻塞

一旦调用get()方法,不管是否计算完成,都会导致程序阻塞,所以get()方法的位置一般放在程序最后

public class FutureAPIDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException, TimeoutException {
        FutureTask futureTask = new FutureTask<>(() ->{
            System.out.println(Thread.currentThread().getName()+"t...come in");
            TimeUnit.SECONDS.sleep(5);
            return "task over";
        });
        Thread t1 = new Thread(futureTask,"t1");
        t1.start();
        System.out.println(Thread.currentThread().getName()+"/t...忙其他任务了");
        //System.out.println(futureTask.get());
        //System.out.println(futureTask.get(3,TimeUnit.SECONDS));
    }
    
}

isDone()轮询

利用if(futureTask.isDone())的方式使得FutureTask在结束之后才get(),但是也会消耗cpu

public class FutureAPIDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException, TimeoutException {
        FutureTask futureTask = new FutureTask<>(() ->{
            System.out.println(Thread.currentThread().getName()+"t...come in");
            TimeUnit.SECONDS.sleep(5);
            return "task over";
        });
        Thread t1 = new Thread(futureTask,"t1");
        t1.start();
        System.out.println(Thread.currentThread().getName()+"/t...忙其他任务了");
        while(true){
            if(futureTask.isDone()){
                System.out.println(futureTask.get());
                break;
            }else{
                //暂停毫秒
                try { TimeUnit.MILLISECONDS.sleep(500); } catch (InterruptedException e) { e.printStackTrace(); }
                System.out.println("正在处理中，不要再催了，越催越慢 ，再催熄火");
            }
        }
    }
}

Future结论

Future对于结果的获取不是很友好,只能通过阻塞或者轮询的方式的到结果.
对于简单的业务场景使用Future完全可以.
应对Future的完成时间,完成了可以告诉我们,也就是我们的回调通知,通过轮询的方式去判断是否完成这样非常占CPU并且代码也不优雅.
创建异步任务: 使用Future + 线程池配合使用
多个任务前后依赖可以组合处理(水煮鱼): 想将多个异步任务的计算结果组合起来，后一个异步任务的计算结果需要前一个异步任务的值，将两个或多个异步计算合成一个异步计算，这几个异步计算相互独立，同时后面这个又依赖前一个处理的结果(比如买鱼-加料-烹饪)
对计算速度选最快完成的（并返回结果）::当Future集合中某个任务最快结束时，返回结果，返回第一名处理结果

3. CompletableFuture类

get()方法在Future计算完成之前会一直处在阻塞状态下,isDone()方法容易耗费CPU资源.
对于真正的异步处理我们希望是可以通过传入回调函数,在Futrue结束时自动调用该回调函数,这样,我们就不用等待结果.
阻塞的方式和异步编程的涉及理念相违背,而轮询的方式会耗费无谓的CPU资源,因此,JDK8设计出CompletableFuture.
CompletableFuture提供了一种观察者模式类的机制,可以让任务执行完成后通知监听的一方.
在Java 8中， Complet able Future提供了非常强大的Future的扩展功能，可以帮助我们简化异步编程的复杂性，并且提供了函数式编程的能力，可以通过回调的方式处理计算结果，也提供了转换和组合Complet able Future的方法
它可能代表一个明确完成的Future，也有可能代表一个完成阶段(Completion Stage) ，它支持在计算完成以后触发一些函数或执行某些
动作。

public class CompletableFuture implements Future, CompletionStage {}

CompletionStage:

CompletionStage代表异步计算过程中的某一个阶段,一个阶段完成以后可能会触发另外一个阶段.
一个阶段的计算执行可以是一个Function,Consumer或者Runnable,比如:stage.thenApply(x - >square(x)).thenAccept(x -> System.out.print(x)).thenRun(() -> System.out.println())
一个阶段的执行可能是被单个阶段的完成触发,也可能是由多个阶段一起触发.
代表异步计算过程中的某一个阶段,一个阶段完成以后可能会触发另外一个阶段,有些类似Linux系统的管道分隔符传参数.

核心的四个静态方法

利用核心的四个静态方法创建一个异步操作 ,不建议直接new
关键就是有没有返回值,是否用了线程池
参数说明：
没有指定Executor的方法，直接使用默认的ForkJoinPool.commPool()作为它的线程池执行异步代码
如果指定线程池，则使用我们定义的或者特别指定的线程池执行异步代码

runAsync无返回值

public static CompletableFuture runAsync(Runnable runnable) {
     return asyncRunStage(asyncPool, runnable);
}

public static CompletableFuture runAsync(Runnable runnable, Executor executor) {
     return asyncRunStage(screenExecutor(executor), runnable);
}

supplyAsync有返回值

public static  CompletableFuture supplyAsync(Supplier supplier) {
    return asyncSupplyStage(asyncPool, supplier);
}

public static  CompletableFuture supplyAsync(Supplier supplier,Executor executor) {
    return asyncSupplyStage(screenExecutor(executor), supplier);
}

上线方法中Executor executor参数说明:

如果没有指定Executor的方法,直接使用默认的ForkJoinPool.commonPool()作为它的线程池执行异步代码.
如果指定线程池,则使用我们自定义的或者特别指定的线程池执行异步代码.

runAsync

public class CompletableFutureBuildDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture completableFuture= CompletableFuture.runAsync(() -> {
            System.out.println(Thread.currentThread().getName());
            //停顿几秒线程
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        System.out.println(completableFuture.get());
    }
}
//ForkJoinPool.commonPool-worker-9 //默认的线程池
//null --- 没有返回值

runAsync + 线程池

public class CompletableFutureBuildDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ExecutorService threadPool = Executors.newFixedThreadPool(3);
        CompletableFuture completableFuture = CompletableFuture.runAsync(() -> {
            System.out.println(Thread.currentThread().getName());
            //暂停几秒钟线程
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }, threadPool);
        System.out.println(completableFuture.get());
    }
}

supplyAsync+线程池

public class CompletableFutureBuildDemo {

    public static void main(String[] args) throws ExecutionException, InterruptedException {

        ExecutorService executorService = Executors.newFixedThreadPool(3);//加入线程池

        CompletableFuture completableFuture = CompletableFuture.supplyAsync(()->{
            System.out.println(Thread.currentThread().getName());
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "helllo supplyasync";
        });
        System.out.println(completableFuture.get());
    }
    //ForkJoinPool.commonPool-worker-9---------默认的线程池
    //helllo supplyasync-------------supplyasync有返回值了
}

CompletableFuture通用演示

从Java8开始引入了CompletableFuture,它是Future的功能增强版,减少阻塞和轮询,可以传入回调对象,当异步任务完成或者发生异常时,自动调用回调对象的回调方法.

CompletableFuture完成与Future相同的功能

public class CompletableFutureUseDemo {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        future1();
    }
    private static void future1() throws ExecutionException, InterruptedException {
        CompletableFuture completableFuture = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + "...come in");
            int result = ThreadLocalRandom.current().nextInt(10);
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("1秒钟后出结果:" + result);
            return result;
        });
        System.out.println(Thread.currentThread().getName()+"线程先去忙其他任务!");
        System.out.println(completableFuture.get());
    }
}

CompletableFuture通用演示

public class CompletableFutureUseDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        //自定义线程池
        ExecutorService threadPool = Executors.newFixedThreadPool(3);
        try {
            CompletableFuture.supplyAsync(() ->{
                System.out.println(Thread.currentThread().getName()+"...come in");
                int result = ThreadLocalRandom.current().nextInt(10);
                try {
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("1秒钟后出结果:"+result);
                if(result > 5){
                    int i = 10/0; //制造异常
                }
                return result;
            },threadPool).whenComplete((v,e) ->{ //v表示result,e表示异常,CompletableFuture通过whenComplete来减少阻塞和轮询（自动回调）
                if(e == null){//判断有没有异常
                    System.out.println("计算完成,更新系统update value:"+v);
                }
            }).exceptionally(e ->{
                e.printStackTrace();
                System.out.println("异常情况:"+e.getCause()+"t"+e.getMessage());
                return null;
            });
            System.out.println(Thread.currentThread().getName()+"线程先去忙其它任务");
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            threadPool.shutdown();
        }
    }

CompletableFuture优点总结:

异步任务结束时，会自动回调某个对象的方法；
主线程设置好回调后，不再关心异步任务的执行，异步任务之间可以顺序执行
异步任务出错时，会自动回调某个对象的方法

尚硅谷JUC并发编程与源码分析2022详细笔记

Java相关栏目本月热门文章