JUC并发编程（一）

1、什么是JUC

java.util.concurrent
java.util.concurrent.atomic 原子性
java.util.concurrent.locks lock
java.util.function

这部分面试高频

java.util 工具包、包、分类

业务：普通的线程代码 Thread
Runnable 没有返回值、效率相比Callable相对较低！

2、线程和进程

1）一句话表示

进程：一个程序 QQ.exe Music.exe 程序的集合
一个进程往往可以包含多个线程，至少包含一个！

java默认有几个线程？ 2个 main、GC（垃圾回收）

线程：开了一个进程，写字，自动保存（线程负责的）

对于java而言开启线程的方式：Thread、Runnable、Callable

java真的可以开启线程吗？不可以！！！

 public synchronized void start() {/*** This method is not invoked for the main method thread or "system"* group threads created/set up by the VM. Any new functionality added* to this method in the future may have to also be added to the VM.** A zero status value corresponds to state "NEW".*/if (threadStatus != 0)throw new IllegalThreadStateException();/* Notify the group that this thread is about to be started* so that it can be added to the group's list of threads* and the group's unstarted count can be decremented. */group.add(this);boolean started = false;try {start0();started = true;} finally {try {if (!started) {group.threadStartFailed(this);}} catch (Throwable ignore) {/* do nothing. If start0 threw a Throwable thenit will be passed up the call stack */}}}//本地方法，底层的C++，java无法直接操作硬件private native void start0();

是通过本地方法，底层的C++，java无法直接操作硬件

2）并发、并行

并发编程：并发、并行

并发（多线程操作同一个资源）

CPU单核，模拟出来多条线程，快速交替

并行（多个人一起行走）

CPU多核，多个线程可以同时执行；线程池可以用来提高性能

并发编程的本质：充分利用CPU的资源

