《JUC并发编程 - 基础篇》 Callable接口 | 辅助类 | 读写锁 | 阻塞队列 | 线程池 | Stream流

11.7 工作中线程池用哪个？

在工作中单一的/固定数的/可变的三种创建线程池的方法哪个用的多？超级大坑

答案是一个都不用，我们工作中只能使用自定义的

Executors中JDK已经给你提供了，为什么不用？

自定义线程池代码

public class MyThreadPoolDemo {
    public static void main(String[] args) {
        ExecutorService threadPool = new ThreadPoolExecutor(
                2,
                5,
                2L,
                TimeUnit.SECONDS,
                new LinkedBlockingDeque <>(3),
                Executors.defaultThreadFactory(),
                //new ThreadPoolExecutor.AbortPolicy()
                //new ThreadPoolExecutor.CallerRunsPolicy()
                //new ThreadPoolExecutor.DiscardOldestPolicy()
                new ThreadPoolExecutor.DiscardPolicy());
        try{
            //模拟有10个顾客来银行办理业务,目前池子里面有5个工作人员提供服务
            for (int i = 1; i <= 9; i++) {
                final int tempInt = i;
                threadPool.execute(()->{
                    System.out.println(Thread.currentThread().getName()+"\t 办理业务"+tempInt);
                });
                //暂停线程 ,线程会变得逐渐有序..
                // try { TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e) { e.printStackTrace(); }
            }
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            threadPool.shutdown();//释放线程池
        }
    }
}

12、Java8之流式计算复习

12.1 函数式接口

java内置核心四大函数式接口

如何使用

//R apply(T t);函数型接口，一个参数，一个返回值
Function<String,Integer> function = t ->{return t.length();};
System.out.println(function.apply("abcd"));
//boolean test(T t);断定型接口，一个参数，返回boolean
Predicate<String> predicate = t->{return t.startsWith("a");};
System.out.println(predicate.test("a"));
// void accept(T t);消费型接口，一个参数，没有返回值
Consumer<String> consumer = t->{
    System.out.println(t);
};
consumer.accept("javaXXXX");
//T get(); 供给型接口，无参数，有返回值
Supplier<String> supplier =()->{return UUID.randomUUID().toString();};
System.out.println(supplier.get());

private static void testFunction() {
    // Function<String,Integer> function = new Function <String, Integer>() {
    //     @Override
    //     public Integer apply(String s) {
    //         return s.length();
    //     }
    // };
    //R apply(T t);函数型接口，一个参数，一个返回值
    Function <String,Integer> function = s->{ return s.length(); };
    System.out.println(function.apply("abc"));
    //boolean test(T t);断定型接口，一个参数，返回boolean
    // Predicate<String> predicate = new Predicate <String>() {
    //     @Override
    //     public boolean test(String s) {
    //         return s.isEmpty();
    //     }
    // };
    Predicate <String> predicate = s->{return s.isEmpty();};
    System.out.println(predicate.test("abc"));
    // void accept(T t);消费型接口，一个参数，没有返回值
    // Consumer<String> consumer = new Consumer <String>() {
    //     @Override
    //     public void accept(String s) {
    //         System.out.println("I love Java!");
    //     }
    // };
    Consumer <String> consumer = s -> {System.out.println("I love Java!"); };
    consumer.accept("java");
    //T get(); 供给型接口，无参数，有返回值
    // Supplier<String> supplier = new Supplier <String>() {
    //     @Override
    //     public String get() {
    //         return "Java";
    //     }
    // };
    Supplier <String> supplier = ()->{
        return "Java";
    };
    System.out.println(supplier.get());
}

12.2 Stream流

12.2.1 是什么?

流(Stream) 到底是什么呢？

是数据渠道，用于操作数据源（集合、数组等）所生成的元素序列。

“集合讲的是数据，流讲的是计算！”

12.2.2 特点

Stream 自己不会存储元素
Stream 不会改变源对象。相反，他们会返回一个持有结果的新Stream。
Stream 操作是延迟执行的。这意味着他们会等到需要结果的时候才执行。

12.2.3 如何用

1.创建一个Stream：一个数据源（数组、集合）

2.中间操作：一个中间操作，处理数据源数据

3.终止操作：一个终止操作，执行中间操作链，产生结果

类比: 源头=>中间流水线=>结果

12.2.4 代码演示

package com.rg.juc;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.sql.SQLOutput;
import java.util.Arrays;
import java.util.List;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.function.Supplier;
/**
 * @author lxy
 * @version 1.0
 * @Description
 * @date 2022/5/4 20:51
 */
@Data
@NoArgsConstructor
@AllArgsConstructor
class User
{
    private Integer id;
    private String  userName;
    private int     age;
}
/**
 * @create 2019-02-26 22:24
 *
 * 题目：请按照给出数据，找出同时满足
 *      偶数ID且年龄大于24且用户名转为大写且用户名字母倒排序
 *      最后只输出一个用户名字
 */
