并发编程JUC分析

概念简介

源码

java线程是通过start方法启动执行的,主要native方法start0中

openjdk写的JNI一般是一一对应的,Thread.java对应的就是Thread.c

start0其实就是JVM_StartThread.此时查看源码可以看到JVM.h中找到声明,jvm.cpp中有实现

底层实际是一个操作系统线程

并发与并行

并发Concurrent:是在同一台实体上的事件,是一台处理器上同时处理多个任务,在统一时刻,其实只有一个事情在发生

并行Parallel 是在不同实体上的多个事件,是在多台处理器上同时处理多个任务 同一时刻,大家都各自做各自的事情

进程线程管程

进程:简单的说,在系统允许的一个应用程序就是一个进程,每一个进程都有它自己的内存空间和系统资源.(操作系统级的概念)

线程: 也被称为轻量级进程,在同一个进程内会有1个或多个线程,是大多数操作系统进行时序调度的基本单元

管程:monitor(监视器) 也就是我们平时说的锁

执行线程就要求先成功持有管程才能执行方法,最后当方法完成(无论是正常完成还是非正常完成)时释放管理.在方法执行期间,执行线程持有了管程,其他任何线程都无法在获取到同一个管程

守护线程理论

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

守护线程(Daemon Thread)是系统的工作线程,它会完成这个程序需要完成的工作

守护线程作为一个服务的线程,没有服务对象就没有必要继续运行了,如果用户线程全部结束了,意味着程序需要完成的操作业务以及结束了,系统可以退出了.所以加入当下只剩下守护线程的时候,java虚拟机会自动退出

Thread中有一个isDaemon方法可以判断线程是守护线程还是用户线程,true为守护线程

如果主动调用setDaemon方法将线程设置为true,那么用户线程结束后守护线程也会结束,setDaemon(true),必须在start()方法之前设置,否则报IIIegalThreadStateException异常

 public static void main(String[] args) throws InterruptedException {
        System.out.println(Thread.currentThread().isDaemon());
    }//输出false

Future

future

future接口(futureTask就是该接口的实现类)定义了操作异步任务执行的一些方法.如果获取异步任务的执行结果,取消任务的执行,判断任务是否被取消,判断任务执行是否完毕等

future在java1.5就出现了

比如主线程让子线程去执行任务,子线程可能比较耗时,启动子线程后开始执行任务后,主线程就去做其他事情了,忙其它的事情或者先执行完,过了一会才去获取子线程任务的执行结果或变更任务的状态

callable

Runnable与callable的区别

Runnable是在jdk1.0的时候提出来的多线程的实现接口,而Callable是在jdk1.5之后提出来的

java.lang.Runnable接口之中只提供有一个run()方法,并且没有返回值

java.util.concyrrent.Callable接口中提供的call()方法可以有返回值

futureTask

futureTask实现了RunnableFuture接口,RunnableFuture实现了Runnable和Future接口

public class Test1 {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        FutureTask<String> futureTask = new FutureTask<>(new MyThread());

        Thread thread = new Thread(futureTask, "t1");
        thread.start();
        System.out.println("111");
        System.out.println(futureTask.isDone());
        System.out.println(futureTask.get());//会等待线程出现返回值后继续执行
        System.out.println("111");
    }
}
class MyThread implements Callable<String>{

    @Override
    public String call() throws Exception {
        Thread.sleep(5000);
        System.out.println("come");
        return "res";
    }
}

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

缺点

get()会阻塞,等到结果出现才会继续执行,容易造成堵塞,可以传入参数,设置超时时间,如果超时就抛异常

isDone可以获取线程是否执行完毕,但如果不停轮询这个方法,也会导致性能下降

future对于结果的获取并不友好,只能通过阻塞或轮询的方式得到任务结果

对于简单的业务场景future完全ok,但对于一些复杂的任务如

• 回调通知:应对future的完成时间,完成了可以告诉主线程,而且通过轮询的方式去判断也不优雅
• 创建异步任务,future+线程池配合
• 多个任务前后依赖组合处理,比如想将多个异步任务计算结果组合起来,后一个异步任务需要前一个异步任务的值.将俩个或多个异步任务计算,这几个异步任务计算相互独立,同时后面又依赖对前一个结果的处理
• 对计算速度选最快:当future集合中某个任务最快结束时,返回第一名的处理结果

CompletableFuture

简介

completableFuture实现了future和completionStage接口

completionStage代表异步计算过程中的某一个阶段,一个阶段完成后可能会触发另一个阶段

一个阶段的计算执行可以是一个function,consumer或者runnable

