异步回调 异步回调对将来某个事件的结果进行建模package com.sw.future; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; /** * @Author suaxi * @Date 2021/2/18 23:40 * 异步回调 */ public class Test01 { 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"); // }); // System.out.println("无返回值的异步回调测试"); // completableFuture.get(); //有返回值的 supplyAsync 异步回调 CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> { System.out.println(Thread.currentThread().getName() + " supplyAsync"); int i = 1/0; return 200; }); System.out.println(completableFuture.whenComplete((t, u) -> { System.out.println("t= " + t); //正常返回的结果 System.out.println("u= " + u); //错误信息：java.util.concurrent.CompletionException: java.lang.ArithmeticException: / by zero }).exceptionally((e) -> { //链式编程 System.out.println(e.getMessage()); return 404; //错误信息的返回结果 }).get()); } }

ForkJoin ForkJoinForkJoin图片来源：狂神说JavaForkJoin 任务窃取这个任务中维护的都是双端队列B的任务执行完了，去帮助A执行图片来源：狂神说Javapackage com.sw.forkjoin; import java.util.concurrent.RecursiveTask; /** * @Author suaxi * @Date 2021/2/16 17:24 * 1、通过forkjoinPool来执行 * 2、计算任务forkjoinPool.execute(ForkJoinTask task) * 3、计算类要继承 ForkJoinTask */ public class ForkJoinTest extends RecursiveTask<Long> { private Long start; private Long end; //临界值 private Long temp = 1000L; public ForkJoinTest(Long start, Long end) { this.start = start; this.end = end; } //计算任务 @Override protected Long compute() { 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; ForkJoinTest task1 = new ForkJoinTest(start, middle); task1.fork(); //拆分任务，把任务压入线程队列 ForkJoinTest task2 = new ForkJoinTest(middle+1, end); task2.fork(); //同理 return task1.join() + task2.join(); } } } 测试类：package com.sw.forkjoin; import java.util.concurrent.ExecutionException; import java.util.concurrent.ForkJoinPool; import java.util.concurrent.ForkJoinTask; import java.util.stream.LongStream; /** * @Author suaxi * @Date 2021/2/16 17:37 */ public class Test { public static void main(String[] args) throws ExecutionException, InterruptedException { //test01(); //计算时间=7166 //test02(); //计算时间=5244 test03(); //计算时间=312 } //常规计算方法 public static void test01(){ Long sum =0L; long start = System.currentTimeMillis(); for (Long i = 1L; i < 10_0000_0000; i++) { sum += i; } long end = System.currentTimeMillis(); System.out.println("sum="+sum+" 计算时间="+ (end - start)); } //ForkJoin public static void test02() throws ExecutionException, InterruptedException { long start = System.currentTimeMillis(); //1、通过ForkJoinPool执行 ForkJoinPool forkJoinPool = new ForkJoinPool(); ForkJoinTask<Long> task = new ForkJoinTest(0L, 10_0000_0000L); //2、计算任务forkjoinPool.execute(ForkJoinTask task) ForkJoinTask<Long> submit = forkJoinPool.submit(task); Long sum = submit.get(); long end = System.currentTimeMillis(); System.out.println("sum="+sum+" 计算时间="+ (end - start)); } //Stream并行流 public static void test03(){ long start = System.currentTimeMillis(); //parallel并行 Long sum = LongStream.rangeClosed(0L,10_0000_0000L).parallel().reduce(0,Long::sum); long end = System.currentTimeMillis(); System.out.println("sum="+sum+" 计算时间="+ (end - start)); } }

Stream流式计算 Stream流式计算将计算交给流来操作package com.sw.stream; import java.util.Arrays; import java.util.List; /** * @Author suaxi * @Date 2021/2/16 10:33 * 有5个用户，进行筛选 * 1、ID为偶数 * 2、年龄大于23岁 * 3、用户名转换为大写 * 4、用户名倒序排列 * 5、只输出一个用户 */ public class StreamTest { public static void main(String[] args) { User user1 = new User(1, "a", 20); User user2 = new User(2, "b", 21); User user3 = new User(3, "c", 22); User user4 = new User(4, "d", 23); User user5 = new User(6, "e", 24); //转换为List集合 List<User> list = Arrays.asList(user1, user2, user3, user4, user5); list.stream() .filter(u ->{return u.getId()%2==0;}) .filter(u ->{return u.getAge()>23;}) .map(u ->{return u.getName().toUpperCase();}) .sorted((u1,u2) ->{return u2.compareTo(u1);}) .limit(1) .forEach(System.out::println); } }

四大函数式接口 四大函数式接口函数式接口只有一个方法的接口4大函数式接口：consumer，funtion，predicate，supplierFunction函数式接口package com.sw.function; import java.util.function.Function; /** * @Author suaxi * @Date 2021/2/15 22:00 * Function函数式接口 */ public class Test01 { public static void main(String[] args) { // Function<String, String> function = new Function<String, String>() { // @Override // public String apply(String s) { // return s; // } // }; //lambda表达式简化 Function<String, String> function = (s) ->{ return s; }; System.out.println(function.apply("函数式接口测试")); } } Predicate断定型接口package com.sw.function; import java.util.function.Predicate; /** * @Author suaxi * @Date 2021/2/15 22:09 * Predicate断定型接口,返回值为boolean类型 */ public class PredicateTest { public static void main(String[] args) { // Predicate<String> predicate = new Predicate<String>() { // @Override // public boolean test(String s) { // return s.isEmpty(); // } // }; Predicate<String> predicate = (str) ->{ return str.isEmpty(); }; System.out.println(predicate.test("abc")); } } Consummer消费型接口package com.sw.function; import java.util.function.Consumer; /** * @Author suaxi * @Date 2021/2/15 22:17 * Consummer消费型接口，只有输入参数，没有返回值 */ public class ConsumerTest { public static void main(String[] args) { // Consumer<String> consumer = new Consumer<String>() { // @Override // public void accept(String s) { // System.out.println(s); // } // }; Consumer<String> consumer = (str) ->{ System.out.println(str); return; }; consumer.accept("消费型接口测试"); } } Supplier供给型接口package com.sw.function; import java.util.function.Supplier; /** * @Author suaxi * @Date 2021/2/15 22:25 * Supplier 供给型接口 没有输入参数，只有返回值 */ public class SupplierTest { public static void main(String[] args) { // Supplier<String> supplier = new Supplier<String>() { // @Override // public String get() { // return "你好，谢谢"; // } // }; Supplier<String> supplier = () -> { return "你好，谢谢"; }; System.out.println(supplier.get()); } }

线程池 线程池线程池：三大方法、7大参数、4种拒绝策略池化技术事先准备好一些资源，要用的时候从池子中取，用完之后还回去优点：降低资源的消耗提高响应速度方便管理三大方法package com.sw.pool; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; /** * @Author suaxi * @Date 2021/2/14 21:23 */ public class PoolTest { public static void main(String[] args) { //ExecutorService pool = Executors.newSingleThreadExecutor(); //单个线程 //ExecutorService pool = Executors.newFixedThreadPool(10); //创建一个固定大小的线程池 ExecutorService pool = Executors.newCachedThreadPool(); //线程池大小可弹性伸缩 try { for (int i = 0; i < 50; i++) { pool.execute(() -> { System.out.println(Thread.currentThread().getName() + "--->Pool Test"); }); } } catch (Exception e) { e.printStackTrace(); } finally { //线程池用完，程序结束，关闭线程池 pool.shutdown(); } } } 七大参数 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.acc = System.getSecurityManager() == null ? null : AccessController.getContext(); this.corePoolSize = corePoolSize; this.maximumPoolSize = maximumPoolSize; this.workQueue = workQueue; this.keepAliveTime = unit.toNanos(keepAliveTime); this.threadFactory = threadFactory; this.handler = handler; }图片来源：狂神说Java手动创建一个线程池package com.sw.pool; import java.util.concurrent.*; /** * @Author suaxi * @Date 2021/2/14 21:23 * 4种拒绝策略： * AbortPolicy() 线程池满了，还有人进来，不处理这个人的请求，同时抛出异常 * CallerRunsPolicy() 哪里来的请求，回到哪里去（请求对应的主方法） * DiscardPolicy() 队列满了，丢掉后来的请求，不抛出异常 * DiscardOldestPolicy() 队列满了，尝试去和最早进行服务的窗口竞争，看那里的服务是否快要结束了，结束就进去提出请求，这种策略也不抛出异常 */ public class ThreadPoolExecutorTest { public static void main(String[] args) { //自定义线程池 ExecutorService pool = new ThreadPoolExecutor( 2, 5, 3, TimeUnit.SECONDS, new LinkedBlockingDeque<>(3), Executors.defaultThreadFactory(), new ThreadPoolExecutor.DiscardOldestPolicy()); try { for (int i = 0; i < 50; i++) { pool.execute(() -> { System.out.println(Thread.currentThread().getName() + "--->Pool Test"); }); } } catch (Exception e) { e.printStackTrace(); } finally { //线程池用完，程序结束，关闭线程池 pool.shutdown(); } } } 4种拒绝策略AbortPolicy() 线程池满了，还有人进来，不处理这个人的请求，同时抛出异常CallerRunsPolicy() 哪里来的请求，回到哪里去（请求对应的主方法）DiscardPolicy() 队列满了，丢掉后来的请求，不抛出异常DiscardOldestPolicy() 队列满了，尝试去和最早进行服务的窗口竞争，看那里的服务是否快要结束了，结束就进去提出请求，这种策略也不抛出异常小结1、线程池不允许使用Executor创建（不安全），而是通过ThreadPoolExecutor这样的方式2、如何定义线程池最大的大小？(1) CPU密集型根据cpu核数来设定Runtime.getRuntime().availableProcessors();(2) IO密集型判断程序中十分耗IO的线程，一般设置为其两倍的大小

阻塞队列 阻塞队列写：如果队列满了，就必须阻塞等待取：如果队列是空的，必须阻塞等待生产阻塞队列的四组API 抛出异常不抛异常，有返回值等待阻塞等待超时添加addofferputoffer(,,)移除removepolltakepoll(,)检测队首元素elementpeek--测试Demo：package com.sw.blockingQueue; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.TimeUnit; /** * @Author suaxi * @Date 2021/2/14 19:57 * 阻塞队列 */ public class BlockingQueueTest { public static void main(String[] args) throws InterruptedException { test04(); } //1.抛出异常 public static void test01(){ //队列初始大小为3 ArrayBlockingQueue<Object> 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.element()); //检测队首元素 System.out.println("-----分割线-----"); System.out.println(blockingQueue.remove()); System.out.println(blockingQueue.remove()); System.out.println(blockingQueue.remove()); //java.util.NoSuchElementException 没有目标元素 System.out.println(blockingQueue.remove()); } //2.不抛出异常，有返回值 public static void test02(){ //队列初始大小为3 ArrayBlockingQueue<Object> 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()); System.out.println("-----分割线-----"); System.out.println(blockingQueue.poll()); System.out.println(blockingQueue.poll()); System.out.println(blockingQueue.poll()); System.out.println(blockingQueue.poll()); //null } //3.等待，阻塞（一直阻塞） public static void test03() throws InterruptedException { //队列初始大小为3 ArrayBlockingQueue<Object> blockingQueue = new ArrayBlockingQueue<>(3); blockingQueue.put("A"); blockingQueue.put("B"); blockingQueue.put("C"); //blockingQueue.put("D"); //位置不够了，会一直阻塞 System.out.println("-----分割线-----"); System.out.println(blockingQueue.take()); System.out.println(blockingQueue.take()); System.out.println(blockingQueue.take()); System.out.println(blockingQueue.take()); //没有这个元素，会一直阻塞 } //4.等待，阻塞（等待超时） public static void test04() throws InterruptedException { //队列初始大小为3 ArrayBlockingQueue<Object> 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",2, TimeUnit.SECONDS)); //等待超过2秒退出 System.out.println(blockingQueue.peek()); System.out.println("-----分割线-----"); System.out.println(blockingQueue.poll()); System.out.println(blockingQueue.poll()); System.out.println(blockingQueue.poll()); System.out.println(blockingQueue.poll(2,TimeUnit.SECONDS)); //等待超过2秒退出 } } SynchronousQueue 同步队列没有容量，put一个元素之后，必须先take取出来，才能再put进去值package com.sw.blockingQueue; import java.util.concurrent.BlockingQueue; import java.util.concurrent.SynchronousQueue; import java.util.concurrent.TimeUnit; /** * @Author suaxi * @Date 2021/2/14 20:58 * 同步队列 */ public class SynchronousQueueTest { public static void main(String[] args) { //没有初始容量，put之后必须进行take，才能再进行put操作 BlockingQueue<String> synchronousQueue = new SynchronousQueue<>(); //put new Thread(() -> { try { System.out.println(Thread.currentThread().getName() + "进行put01操作"); synchronousQueue.put("Hello"); System.out.println(Thread.currentThread().getName() + "进行put02操作"); synchronousQueue.put("nihao"); System.out.println(Thread.currentThread().getName() + "进行put03操作"); synchronousQueue.put("xiexie"); } catch (InterruptedException e) { e.printStackTrace(); } }, "A").start(); //take new Thread(() -> { try { TimeUnit.SECONDS.sleep(3); System.out.println(Thread.currentThread().getName() + "--->" + synchronousQueue.take()); TimeUnit.SECONDS.sleep(3); System.out.println(Thread.currentThread().getName() + "--->" + synchronousQueue.take()); TimeUnit.SECONDS.sleep(3); System.out.println(Thread.currentThread().getName() + "--->" + synchronousQueue.take()); } catch (InterruptedException e) { e.printStackTrace(); } }, "B").start(); /* 输出结果： A进行put01操作 B--->Hello A进行put02操作 B--->nihao A进行put03操作 B--->xiexie */ } }

ReadWriteLock读写锁 读写锁ReadWriteLock读写锁，一个用于只读操作，一个用于写入package com.sw.readWriteLock; import java.util.HashMap; import java.util.Map; import java.util.concurrent.locks.ReadWriteLock; import java.util.concurrent.locks.ReentrantReadWriteLock; /** * @Author suaxi * @Date 2021/2/14 0:10 * ReadWriteLock 读写锁 * 独占锁（写） 一次只能被一个线程占有 * 共享锁（读） 可以被多个线程同时占有 * 读-读 可以共存 * 读-写 不能共存 * 写-写 不能共存 */ public class ReadWriteLockTest { 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 MyCacheLock{ private volatile Map<String,Object> map = new HashMap<>(); //相比于之前的lock锁，此处对线程的控制更加细粒度 private ReadWriteLock readWriteLock = new ReentrantReadWriteLock(); //存 写 public void put(String key,Object value){ try { readWriteLock.writeLock().lock(); System.out.println(Thread.currentThread().getName()+"写入---"+key); map.put(key, value); System.out.println(Thread.currentThread().getName()+"写入完成"); }catch (Exception e){ e.printStackTrace(); }finally { readWriteLock.writeLock().unlock(); } } //取 读 public void get(String key){ try { readWriteLock.readLock().lock(); System.out.println(Thread.currentThread().getName()+"读取---"+key); map.get(key); System.out.println(Thread.currentThread().getName()+"读取完成"); }catch (Exception e){ e.printStackTrace(); }finally { readWriteLock.readLock().unlock(); } } } class MyCache{ private volatile Map<String,Object> map = new HashMap<>(); //存 写 public void put(String key,Object value){ System.out.println(Thread.currentThread().getName()+"写入---"+key); map.put(key, value); System.out.println(Thread.currentThread().getName()+"写入完成"); } //取 读 public void get(String key){ System.out.println(Thread.currentThread().getName()+"读取---"+key); map.get(key); System.out.println(Thread.currentThread().getName()+"读取完成"); } }未使用读写锁之前，存在线程占用的问题：使用读写锁之后，执行顺序为：A写入--->A写入完成，B写入--->B写入完成，依次进行

JUC常用的辅助类 常用的辅助类1、CountDownLatch允许一个或多个线程等待其他线程完成操作package com.sw.assist; import java.util.concurrent.CountDownLatch; /** * @Author suaxi * @Date 2021/2/12 21:41 * <p> * assist 辅助类 */ //减法计数器 public class CountDownLatchTest { 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() + "--->走出教室"); countDownLatch.countDown(); //数量减1 }, String.valueOf(i)).start(); } countDownLatch.await(); //等待计数器归零，再向下执行 System.out.println("学生放学离开教室，班长锁门"); } } CountDownLatch原理：countDownLatch.countDown(); //数量减1countDownLatch.await(); //等待计数器归零，再向下执行执行方法，线程调用countDown();，使计数器数量减1，当计数器变为0时，countDownLatch.await();就会被唤醒，继续执行方法2、CyclicBarrier实现一组线程互相等待，当所有的线程都到达某个屏障点后再进行后续的操作package com.sw.assist; import java.util.concurrent.BrokenBarrierException; import java.util.concurrent.CyclicBarrier; /** * @Author suaxi * @Date 2021/2/12 21:56 */ //加法计数器 public class CyclicBarrierTest { public static void main(String[] args) { //10个学生都到教室后，老师才开始上课 CyclicBarrier cyclicBarrier = new CyclicBarrier(10, () -> { System.out.println("孙老师开始上课"); }); for (int i = 1; i <= 10; i++) { //注：lambda表达式不能直接操作for循环中的i，它只是简化了new对象的过程 final int temp = i; new Thread(() -> { 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.sw.assist; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; /** * @Author suaxi * @Date 2021/2/12 22:10 */ //停车位 public class SemaphoreTest { public static void main(String[] args) { //线程数量：提供3个停车位 Semaphore semaphore = new Semaphore(3); for (int i = 1; i <= 6; i++) { new Thread(() -> { try { semaphore.acquire(); //得到 System.out.println(Thread.currentThread().getName() + "---驶入停车位"); TimeUnit.SECONDS.sleep(3); System.out.println(Thread.currentThread().getName() + "---离开停车位"); } catch (InterruptedException e) { e.printStackTrace(); } finally { semaphore.release(); //释放 } }, String.valueOf(i)).start(); } } } semaphore.acquire(); //得到，如果当前提供的线程已经满了，则等待，直到线程被释放semaphore.release(); //释放，释放当前的信号量，然后唤醒等待的线程作用：多个共享资源互斥的使用；并发限流（控制最大线程数）