public class StreamDemo {
    public static void main(String[] args) {
        User u1 = new User(11, "a", 23);
        User u2 = new User(12, "b", 24);
        User u3 = new User(13, "c", 22);
        User u4 = new User(14, "d", 28);
        User u5 = new User(16, "e", 26);
        List <User> list = Arrays.asList(u1, u2, u3, u4, u5);
        list.stream().filter(u->{
            return u.getId() % 2 ==0;
        }).filter(t->{
            return t.getAge() > 24;
        }).map(m->{
            return m.getUserName().toUpperCase();
        }).sorted((o1,o2)->{
            return o2.compareTo(o1);
        }).forEach(System.out::println);
    }
    private static void testFunction() {
        // Function<String,Integer> function = new Function <String, Integer>() {
        //     @Override
        //     public Integer apply(String s) {
        //         return s.length();
        //     }
        // };
        //R apply(T t);函数型接口，一个参数，一个返回值
        Function <String,Integer> function = s->{ return s.length(); };
        System.out.println(function.apply("abc"));
        //boolean test(T t);断定型接口，一个参数，返回boolean
        // Predicate<String> predicate = new Predicate <String>() {
        //     @Override
        //     public boolean test(String s) {
        //         return s.isEmpty();
        //     }
        // };
        Predicate <String> predicate = s->{return s.isEmpty();};
        System.out.println(predicate.test("abc"));
        // void accept(T t);消费型接口，一个参数，没有返回值
        // Consumer<String> consumer = new Consumer <String>() {
        //     @Override
        //     public void accept(String s) {
        //         System.out.println("I love Java!");
        //     }
        // };
        Consumer <String> consumer = s -> {System.out.println("I love Java!"); };
        consumer.accept("java");
        //T get(); 供给型接口，无参数，有返回值
        // Supplier<String> supplier = new Supplier <String>() {
        //     @Override
        //     public String get() {
        //         return "Java";
        //     }
        // };
        Supplier <String> supplier = ()->{
            return "Java";
        };
        System.out.println(supplier.get());
    }
}

13、分支合并框架

13.1 原理

Fork：把一个复杂任务进行分拆，大事化小

Join：把分拆任务的结果进行合并

13.2 相关类

ForkJoinPool

分支合并池类比=> 线程池

ForkJoinTask

ForkJoinTask 类比=> FutureTask

RecursiveTask

递归任务：继承后可以实现递归(自己调自己)调用的任务

 class Fibonacci extends RecursiveTask<Integer> {
   final int n;
   Fibonacci(int n) { this.n = n; }
   Integer compute() {
     if (n <= 1)
       return n;
     Fibonacci f1 = new Fibonacci(n - 1);
     f1.fork();
     Fibonacci f2 = new Fibonacci(n - 2);
     return f2.compute() + f1.join();
   }
 }

13.3 入门案例

/**
 * @author lxy
 * @version 1.0
 * @Description  分支合并案例
 * @date 2022/5/5 17:07
 */
class MyTask extends RecursiveTask<Integer> {//RecursiveTask:递归任务
    private static final Integer ADJUST_VALUE = 10;
    private int begin;
    private int end;
    private int result;
    public MyTask(int begin, int end) {
        this.begin = begin;
        this.end = end;
    }
    @Override
    protected Integer compute() {
        if(end-begin <= ADJUST_VALUE){
            for (int i = begin; i <= end; i++) {
                result = result + i;
            }
        }else{
            int middle = (end + begin) / 2;
            MyTask task01 = new MyTask(begin, middle);
            MyTask task02 = new MyTask(middle + 1, end);
            task01.fork();//进行递归执行
            task02.fork();
            result = task01.join()+task02.join();//将结果进行合并
        }
        return result;
    }
}
public class ForkJoinDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        MyTask myTask = new MyTask(0, 10);
        ForkJoinPool threadPool = new ForkJoinPool();
        ForkJoinTask <Integer> forkJoinTask = threadPool.submit(myTask);
        System.out.println(forkJoinTask.get());
        threadPool.shutdown();//关闭线程池
    }
}

14、异步回调

14.1 原理

14.2 入门案例

/**
 * @author lxy
 * @version 1.0
 * @Description 同步，异步，异步回调 案例
 * @date 2022/5/5 18:18
 */
public class CompletableFutureDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        //同步，异步，异步回调
        CompletableFuture<Void> completableFuture = CompletableFuture.runAsync(()->{
            System.out.println(Thread.currentThread().getName()+"没有返回,update mysql ok");
        });
        completableFuture.get();
        //异步回调
        CompletableFuture<Integer> completableFuture2 = CompletableFuture.supplyAsync(()->{
            System.out.println(Thread.currentThread().getName()+"\t completableFuture2");
            int age = 10 / 0;
            return 1024;
        });
        completableFuture2.whenComplete((t,u)->{
            System.out.println("------t="+t);
            System.out.println("------u="+u);;
        }).exceptionally(f->{//t是正常结果 u和f都是异常信息
            System.out.println("------exception:"+f.getMessage());
            return 444;
        }).get();
    }
}

总结

OK，今天关于 JUC的知识分享 就到这里，希望本篇文章能够帮助到大家，同时也希望大家看后能学有所获！！！

后序博主忙完找暑假实习，将对JUC并发进行深入的学习和分享，欢迎大家继续观看我的JUC并发编程专栏！

《JUC并发编程 - 基础篇》 Callable接口 | 辅助类 | 读写锁 | 阻塞队列 | 线程池 | Stream流 | 分支合并框架（三）