一个阶段的执行可能是呗单个阶段的完成触发,也可能是由多个阶段一起触发

代表异步计算中的某一个阶段,一个阶段完成后可能会触发另一个阶段,有些类似linux系统的管道分割符传参数

completableFuture

在java8中引入,completableFuture提供了非常强大的future的扩展功能,可以帮助我们简化异步编程的复杂性,并且提供了函数式编程的能力,可以通过回调的方式处理计算结果,也提供了转换和组合CompletableFuture方法

它可能代表一个明确完成的future,也有可能代表一个完成阶段(CompletionStage),它仅支持在计算完成后出发一些函数或执行某些动作.

代码

无返回值

package test;

import java.util.concurrent.*;

/**
 * @Author: suiyi
 * @Date: 2023/10/2 16:01
 */
public class Test1 {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        //可以指定CompletableFuture接口的线程池
        ExecutorService executorService = Executors.newFixedThreadPool(3);

        //创建有返回值的异步任务
        CompletableFuture<Void> voidCompletableFuture = CompletableFuture.runAsync(() -> {
            System.out.println(Thread.currentThread().getName());
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },executorService);
        System.out.println(voidCompletableFuture.get()); //为null

        //创建没有返回值的异步任务
        CompletableFuture<String> objectCompletableFuture = CompletableFuture.supplyAsync(()->{
            System.out.println(Thread.currentThread().getName());
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "hello world!";
        });
        System.out.println(objectCompletableFuture.get());//成功获取返回值
        executorService.shutdown();
    }
}

按照以上的方式也是同步的操作,以下是异步操作

public class Test1 {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        //可以指定CompletableFuture接口的线程池
        CompletableFuture.supplyAsync(() -> {
            int res = ThreadLocalRandom.current().nextInt(10);
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return res;
        }).whenComplete((v,e)->{
            //当第一阶段完成后会执行这部分代码,v为上一步的值,e为异常
            if (e==null){
                System.out.println("异步计算完成!"+v);
            }
        }).exceptionally(e->{
            e.printStackTrace();
            System.out.println("出现异常");
            return null;
        });
        //如果不加这行,就不会执行到计算完成那一步,因为主线程结束了,守护线程也会一起结束
        //可以使用线程池规避这个情况
        
        TimeUnit.SECONDS.sleep(5);
    }
}

优点

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

completableFuture的get和join相似,一个在编译期间会抛出异常

常用方法

对结果和触发计算

获取结果

public T get();
public get(long timeout,TimeUnit unit);
public T join();
public T getNow(T valuelfAbsent);//如果计算出结果了返回计算结果,如果没有出结果返回传入的值

主动触发计算

public boolean complete(T value)

可以根据返回值判断获取的是否是默认值,可以配合join获取或者判断

对计算结果进行处理

thenApply

public static void main(String[] args) {
        CompletableFuture<Integer> integerCompletableFuture = CompletableFuture.supplyAsync(() -> {
            return 1;
        }).thenApply(f -> {
            return f + 3;
        }).thenApply(f -> {
            return f + 4;
        });

        System.out.println(integerCompletableFuture.join());
    }

计算结果存在依赖关系,这俩个线程串行画,存在依赖关系,当前步错,不走下一步.

handle

这个方法允许异常,根据带着的参数可以进行进一步的处理

public static void main(String[] args) {
        CompletableFuture<Integer> integerCompletableFuture = CompletableFuture.supplyAsync(() -> {
            return 1;
        }).handle((f,e) -> {
            return f / 0 ;
        }).handle((f,e) -> {
            if (e==null){
                return f + 3;
            }else{
                return -999;
            }
        }).exceptionally(e->{
            System.out.println(e.getMessage());
            return null;
        });

        System.out.println(integerCompletableFuture.join());//输出999
    }

对结果进行消费

thenAccept

public static void main(String[] args) {
        CompletableFuture<Void> integerCompletableFuture = CompletableFuture.supplyAsync(() -> {
            return 1;
        }).thenApply(f->{
            return f+4;
        }).thenAccept(r->{
            System.out.println(r);//这里直接对结果消费,所以返回为void
        });
    }

也可以这样写

 public static void main(String[] args) {
        CompletableFuture<Void> integerCompletableFuture = CompletableFuture.supplyAsync(() -> {
            return 1;
        }).thenApply(f->{
            return f+4;
        }).thenAccept(System.out::println);
    }

• thenRun(Runnable runnable):任务A执行完执行任务b,B不需要A的结果
• thenAccept(Consumer action):任务A执行完成B,B需要A的结果,但是任务B无返回值
• thenApply(Function fn)任务A执行B,B需要A的结果,同时任务B有返回值