3）线程有几个状态

 public enum State {//新生NEW,/**运行*/RUNNABLE,/**阻塞*/BLOCKED,/**等待，死死的等*/WAITING,/**超时等待*/TIMED_WAITING,/**终止*/TERMINATED;}

4）wait、sleep区别

1、来自不同的类
wait-》Object
Sleep-》Thread

import java.util.concurrent.TimeUnit;// 企业中使用睡眠的方式，不用sleepTimeUnit.DAYS.sleep(1);  //睡一天TimeUnit.SECONDS.sleep(2);  //睡2秒

2、关于锁的释放
wait会释放锁
sleep睡觉了，抱着锁睡觉，不会释放！

3、使用的范围不同

wait：必须在同步代码块中
sleep：可以在任何地方睡

4、是否需要捕获异常
wait 不需要捕获异常，但是有中断异常（只要是线程都会有中断异常）
sleep必须要捕获异常

   try {Thread.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}

3、Lock锁（重点）

1）传统 synchronized

非企业常用的方式：

企业的写法：

        synchronized (this){//this默认是锁这个方法
//            但是也可以用Ticket.class代替，代表锁这个对象}

2）lock

所有已知实现类：
ReentrantLock 可重入锁， ReentrantReadWriteLock.ReadLock 读写， ReentrantReadWriteLock.WriteLock 写锁

公平锁：十分公平，可以先来后到
非公平锁：十分不公平，可以插队（默认）

synchronized 非公平锁

3）synchronized 与 lock 区别

1、synchronized 内置的java关键字，Lock是一个java类
2、synchronized 无法判断获取锁的状态，Lock 可以判断是否获取到了锁
3、synchronized 会自动释放锁，Lock 必须手动释放锁！如果不释放锁，死锁
4、synchronized 线程1（获得锁，阻塞）、线程2（等待，傻傻的等）；Lock锁不一定会等待下去（lock.tryLock();//尝试获取锁）
5、synchronized 可重入锁，不可以中断的，非公平的；Lock 可重入锁，可以判断锁，非公平（可以自己设置）
6、synchronized 适合锁少量的代码同步问题，Lock适合锁大量的同步代码！

4）锁是什么？如何判断锁的是谁？

4、生产者和消费者问题

synchronized 版 wait 、notify
juc lock

1）生产者和消费者问题synchronized 版

面试：单例模式、排序算法、生产者和消费者、死锁

2）但是存在问题：

当有多个生产者和消费者的时候有问题：

注意点：防止虚假唤醒问题！！！

方法： if 改为while方法

3）juc版的生产者和消费者问题

通过lock找到condition

代码实现：

怎么实现包裹：

任何一个新的技术，绝对不是仅仅只是覆盖了原来的技术，会有新的！

4）Condition 精准的通知和唤醒线程

代码实现：

运行结果：

5、8锁现象

如果判断锁的是谁？

锁的只有2个：对象和class

1）2）

3）

4）

5）6）

7）8）

总结

new this 具体的一个手机
static class唯一的一个模板

6、集合类不安全

1）List不安全

2）set不安全

hashSet底层是什么？

底层是hashMap

add set本质就是map key是无法重复的！

这是一个不变的值！

3）map不安全

7、Callable

Callable接口类似于Runnable ，因为它们都是为其实例可能由另一个线程执行的类设计的。然而，A Runnable不返回结果，也不能抛出被检查的异常。

1、可以有返回值
2、可以抛出异常
3、方法不同，run（）、call（）

代码测试：

泛型的参数等于函数的返回值

package com.kuang.callable;import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;public class CallableTset {public static void main(String[] args) throws ExecutionException, InterruptedException {//        new Thread(new MyThread()).start();MyThread myThread = new MyThread();//适配类FutureTask futureTask = new FutureTask(myThread);new Thread(futureTask,"A").start();String s = (String)futureTask.get();//获取Callable返回结果System.out.println(s);/*** Thread只能有Runnable为参数* 所以如果要使用Callable作为参数，需要使用 FutureTask作为中间的关系*/}
}
//class MyThread implements Runnable{//
//    @Override
//    public void run() {//
//    }
//}
class MyThread implements Callable<String> {@Overridepublic String call(){System.out.println("call()");return "123";}
}

FutureTask 是 Runnable 的实现类

注意：
1、会有缓存
2、结果可能需要等待，会阻塞！

8、常用的辅助类

1）CountDownLatch

代码：

package com.kuang.add;import java.util.concurrent.CountDownLatch;//计数器
public class CountDownLatchDemo {public static void main(String[] args) throws InterruptedException {//总数为6 , 必须要执行任务的时候在使用CountDownLatch countDownLatch = new CountDownLatch(6);for(int i=1;i<=6;i++){new Thread(()->{System.out.println(Thread.currentThread().getName()+" go out");countDownLatch.countDown();//数量-1},String.valueOf(i)).start();}countDownLatch.await();//等待计数器归零，然后在向下执行System.out.println("close door");}
}

运行结果：

原理：

countDownLatch.countDown();//数量-1
countDownLatch.await();//等待计数器归零，然后在向下执行

每次有线程调用countDown（）方法就会数量-1，假设计数器变为0，countDownLatch.await()就会被唤醒，继续执行！

减法计数器

2）CyclicBarrier

加法计数器

代码：

package com.kuang.add;import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;//加法计数器
public class CyclicBarrierDemo {public static void main(String[] args) {//        集齐7可龙珠召唤神龙//        召唤龙珠的线程CyclicBarrier cyclicBarrier = new CyclicBarrier(7,()->{System.out.println("召唤神龙成功！");});for (int i = 1; i <=7 ; i++) {final int temp=i;//lambda 能操作到 i 吗new Thread(()->{//                System.out.println(i);
//                System.out.println(temp);System.out.println(Thread.currentThread().getName()+" 收集"+temp+"个龙珠");try {cyclicBarrier.await();//等待} catch (InterruptedException e) {e.printStackTrace();} catch (BrokenBarrierException e) {e.printStackTrace();}}).start();}}
}

运行结果：

3）Semaphore

信号量

代码：

package com.kuang.add;import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;public class SemaphoreDemo {public static void main(String[] args) {//线程数量，停车位     限流的时候可以用！Semaphore semaphore = new Semaphore(3);for (int i = 1; i <=6 ; i++) {new Thread(()->{//acquire()得到try {semaphore.acquire();System.out.println(Thread.currentThread().getName()+" get");TimeUnit.SECONDS.sleep(2);//等待2秒System.out.println(Thread.currentThread().getName()+" leave");} catch (InterruptedException e) {e.printStackTrace();}finally{semaphore.release();//release() 释放}//release() 释放}).start();}}
}

运行结果：

原理：

semaphore.acquire();//acquire()得到
假如如果已经满了，等待，等待被释放为止！

semaphore.release();//release() 释放
会将当前的信号量释放+1，然后唤醒等待的线程

作用：
多个共享资源互斥使用！并发限流，控制最大的线程数！

9、读写锁

1）ReadWriteLock

读的时候可以被多线程同时去读，写的时候只能有一个线程去写。

代码：

package com.kuang.rw;import jdk.nashorn.internal.ir.CallNode;
import org.omg.CORBA.PUBLIC_MEMBER;import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;/*** ReadWriteLock*/
//要求写入只能一个一个的来
public class ReadWriteLockDemo {public static void main(String[] args) {MyCacheLock myCache = new MyCacheLock();//写入for (int i = 1; i <=5 ; i++) {final int temp=i;new Thread(()->{myCache.put(temp+"",temp+"");},String.valueOf(i)).start();}//读取for (int i = 1; i <=5 ; i++) {final int temp=i;new Thread(()->{myCache.get(temp+"");},String.valueOf(i)).start();}}
}
/*** 自定义缓存*/
class MyCache{private volatile Map<String,Object> map=new HashMap<>();//存，写public void put(String key,Object value){System.out.println(Thread.currentThread().getName()+" write "+ key);map.put(key,value);System.out.println(Thread.currentThread().getName()+" write ok!");}//取，写public void get(String key){System.out.println(Thread.currentThread().getName()+" get "+ key);Object o=map.get(key);System.out.println(Thread.currentThread().getName()+" get ok! ");}
}
//加锁的
class MyCacheLock{private volatile Map<String,Object> map=new HashMap<>();//读写锁，更加细粒度的控制private ReadWriteLock readWriteLock=new ReentrantReadWriteLock();//存，写  ，写入的时候，只希望同时只有一个线程写public void put(String key,Object value){readWriteLock.writeLock().lock();try {System.out.println(Thread.currentThread().getName()+" write "+ key);map.put(key,value);System.out.println(Thread.currentThread().getName()+" write ok!");} catch (Exception e) {e.printStackTrace();}finally {readWriteLock.writeLock().unlock();}}//取，写  ，所有人都可以读！public void get(String key){readWriteLock.readLock().lock();try {System.out.println(Thread.currentThread().getName()+" get "+ key);Object o=map.get(key);System.out.println(Thread.currentThread().getName()+" get ok! ");} catch (Exception e) {e.printStackTrace();} finally {readWriteLock.readLock().unlock();}}
}

没有加锁的情况：

加入锁之后：

10、阻塞队列

1）BlockingQueue

阻塞队列：

SynchronousQueue ：同步队列

BlockingQueue 不是新的东西

什么时候会使用阻塞队列：
多线程并发处理、线程池！

deque：双端队列

学会使用队列！！！

添加、移除

四组API
1、抛出异常
2、不会抛出异常
3、阻塞等待
4、超时等待

package com.kuang.bq;import java.util.concurrent.ArrayBlockingQueue;public class Test {public static void main(String[] args) {/*** Collection* set* list* BlockingQueue*/test1();}/*** 抛出异常*/public static void test1(){//队列的大小ArrayBlockingQueue blockingQueue = new ArrayBlockingQueue<>(3);System.out.println(blockingQueue.add("a"));System.out.println(blockingQueue.add("b"));System.out.println(blockingQueue.add("c"));//java.lang.IllegalStateException: Queue full   抛出异常
//        System.out.println(blockingQueue.add("d"));System.out.println(blockingQueue.remove());System.out.println(blockingQueue.remove());System.out.println(blockingQueue.remove());//        java.util.NoSuchElementExceptionSystem.out.println(blockingQueue.remove());}
}

    /*** 不抛出异常的*/public static void test2(){//队列的大小ArrayBlockingQueue blockingQueue = new ArrayBlockingQueue<>(3);System.out.println(blockingQueue.offer("a"));System.out.println(blockingQueue.offer("b"));System.out.println(blockingQueue.offer("c"));
//        System.out.println(blockingQueue.offer("d"));//false  没有异常抛出System.out.println(blockingQueue.peek());//查看队首元素是谁//aSystem.out.println(blockingQueue.poll());System.out.println(blockingQueue.poll());System.out.println(blockingQueue.poll());System.out.println(blockingQueue.poll());//null 不抛出异常}

    /*** 等待，阻塞（一直等待）*/public static void test3() throws InterruptedException {//队列的大小ArrayBlockingQueue blockingQueue = new ArrayBlockingQueue<>(3);//一直阻塞blockingQueue.put("a");blockingQueue.put("b");blockingQueue.put("c");
//        blockingQueue.put("d");//队列没有位置了，一直阻塞System.out.println(blockingQueue.take());System.out.println(blockingQueue.take());System.out.println(blockingQueue.take());
//        System.out.println(blockingQueue.take());//没有这个元素，一直阻塞}

  /*** 等待，阻塞（等待超时）*/public static void test4() throws InterruptedException {//队列的大小ArrayBlockingQueue blockingQueue = new ArrayBlockingQueue<>(3);blockingQueue.offer("a");blockingQueue.offer("b");blockingQueue.offer("c");blockingQueue.offer("d",2, TimeUnit.SECONDS);//超时退出blockingQueue.poll();blockingQueue.poll();blockingQueue.poll();blockingQueue.poll(2,TimeUnit.SECONDS);//等待超过2秒就退出}

2） SynchronousQueue ：同步队列

没有容量

进去一个元素，必须等待取出来之后，才能再往里面放进一个元素！

put、take

相当于容量为1

代码：

package com.kuang.bq;import java.util.concurrent.BlockingQueue;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;//同步队列/*** 和其他的BlockingQueue不一样，SynchronousQueue 不存储元素* put了一个元素，必须从里面先take取出来，否则不能再put进去值*/
public class SynchronousQueueDemo {public static void main(String[] args) {BlockingQueue<String> blockingQueue = new SynchronousQueue<>();new Thread(()->{try {System.out.println(Thread.currentThread().getName()+" put 1");blockingQueue.put("1");System.out.println(Thread.currentThread().getName()+" put 2");blockingQueue.put("2");System.out.println(Thread.currentThread().getName()+" put 3");blockingQueue.put("3");} catch (InterruptedException e) {e.printStackTrace();}},"T1").start();new Thread(()->{try {TimeUnit.SECONDS.sleep(3);System.out.println(Thread.currentThread().getName()+" "+blockingQueue.take());TimeUnit.SECONDS.sleep(3);System.out.println(Thread.currentThread().getName()+" "+blockingQueue.take());TimeUnit.SECONDS.sleep(3);System.out.println(Thread.currentThread().getName()+" "+blockingQueue.take());} catch (InterruptedException e) {e.printStackTrace();}},"T2").start();}
}

运行结果：

11、线程池

线程池：三大方法、7大参数、4种拒绝策略

1）池化技术

程序的运行，本质：占用系统的资源！
所以，我们要优化资源的使用-》池化技术！

线程池、连接池、内存池、对象池。。。

jdk连接池：最小的值、和最大的值

**池化技术：**事先准备好一些资源，有人要用，就来我这里拿，用完之后还给我。
（默认大小、max）

线程池的好处：
1、降低资源的消耗
2、提高响应的速度
3、方便管理

线程复用，可以控制最大并发数，可以管理线程

2）线程池：三大方法

package com.kuang.pool;import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;//Executors 工具类，3大方法
public class Demo01 {public static void main(String[] args) {//        ExecutorService service = Executors.newSingleThreadExecutor();//单个线程
//        ExecutorService service = Executors.newFixedThreadPool(5);//创建一个固定的线程池的大小ExecutorService service =Executors.newCachedThreadPool();//可伸缩的try{for (int i = 0; i < 10; i++) {// 使用了线程池之后，要使用线程池来创建线程service.execute(()->{System.out.println(Thread.currentThread().getName()+" ok!");});}}catch (Exception e){e.printStackTrace();}finally {// 线程池用完，程序结束，关闭线程池service.shutdown();}}}

3）7大参数

源码分析：

 public static ExecutorService newSingleThreadExecutor() {return new FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>()));}public static ExecutorService newFixedThreadPool(int nThreads) {return new ThreadPoolExecutor(nThreads, nThreads,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>());}public static ExecutorService newCachedThreadPool() {return new ThreadPoolExecutor(0, Integer.MAX_VALUE,//21亿，OOM60L, TimeUnit.SECONDS,new SynchronousQueue<Runnable>());}

本质：ThreadPoolExecutor

 public ThreadPoolExecutor(int corePoolSize,//核心线程大小int maximumPoolSize,//最大核心线程池大小long keepAliveTime,//超时了没有人调用就会释放TimeUnit unit,//超时单位BlockingQueue<Runnable> workQueue,//阻塞队列ThreadFactory threadFactory,// 线程工厂，创线程池，一般不用动RejectedExecutionHandler handler//拒绝策略) {if (corePoolSize < 0 ||maximumPoolSize <= 0 ||maximumPoolSize < corePoolSize ||keepAliveTime < 0)throw new IllegalArgumentException();if (workQueue == null || threadFactory == null || handler == null)throw new NullPointerException();this.corePoolSize = corePoolSize;this.maximumPoolSize = maximumPoolSize;this.workQueue = workQueue;this.keepAliveTime = unit.toNanos(keepAliveTime);this.threadFactory = threadFactory;this.handler = handler;}

超时：

拒绝策略：

手动创建一个线程池

package com.kuang.pool;import java.util.concurrent.*;//Executors 工具类，3大方法
public class Demo01 {public static void main(String[] args) {//        ExecutorService service = Executors.newSingleThreadExecutor();//单个线程
//        ExecutorService service = Executors.newFixedThreadPool(5);//创建一个固定的线程池的大小
//         ExecutorService service =Executors.newCachedThreadPool();//可伸缩的//自定义线程池！工作中  ThreadPoolExecutorExecutorService service = new ThreadPoolExecutor(2,5,3,TimeUnit.SECONDS,new LinkedBlockingDeque<>(3),Executors.defaultThreadFactory(),
//                new ThreadPoolExecutor.AbortPolicy()//银行满了，还有人进来，不处理这个人的，抛出异常
//                new ThreadPoolExecutor.CallerRunsPolicy()//哪来的去哪里！   main ok!
//                new ThreadPoolExecutor.DiscardPolicy()//队列满了，丢掉任务，不会抛出异常！new ThreadPoolExecutor.DiscardOldestPolicy()//队列满了，尝试和最早的去竞争，不会抛出异常！);try{//最大承载：deque + max
//            超出java.util.concurrent.RejectedExecutionException:for (int i = 0; i < 9; i++) {// 使用了线程池之后，要使用线程池来创建线程service.execute(()->{System.out.println(Thread.currentThread().getName()+" ok!");});}}catch (Exception e){e.printStackTrace();}finally {// 线程池用完，程序结束，关闭线程池service.shutdown();}}}

4）4种拒绝策略

//                new ThreadPoolExecutor.AbortPolicy()//银行满了，还有人进来，不处理这个人的，抛出异常
//                new ThreadPoolExecutor.CallerRunsPolicy()//哪来的去哪里！   main ok!
//                new ThreadPoolExecutor.DiscardPolicy()//队列满了，丢掉任务，不会抛出异常！new ThreadPoolExecutor.DiscardOldestPolicy()//队列满了，尝试和最早的去竞争，不会抛出异常！

5）小结和拓展

问题在于：池的最大的大小如何设置！

了解：IO密集型、CPU密集型（调优）

12、四大函数式接口

必会的：lambda表达式、链式编程、函数式接口、Stream流式计算

1）函数式接口：只有一个方法的接口

@FunctionalInterface
public interface Runnable {public abstract void run();
}

java中有很多的FunctionalInterface

简化编程模型，在新版本的框架底层大量应用！

foreach（消费者类的函数式接口）

代码测试：

1）Function函数式接口

package com.kuang.function;import java.util.function.Function;/*** Function 函数式接口* 有一个输入参数，有一个输出** 只要是函数型接口  可以用 lambda 表达式简化*/
public class Demo01 {public static void main(String[] args) {//工具类，输出输入的值
//        Function function = new Function<String,String>(){//            @Override
//            public String apply(String str) {//                return str;
//            }
//        };Function function=(str)->{return str;};System.out.println(function.apply("asd"));}
}

2）Predicate断定型接口：有一个输入参数，返回值只能是布尔值！

package com.kuang.function;import java.util.function.Predicate;/*** Predicate 断定型接口：* 有一个输入参数，返回值只能是 布尔值*/
public class Demo02 {public static void main(String[] args) {//判断字符串是否为空
//        Predicate predicate = new Predicate<String>(){//            @Override
//            public boolean test(String o) {//                return o.isEmpty();
//            }
//        };Predicate<String> predicate=(o)->{ return o.isEmpty();};System.out.println(predicate.test(""));System.out.println(predicate.test("abc"));}
}

3）Consumer 消费型接口

package com.kuang.function;import com.sun.xml.internal.ws.api.model.wsdl.WSDLOutput;
import jdk.nashorn.internal.ir.CallNode;
import org.w3c.dom.ls.LSOutput;import java.util.function.Consumer;/*** Consumer 消费型接口：只有输入，没有返回值*/public class Demo03 {public static void main(String[] args) {//        Consumer<String> consumer=new Consumer<String>(){//            @Override
//            public void accept(String o) {//                System.out.println(o);
//            }
//        };Consumer<String> consumer=(o)->{ System.out.println(o);} ;consumer.accept("123");}}

4）Supplier 供给型接口

package com.kuang.function;import java.util.function.Supplier;/*** Supplier : 供给型接口* 没有参数，只有返回值*/
public class Demo04 {public static void main(String[] args) {//        Supplier supplier = new Supplier<Integer>(){//
//            @Override
//            public Integer get() {//                System.out.println("get()");
//                return 1024;
//            }
//        };Supplier supplier=()->{ return 1024;};System.out.println(supplier.get());}
}

13、Stream流式计算

1）什么是Stream流式计算

大数据：存储+计算

存储：集合、mysql等等，他们的本质就是存储东西的

计算都应该来交给流来操作！

package com.kuang.stream;import java.util.Arrays;
import java.util.List;/*** 题目要求：一分钟之内完成，只能用一行代码实现！* 现有5个用户！筛选：* 1、ID必须为偶数* 2、年龄必须大于23岁* 3、用户名转为大写字母* 4、用户名字母倒着排序* 5、只能输出一个用户！*/
public class Test {public static void main(String[] args) {User u1=new User(1,"a",21);User u2=new User(2,"b",22);User u3=new User(3,"c",23);User u4=new User(4,"d",24);User u5=new User(6,"e",25);//集合就是用来存储，流用来计算List<User> users = Arrays.asList(u1, u2, u3, u4, u5);//        计算交给Stream流//链式编程users.stream().filter(u->{return u.getId()%2==0;}).filter(u->{return u.getAge()>23;}).map(u->{return u.getName().toUpperCase();}).sorted((uu1,uu2)->{return uu2.compareTo(uu1);}).limit(1).forEach(System.out::println);}
}

package com.kuang.stream;import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;//有参、无参构造、get、set、toString方法！
@Data
@NoArgsConstructor
@AllArgsConstructor
public class User {private int id;private String name;private int age;
}

14、ForkJoin

1）什么是ForkJoin

ForkJoin在JDK 1.7之后，并行执行任务！提高效率，大数据量！

大数据：Map Reduce（把大任务拆分成小任务）

2）ForkJoin特点：工作窃取

这个里面维护的都是双端队列

3）ForkJoin的操作

package com.kuang.forkjoin;
//求和计算
// 基本----》ForkJoin-----》Stream并行流import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;/*** 如何使用ForkJoin* 1、ForkJoinPool 通过它来执行* 2、计算任务 execute(ForkJoinTask<?> task)* 3、计算类要继承RecursiveTask<？>*/public class ForkJoinDemo extends RecursiveTask<Long> {private Long start;private Long end;//临界值private Long temp=10000L;public ForkJoinDemo(Long start,Long end){this.start=start;this.end=end;}//计算方法@Overrideprotected Long compute() {//基本操作
//        int sum=0;
//        for (int i = 1; i <=10000; i++) {//            sum+=i;
//        }
//        System.out.println(sum);if((end-start)>temp){Long sum=0L;for (Long i = start; i <=end; i++) {sum+=i;}return sum;}else{//forkJoin  递归Long middle=(start+end)/2;//中间值ForkJoinDemo task1 = new ForkJoinDemo(start, middle);task1.fork();//拆分任务，把任务压入线程队列ForkJoinDemo task2 = new ForkJoinDemo(start, middle);task2.fork();//拆分任务，把任务压入线程队列return task1.join() + task2.join();}}
}

测试：

package com.kuang.forkjoin;import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.stream.LongStream;//同一个任务，效率！！！
public class Test {public static void main(String[] args) throws ExecutionException, InterruptedException {test1();//22749
//        test2();//12159
//        test3();//427}public static void test1(){Long sum=0L;long start=System.currentTimeMillis();for (Long i = 1L; i <=10_0000_0000L; i++) {sum+=i;}long end=System.currentTimeMillis();System.out.println("sum="+sum+"time="+(end-start));}public static void test2() throws ExecutionException, InterruptedException {long start=System.currentTimeMillis();/*** 会使用ForkJoin*/ForkJoinPool forkJoinPool = new ForkJoinPool();ForkJoinTask<Long> task = new ForkJoinDemo(0L,10_0000_0000L);
//        forkJoinPool.execute(task);//执行任务ForkJoinTask<Long> submit = forkJoinPool.submit(task);//提交任务Long sum = submit.get();long end=System.currentTimeMillis();System.out.println("sum="+sum+"time="+(end-start));}public static void test3(){long start=System.currentTimeMillis();/*** Stream 并行流*/long sum=LongStream.rangeClosed(0L,10_0000_0000L).parallel().reduce(0,Long::sum);long end=System.currentTimeMillis();System.out.println("sum="+sum+"time="+(end-start));}
}

15、异步回调

1)Future设计的初衷：对将来的某个事件的结果进行建模

package com.kuang.future;import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;/*** 异步调用：Ajax* CompletableFuture** 异步执行* 成功回调* 失败回调*/
public class Demo01 {public static void main(String[] args) throws ExecutionException, InterruptedException {//        //发起一个请求
        没有返回值的 runAsync 异步回调
//        CompletableFuture<Void> completableFuture = CompletableFuture.runAsync(()->{//            try {//                TimeUnit.SECONDS.sleep(2);
//            } catch (InterruptedException e) {//                e.printStackTrace();
//            }
//            System.out.println(Thread.currentThread().getName()+"runAsync=>Void");
//        });
//
//        System.out.println("1111");
//
//        completableFuture.get();//获取阻塞执行结果// 有返回值的 runAsync 异步回调//ajax，成功和失败的回调
//        返回的是错误的信息CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {System.out.println(Thread.currentThread().getName()+"supplyAsync->Integer");int i=10/0;return 1024;});System.out.println(completableFuture.whenComplete((t, u) -> {System.out.println("t->" + t);//正常的返回结果System.out.println("u->" + u);//错误信息  java.util.concurrent.CompletionException}).exceptionally((e) -> {System.out.println(e.getMessage());return 233;//可以获取到错误的返回结果   java.lang.ArithmeticException: / by zero}).get());}
}

运行结果：

16、JMM

1）请你谈谈对Volatile的理解

Volatile 是java虚拟机提供的轻量级的同步机制

1、保证可见性
2、不保证原子性
3、禁止指令重排

2）什么是JMM

JVM：java虚拟机
JMM：java内存模型，不存在的东西，概念！约定！

关于JMM的一些同步的约定：

1、线程解锁前，必须把共享变量立刻刷回主存
2、线程加锁前，必须读取注册中的最新值到工作内存中！
3、加锁和解锁必须是同一把锁。

线程： 工作内存、主内存
8种操作：

java内存模型定义了8种操作来完成。这8种操作每一种都是原子操作。8种操作如下：

lock(锁定)：作用于主内存，它把一个变量标记为一条线程独占状态；
read(读取)：作用于主内存，它把变量值从主内存传送到线程的工作内存中，以便随后的load动作使用；
load(载入)：作用于工作内存，它把read操作的值放入工作内存中的变量副本中；
use(使用)：作用于工作内存，它把工作内存中的值传递给执行引擎，每当虚拟机遇到一个需要使用这个变量的指令时候，将会执行这个动作；
assign(赋值)：作用于工作内存，它把从执行引擎获取的值赋值给工作内存中的变量，每当虚拟机遇到一个给变量赋值的指令时候，执行该操作；
store(存储)：作用于工作内存，它把工作内存中的一个变量传送给主内存中，以备随后的write操作使用；
write(写入)：作用于主内存，它把store传送值放到主内存中的变量中。
unlock(解锁)：作用于主内存，它将一个处于锁定状态的变量释放出来，释放后的变量才能够被其他线程锁定；

Java内存模型还规定了执行上述8种基本操作时必须满足如下规则:

（1）不允许read和load、store和write操作之一单独出现（即不允许一个变量从主存读取了但是工作内存不接受，或者从工作内存发起会写了但是主存不接受的情况），以上两个操作必须按顺序执行，但没有保证必须连续执行，也就是说，read与load之间、store与write之间是可插入其他指令的。

（2）不允许一个线程丢弃它的最近的assign操作，即变量在工作内存中改变了之后必须把该变化同步回主内存。

（3）不允许一个线程无原因地（没有发生过任何assign操作）把数据从线程的工作内存同步回主内存中。

（4）一个新的变量只能从主内存中“诞生”，不允许在工作内存中直接使用一个未被初始化（load或assign）的变量，换句话说就是对一个变量实施use和store操作之前，必须先执行过了assign和load操作。

（5）一个变量在同一个时刻只允许一条线程对其执行lock操作，但lock操作可以被同一个条线程重复执行多次，多次执行lock后，只有执行相同次数的unlock操作，变量才会被解锁。

（6）如果对一个变量执行lock操作，将会清空工作内存中此变量的值，在执行引擎使用这个变量前，需要重新执行load或assign操作初始化变量的值。

（7）如果一个变量实现没有被lock操作锁定，则不允许对它执行unlock操作，也不允许去unlock一个被其他线程锁定的变量。

（8）对一个变量执行unlock操作之前，必须先把此变量同步回主内存（执行store和write操作）。

问题：程序不知道主内存中的值已经被修改过了！

package com.kuang.volatilet;import java.util.concurrent.TimeUnit;//存在问题的
public class JMMDemo {private static int num=0;public static void main(String[] args) {//mainnew Thread(()->{//线程1while(num==0){}}).start();try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}num=1;System.out.println(num);}
}

程序不会停止

17、Volatile

1）保证可见性

package com.kuang.volatilet;import java.util.concurrent.TimeUnit;//存在问题的
public class JMMDemo {//不加volatile 程序就会死循环！
//    加volatile可以保证可见性private volatile static int num=0;public static void main(String[] args) {//mainnew Thread(()->{//线程1    对主内存的变化不知道while(num==0){}}).start();try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}num=1;System.out.println(num);}
}

2）不保证原子性

原子性：不可分割
线程A执行任务的时候，不能被打扰，也不能被分割，要么同时成功，要么同时失败

package com.kuang.volatilet;
//不保证原子性
public class VDemo02 {private volatile static int num=0;public  static void add(){num++;}public static void main(String[] args) {//理论上num结果为2万for (int i = 1; i <=20; i++) {new Thread(()->{for(int j=0;j<1000;j++){add();}}).start();}while(Thread.activeCount()>2){//main  gcThread.yield();//yield就是当前线程把自己的cpu调度时间让给其它线程}System.out.println(Thread.currentThread().getName()+" "+num);//main 19165//main 20000   有synchronized//main 19205   有volatile   不保证原子性}
}

如果不加lock和synchronized怎么样保证原子性？？

使用原子类，解决原子性问题

原子类为什么这么高级？？

package com.kuang.volatilet;import java.util.concurrent.atomic.AtomicInteger;//不保证原子性
public class VDemo02 {//    volatile不保证原子性private static AtomicInteger num=new AtomicInteger();public  static void add(){//num++;// 不是一个原子性操作num.getAndIncrement();//AtomicInteger +1 方法  ，使用的是CAS}public static void main(String[] args) {//理论上num结果为2万for (int i = 1; i <=20; i++) {new Thread(()->{for(int j=0;j<1000;j++){add();}}).start();}while(Thread.activeCount()>2){//main  gcThread.yield();//yield就是当前线程把自己的cpu调度时间让给其它线程}System.out.println(Thread.currentThread().getName()+" "+num);//main 19165//main 20000   有synchronized//main 19205   有volatile   不保证原子性}
}

这些类的底层都直接和操作系统挂钩！直接在内存中修改值！Unsafe类是一个很特殊的存在！

3）指令重排

什么是指令重排：你写的程序，计算机并不是按照你写的那样去执行的。

源代码-》编译器优化的重排-》指令并行也可能会重排-》内存系统也会重排-》执行

处理器在进行指令重排的时候，考虑：数据之间的依赖性！

int x=1;
int y=2;
x=x+5;
y=x*x;我们期望的是：1234
但是可能执行的时为：2134 1324
但是不可能为4123

volatile 可以避免指令重排

内存屏障，CPU指令，作用：

1、保证特定的操作的执行顺序！
2、可以保证某些变量的内存可见性（利用这些特性volatile 实现了可见性）

volatile 可以保证可见性，不能保证原子性，由于内存屏障，可以保证避免指令重排的现象产生！

18、彻底玩转单例模式

饿汉式、DCL懒汉式

volatile 的内存屏障在单例模式中使用的最多

1）饿汉式

package com.kuang.single;//饿汉式单例
public class Hungry {//可能会浪费空间private byte[] data1=new byte[1024*1024];private byte[] data2=new byte[1024*1024];private byte[] data3=new byte[1024*1024];private byte[] data4=new byte[1024*1024];private Hungry(){}private final static Hungry HUNGRY=new Hungry();public static Hungry getInstance(){return HUNGRY;};
}

2）DCL懒汉式

package com.kuang.single;import java.lang.reflect.Constructor;
import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;//懒汉式单例
public class LazyMan {private static boolean qinjiang=false;private LazyMan(){synchronized (LazyMan.class){if(qinjiang==false){qinjiang=true;}else{throw new RuntimeException("不要试图使用反射破坏异常");}}System.out.println(Thread.currentThread().getName()+" ok!");}private volatile static LazyMan lazyMan;//双重检测模式的 懒汉式单例 DCL懒汉式单例public static LazyMan getInstance(){if(lazyMan==null){synchronized (LazyMan.class){if(lazyMan==null){lazyMan=new LazyMan();//不是原子性操作/*** 1、分配内存空间* 2、执行构造方法，初始化对象* 3、把这个对象指向这个空间** 所以 private volatile static LazyMan lazyMan;* 必须要加上 volatile*/}}}return lazyMan;}//    //多线程并发  多线程下有问题，所以需要加锁
//    public static void main(String[] args) {//        for (int i = 0; i < 10; i++) {//            new Thread(()->{//                lazyMan.getInstance();
//            }).start();
//        }
//    }//反射！public static void main(String[] args) throws NoSuchMethodException, IllegalAccessException, InvocationTargetException, InstantiationException, NoSuchFieldException {//        LazyMan instance = LazyMan.getInstance();Field qinjiang = LazyMan.class.getDeclaredField("qinjiang");qinjiang.setAccessible(true);Constructor<LazyMan> declaredConstructor = LazyMan.class.getDeclaredConstructor(null);declaredConstructor.setAccessible(true);LazyMan instance = declaredConstructor.newInstance();qinjiang.set(instance,false);LazyMan instance2 = declaredConstructor.newInstance();System.out.println(instance);System.out.println(instance2);/*** main ok!* main ok!* com.kuang.single.LazyMan@5cad8086* com.kuang.single.LazyMan@6e0be858*/}
}

3）静态内部类

package com.kuang.single;//静态内部类
public class Holder {//构造器私有private Holder(){}public static Holder getInstance(){return InnerClass.HOLDER;}public static class InnerClass{private static final Holder HOLDER=new Holder();}
}

4）单例不安全，因为反射的存在

package com.kuang.single;import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;//enum 是什么？ 本身是一个Class类
public enum EnumSingle {INSTANCE;public EnumSingle getInstance(){return INSTANCE;}
}
class Test{public static void main(String[] args) throws NoSuchMethodException, IllegalAccessException, InvocationTargetException, InstantiationException {EnumSingle instance1 = EnumSingle.INSTANCE;Constructor<EnumSingle> declaredConstructor = EnumSingle.class.getDeclaredConstructor(String.class,int.class);declaredConstructor.setAccessible(true);//Exception in thread "main" java.lang.NoSuchMethodExceptionEnumSingle instance2 = declaredConstructor.newInstance();System.out.println(instance1);System.out.println(instance2);}
}

有参构造：

枚举类型的最终反编译源码：