AI 助理
备案控制台
开发者社区 开发与运维 文章 正文

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

2022-12-12 116
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： 《JUC并发编程 - 基础篇》 Callable接口 | 辅助类 | 读写锁 | 阻塞队列 | 线程池 | Stream流 | 分支合并框架（一）

11.7 工作中线程池用哪个？

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

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

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

4f17ca7e97a76878e50f5d3847733a8f.png

自定义线程池代码

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 函数式接口

615206f472e54449792f38d0a1f0f185.png

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

49a730dcad4def09bc1a298c506936d9.png

如何使用

//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：把分拆任务的结果进行合并


8a14a498b619cc91919de10fe42b5c8e.png

8255d2185fa5ce61b231ffc3bf22375d.png

13.2 相关类

ForkJoinPool

6ebb48f15803a70d9f4ce3ad977dfdc4.png


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

ForkJoinTask

398204ff19eb638d4e4d772ff45fbc73.png

ForkJoinTask 类比=> FutureTask

RecursiveTask

a1602d175b961f9271174087803b5ef9.png

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

 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 原理

81194c1978c8c9c5ae6cff003cae9631.png

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并发编程专栏！

文章标签：
Java
存储
关键词：
接口线程
锁线程
类线程
callable接口
阻塞队列线程
爱编程的大李子
目录
相关文章
三三是该溜子
|
2天前
|
Java
【JavaEE】——多线程常用类
Callable的call方法，FutureTask类，ReentrantLock可重入锁和对比，Semaphore信号量（PV操作）CountDownLatch锁存器，
三三是该溜子
10 0
三三是该溜子
|
2天前
|
Java 关系型数据库 MySQL
【JavaEE“多线程进阶”】——各种“锁”大总结
乐/悲观锁，轻/重量级锁，自旋锁，挂起等待锁，普通互斥锁，读写锁，公不公平锁，可不可重入锁，synchronized加锁三阶段过程，锁消除，锁粗化
三三是该溜子
13 0
三三是该溜子
|
2天前
|
Java 程序员 调度
【JavaEE】线程创建和终止，Thread类方法，变量捕获（7000字长文）
创建线程的五种方式，Thread常见方法（守护进程.setDaemon() ，isAlive），start和run方法的区别，如何提前终止一个线程，标志位，isinterrupted，变量捕获
三三是该溜子
19 0
三三是该溜子
|
2天前
|
安全 Java API
【JavaEE】多线程编程引入——认识Thread类
Thread类，Thread中的run方法，在编程中怎么调度多线程
三三是该溜子
16 0
左诗右码
|
1月前
|
供应链 安全 NoSQL
PHP 互斥锁：如何确保代码的线程安全？
在多线程和高并发环境中，确保代码段互斥执行至关重要。本文介绍了 PHP 互斥锁库 `wise-locksmith`，它提供多种锁机制（如文件锁、分布式锁等），有效解决线程安全问题，特别适用于电商平台库存管理等场景。通过 Composer 安装后，开发者可以利用该库确保在高并发下数据的一致性和安全性。
左诗右码
39 6
2G冲浪词条
|
1月前
|
安全 Java
Java多线程集合类
本文介绍了Java中线程安全的问题及解决方案。通过示例代码展示了使用`CopyOnWriteArrayList`、`CopyOnWriteArraySet`和`ConcurrentHashMap`来解决多线程环境下集合操作的线程安全问题。这些类通过不同的机制确保了线程安全，提高了并发性能。
2G冲浪词条
43 1
龙大吉
|
2月前
|
Java 开发者
在Java多线程编程中，创建线程的方法有两种：继承Thread类和实现Runnable接口
【10月更文挑战第20天】在Java多线程编程中，创建线程的方法有两种：继承Thread类和实现Runnable接口。本文揭示了这两种方式的微妙差异和潜在陷阱，帮助你更好地理解和选择适合项目需求的线程创建方式。
龙大吉
33 3
东方睿赢
|
2月前
|
Java
在Java多线程编程中，实现Runnable接口通常优于继承Thread类
【10月更文挑战第20天】在Java多线程编程中，实现Runnable接口通常优于继承Thread类。原因包括：1) Java只支持单继承，实现接口不受此限制；2) Runnable接口便于代码复用和线程池管理；3) 分离任务与线程，提高灵活性。因此，实现Runnable接口是更佳选择。
东方睿赢
57 2
东方睿赢
|
2月前
|
Java
Java中多线程编程的基本概念和创建线程的两种主要方式：继承Thread类和实现Runnable接口
【10月更文挑战第20天】《JAVA多线程深度解析：线程的创建之路》介绍了Java中多线程编程的基本概念和创建线程的两种主要方式：继承Thread类和实现Runnable接口。文章详细讲解了每种方式的实现方法、优缺点及适用场景，帮助读者更好地理解和掌握多线程编程技术，为复杂任务的高效处理奠定基础。
东方睿赢
41 2
龙大吉
|
2月前
|
Java 开发者
Java多线程初学者指南：介绍通过继承Thread类与实现Runnable接口两种方式创建线程的方法及其优缺点
【10月更文挑战第20天】Java多线程初学者指南：介绍通过继承Thread类与实现Runnable接口两种方式创建线程的方法及其优缺点，重点解析为何实现Runnable接口更具灵活性、资源共享及易于管理的优势。
龙大吉
47 1