结合线程池

没有传入自定义线程池,都用默认线程池forkJoinPool(一个守护线程)

执行第一个任务的时候传入了一个自定义线程池

调用thenRun方法执行第二个任务时,则第二个任务和第一个任务共用同一个线程池

调用thenRunAsync执行第二个任务时,啧第一个任务使用的是自己传入的线程池,第二个任务是forkJoin线程池

备注

可能处理太快,系统优化切换原则,直接使用main线程

对计算速度进行选用

applyToEither

  public static void main(String[] args) {
        CompletableFuture<String> playA = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "A";
        });
        CompletableFuture<String> playB = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "B";
        });
        CompletableFuture<String> res = playA.applyToEither(playB, f -> {
            return f + "winer";
        });
        System.out.println(res.join());//输出Awiner
    }

结果合并

thenCombline

public static void main(String[] args) {
        CompletableFuture<Integer> integerCompletableFuture = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + "go!");
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return 10;
        });
        CompletableFuture<Integer> integerCompletableFuture1 = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + "go!");
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return 20;
        });
        CompletableFuture<Integer> integerCompletableFuture2 = integerCompletableFuture.thenCombine(integerCompletableFuture1, (x, y) -> {
            System.out.println();
            return x + y;
        });
        System.out.println(integerCompletableFuture2.join());
    }

多个线程的情况下可以这样写

 public static void main(String[] args) {
        CompletableFuture<Integer> integerCompletableFuture = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + "go!");
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return 10;
        });
        CompletableFuture<Integer> integerCompletableFuture1 = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + "go!");
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return 20;
        });
        CompletableFuture<Void> voidCompletableFuture = CompletableFuture.allOf(integerCompletableFuture, integerCompletableFuture1);
        int a = integerCompletableFuture.join().intValue();
        int b = integerCompletableFuture1.join().intValue();
        System.out.println(a+b);

    }

线程中断机制

简介

首先一个线程不应该由其他线程来强制中断或者停止,而是应该由线程自己停止,自己来决定自己的命运.所以Thread.stop,Thread.suspend,Thread.resume已经被废弃了

其次java中没有办法立即停止一个线程,然而停止线程却显得游为重要,比如取消一个耗时操作

因此java提供了一种用于停止线程的协商机制--中断,也即中断标识协商机制,也即中断标识协商机制

中断只是一种协商机制,java没有给中断增加任何语法,中断的过程完全需要程序员自己实现

若要中断一个线程,你需要手动调用该线程的inerrupt方法,该方法也仅仅是讲线程对象中断标识设置成true

按照你需要自己写代码不断地检测当前线程的标识位,如果为true,标识别的线程请求这条线程中断

此时究竟该做什么需要你自己实现

每个线程对象都有一个中断标识位,用于标识线程是否被中断,该标识位为true表示中断,为false表示为中断

通过调用线程对象interrupt方法,讲该线程的标识位设置为true;可以砸你别的线程中调用,也可以在自己线程中调用

• public void interrupt() 静态方法,实例方法中断设置为true,发起一个协商,而不会立刻停止线程

• public static boolean interrupt()：静态方法Thread.interrupted()判断线程是否被中断清除当前中断状态.

  • 这个方法做了俩件事
  • 放回当前线程的中断状态，测试当前线程是否已被中断
  • 将当前线程的中断状态清零并重新设为false，清除线程中断状态
• public boolean isInterrupted 判断当前线程是否被中断

代码

普通实现

public class Test2 {
    static volatile boolean isStop=false;
    public static void main(String[] args) throws InterruptedException {
        new Thread(()->{
            while (true){
                if (isStop){
                    System.out.println(Thread.currentThread().getName());
                    break;
                }
                System.out.println("hello!");

            }
        }).start();
        TimeUnit.SECONDS.sleep(1);
        isStop=true;
    }

}

unterrput实现

public class Test1 {
    public  static int a=0;
    public static void main(String[] args) throws InterruptedException {
        Thread t1 = new Thread(() -> {
            while (true) {
                if (Thread.currentThread().isInterrupted()) {
                    System.out.println(Thread.currentThread().getName() + "被修改为true,程序停止");
                    break;
                }
                System.out.println("gogogo");
            }
        });
        t1.start();
        TimeUnit.SECONDS.sleep(3);
        t1.interrupt();
    }
}

具体来说当一个线程调用interrupt时

如果线程处于正常活动状态,那么该线程的中断标志位设置为true,仅此而已(不活动的线程不受影响)

被设置中断标志的线程讲继续正常运行,不受影响

所以interrupt并不能中断线程,需要被调用的线程自己配合才行

如果线程处于被阻塞状态,在别的线程中调用对象的interrupt方法,那么线程讲立即退出被阻塞状态,并抛出一个InterruptedException异常

如果线程调用wait或join,sleep这个了哦的中断方法将会被清除,并受到InterruptedException

如果sleep中捕获了异常,将会导致无法正常的sleep

静态方法

public static boolean interrupted()

判断线程是否被中断并清除当前中断状态

这个方法做了俩件事情

1. 返回当前线程的中断状态,测试是否已经被中断
2. 将当前线程的中断状态清零并重新设置为false,清除线程的中断状态

LockSupport

locksupport是用来创建锁和其他同步类的基本线程阻塞原语

线程等待与唤醒

1. 使用Object中的wait()方法让线程等待,用Object中的notify()方法唤醒线程
2. JUC中的Condition.await方法让线程等待,使用signal方法唤醒线程
3. locksupport类可以阻塞当前下次你以及唤醒被阻塞的线程

wait和notify必须放在同步块或者方法里,且成堆出现对使用先wait后notify才ok

condition的俩个方法同理

    public static void main(String[] args){

        ReentrantLock lock = new ReentrantLock();
        Condition condition = lock.newCondition();
        new Thread(()->{
            lock.lock();
            try {
                condition.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }finally {
                lock.unlock();
            }
        });
        lock.lock();
        try {
            TimeUnit.SECONDS.sleep(3);
            condition.signal();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            lock.unlock();
        }

    }

Condition中的线程等待和唤醒方法,需要先获取锁,一定要先await和signal,不能反了

LockSupport的等待和唤醒

lockSupport类使用了一种名为permit(许可)的概念来做到阻塞和唤醒线程的功能,每个线程都有一个许可(permit)

 public static void park() {
        UNSAFE.park(false, 0L);
    }

默认设置的是永远不放行,所以一开始调用park()方法线程就会阻塞,直到别的线程给当前线程发放permit,park方法才会被唤醒

public static void main(String[] args) throws InterruptedException {
        Thread thread = new Thread(() -> {
            System.out.println("gogogo");
            LockSupport.park();
            System.out.println("gogogo");
        });
        thread.start();
        TimeUnit.SECONDS.sleep(3);
        Thread thread1 = new Thread(() -> {
            System.out.println("唤醒");
            LockSupport.unpark(thread);
        });
        thread1.start();
    }

b线程唤醒a线程

没有加锁的需求,之前错误的先唤醒后等待,lockSupport一样支持

public class Test1 {
    public static void main(String[] args) throws InterruptedException {
        Thread thread = new Thread(() -> {

            System.out.println("gogogo0");
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("gogogo1");
            LockSupport.park();//这里会直接继续，已经park过了
//            LockSupport.park();//连续俩次park就不行了
            System.out.println("gogogo2");
        });
        thread.start();
        Thread thread1 = new Thread(() -> {
            System.out.println("唤醒");
            LockSupport.unpark(thread);
        });
        thread1.start();
    }
}

许可证的累加上线是1,所以第二次park就无效了

JMM

java内存模型

cpu和物理内存的速度是不一致的

cou的运行并不是直接操作内存而是先把内UC你里的数据读到缓存,而内存的读和写操作的时候就会导致不一致的问题

jvm规范视图定义一种java内存模型(java memory model)简称jmm来屏蔽掉各种硬件和操作系统内存访问差异

以实现让java程序在各个平台下都能达到一致的内存访问效果

定义

JMM本身是一种抽象的概念,并不真实存在,只是描述一组约定的规范,这组规范定义了程序中,各个变量的读写访问方式并决定一个线程对于共享变量的写入合适以及如何变成对另一个线程可见,关键技术都是围绕多线程的,原子性,可见性,和有序性展开的

通过jmm来实现线程和主内存之间的抽象关系

屏蔽各个硬件平台和操作系统的内存访问差异,让java程序在各种平台下能达到一致的内存访问效果

三大特性

可见性,原子性,有序性,这也是并发编程的三大问题

可见性

是指当一个线程修改了某一个共享变量的值,其他线程是否能立即知道变更

系统主内存共享变量数据被写入的时机是不确定的,多线程并发下,可能出现脏读,每个线程都有自己的工作内存,线程自己的工作内存中保存了该线使用到的变量主内存副内存拷贝,线程对变量的所有操作(读取,赋值等)都必须在线程自己的工作内存中进行,而不能直接读写主内存的变量,不同线程之间也无法访问对方工作内存中的变量,线程间变量值的传递均需要通过主内存来完成

有序性

指令会根据 as-if-serial和happens-before规则对代码进行重排序

原子性

在一次或多次=操作中,要么所有的操作都执行,并且不会受其他因素干扰而中断,要么所有操作都不执行

内存访问流程

线程之间共享变量存储在主内存中(内存条)

每个线程都有一个本地的工作内存,本地工作内存中存储了该线程用来读写共享变量的副本(从硬件角度来讲就是cpu的L1,L2,L3)缓存

volatile

volatile能保证程序的可见性和有序性

当写一个volatile变量的时候,JMM会把该线程对应的本地内存中的共享变量值立即刷新回主内存中

当读一个volatile变量时,JMM会把该线程对应的本地内存设置为无效,重新回到主内存中读取最新共享变量

所以volatile的写内存语义是直接刷新到主内存中,读的内存语义是直接从主内存中获取

内存屏障

内存屏障也称内存栅栏,屏障指令等,是一类同步屏障指令,是cpu活编译器在堆内存随机访问的操作中的一个同步点,使得此点之前的所有读写操作都执行后才开始执行此点之后的操作.避免代码重排序.内存屏障其实就是一种jvm指令,java内存模型的重排序规则会要求java编译器在生成jvm指令时插入特定的内存屏障指令,通过这些内存屏障指令,volatile实现了java内存模型中的可见性和有序性,但volatile无法保证原子性

内存屏障之前所有写操作都要回写到主内存

内存屏障之后的所有读操作都能获得内存屏障之前所有写操作的最新结果

因此重排序时,不允许把内存屏障之后的指令重排序的到内存屏障之前.

对volatile变量的写,先发生于任意后续对这个vlatile的读也叫写后读

• 读屏障,在读指令之前插入读屏障,让工作内存或cpu告诉缓存中的缓存数据失效,重新回到主内存中获取最新数据
• 写屏障:在写指令后插入写屏障,强制把缓冲区的数据刷回到主内存中去
  下面是基于保守策略的JMM内存屏障插入策略：

在每个volatile写操作的前面插入一个StoreStore屏障。

在每个volatile写操作的后面插入一个StoreLoad屏障。

在每个volatile读操作的前面插入一个LoadLoad屏障。

在每个volatile读操作的后面插入一个LoadStore屏障。
其实分为四个屏障

storeload它的开销是四种屏障中最大的（冲刷写缓冲器，清空无效化队列）。
在大多数处理器的实现中，这个屏障是个万能屏障，兼具其他三种内存屏障的功能。

• Load：将主内存中的数据拷贝到处理器的缓存中
• Store：将处理器缓存的数据刷新到主内存中

规则

第一个操作为volatile读时,不论第二个操作是什么,都不能重排序.这个操作保证,volatile读之后的操作不会被重排到volatile读之前

第二个操作是volatile写时,不论第一个操作是什么,都不能重排序.这个操作保证了volatile写之前的操作不会重排到volatile之后

当一个操作为volatile写时,第二个操作为volatile读时不能重排

public class Test1 {
    static boolean flag=true;
    public static void main(String[] args) throws InterruptedException {
        Thread thread = new Thread(() -> {
            while (flag) {
                //不要输出会加锁
            }
            System.out.println("程序停止");
        });
        thread.start();
        TimeUnit.SECONDS.sleep(2);
        flag=false;
        System.out.println("stop");
    }
}

程序会一直执行不会停下来

如果给flag设置为volatile就不会发生这个问题

业务场景

• 单一赋值标志,but含符合运算赋值不可以i++之类
• 状态标志,判断业务是否结束
• 开销较低,写锁的策略
• DCL双端锁的发布

class Main{
    private static Main main;
    private Main(){};
    public static Main getMain() {
        //在synchronized外面加一个判断是防止代码每次都进入synchronized保证效率
        if (main==null){
            synchronized (Main.class){
                if (main==null){
                    main = new Main();
                }
            }
        }
        return main;

    }
}

• 创建一个对象,将对象引用入栈
• 复制一份对象的引用
• 利用对象的引用调用构造方法
• 利用对象的引用赋值给static main

也许jvm会优化为,先将对象的引用赋值,再调用构造方法

如果构造对象未完成,其他线程进入了第一个if判断就会直接通过

synchronized只有完全保护一个变量,才能解决原子性,可见性,有序性问题,而上面的例子没有完全保护该变量

解决的思路是,给main加上volatile

利用volatile禁止初始化对象和设置singleton指向内存空间的重排序

jvm在吧字节码生成机器码的时候,发现操作是volatile变量的话,会按照JMM的规范,在响应位置插入内存屏障