多线程编程范式(一) 协作范式

简介: 多线程编程范式(一) 协作范式

前言

本来本篇有个前置文章,但是有点卡文,所以本篇缩小了需要的前置内容,阅读本篇需要知道线程、线程池的概念。

Java中任意一段代码在执行的时候都在一个线程当中。

CountDownLatch 示例

假设你需要在某个方法中,后面的操作你委托给了线程池进行处理,但是你希望提交给线程池的任务处理完毕,方法才接着执行,这也就是线程互相等待:

public static void main(String[] args) {
    // 执行main方法的是main线程 
    doSomeThing();
    // 希望将requestThread方法委托给线程池处理
    // doThing方法在requestThread方法执行之后执行
    requestThread();
    doThing();
}

我们就可以这么写:

public class CountDownLatchDemo {
    /**
     * 推荐用ThreadPoolExecutor来创建线程池
     */
  private static final ExecutorService EXECUTOR_SERVICE = Executors.newFixedThreadPool(4);
  public static void main(String[] args) {
        doSomeThing();
        // 希望将requestThread方法委托给线程池处理
        // doThing方法在requestThread方法执行之后执行
       // 构造函数代表只有一个任务需要委托给线程池来完成
        CountDownLatch countDownLatch = new CountDownLatch(1);
        EXECUTOR_SERVICE.submit(()->{
            try {
                requestThread();
            }catch (Exception e){
                // 做日志输出
            }finally {      
                //countDown 方法代表完成一个任务
                countDownLatch.countDown();
            }
        });
        try {
            // 如果requestThread方法没被执行完,count的调用次数不等于我们在构造函数中给定的次数
            // 调用此行代码的线程会陷入阻塞
            countDownLatch.await();
        } catch (InterruptedException e) {
            // 做日志输出
            e.printStackTrace();
        }
        doThing();
  }
}

上面这种是任务完成了接着执行下面的代码,其实我们也可以反着用,假设在某场景下,多个线程需要等待一个线程的结果,类似于赛跑,主线程是给信号的人,其他线程是运动员:

public class CountDownLatchDemo {
    /**
     * 推荐用ThreadPoolExecutor来创建线程池
     */
  private static final ExecutorService EXECUTOR_SERVICE = Executors.newFixedThreadPool(4);
    public static void main(String[] args) {
        //一共四个运动员
        CountDownLatch countDownLatch = new CountDownLatch(1);
        playerWait(countDownLatch);
        System.out.println(Thread.currentThread().getName()+"裁判开枪:");
        countDownLatch.countDown();
        EXECUTOR_SERVICE.shutdown();
    }
    private static void playerWait(CountDownLatch countDownLatch) {
        // 一共四个运动员
        for (int i = 0; i < 4 ; i++) {
            EXECUTOR_SERVICE.submit(()->{
                try {             
                    countDownLatch.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(Thread.currentThread().getName()+"听到了枪声开始跑");
            });
        }
    }

上面两种其实就是CountDownLatch的经典用法:

  • 一个线程等待线程池的任务处理完毕再接着往下执行
  • 线程池的任务等待主线程执行完毕再接着往下执行。

总结一下CountDownLatch,CountDownLatch的构造方法用于指定任务数,调用countDown方法一次代表完成一个任务,如果没有完成所有任务数,调用await方法的线程会陷入阻塞,完成了所有任务数之后,调用await方法陷入阻塞的线程会被唤醒。

一个线程池等待线程池的任务处理完毕再接着往下执行,其实还有另一种简洁的写法,即用CompletabeFuture来写。

CompletableFuture  示例

计算模型

public static void main(String[] args) {
    // doSomeThing  requestThread doThing 都是空方法
    // 不给出具体实现
    doSomeThing(); 
    CompletableFuture.runAsync(()->{  
        requestThread();
    },EXECUTOR_SERVICE).exceptionally(e->{
        System.out.println(e);
        return null;
    }).join();
    doThing();
    EXECUTOR_SERVICE.shutdown();
}

CompletableFuture方法中的runAsync中第一个参数是Runnable,第一个参数是Executor。代表我们指示CompletableFuture用我们提供的线程池执行Runable任务。在runAsync方法中发生异常进入exceptionally中。join方法用于获取runAsnc的计算结果,如果计算未完会陷入等待。get方法和join方法类似都是用于获取最后的计算结果,但是get方法强制要求处理计算过程中的异常,也就是get方法上有检查异常,join上是未检查性异常。

CompletableFuture 提供的线程计算模型是丰富的,粗略的可以这么说,我们有一个任务我们将其分割为若干个阶段,不同的阶段委托给不同的线程进行处理,有点流水线的感觉。假设我们有两个运算,我们将这两个运算给线程池,但是我们希望这两个运算完成触发某个方法或者是做个某操作,或者是回调。计算模型如下图所示:

这正是CompletableFuture的典型应用场景,我们用代码模拟如下:

public class CompletableFutureDemo {
 // 实际项目中推荐用ThreadPoolExecutor创建线程池
    private static final ExecutorService EXECUTOR_SERVICE = Executors.newFixedThreadPool(4);
    
    public static void main(String[] args) {
        CompletableFuture<String> operationTwoTask = CompletableFuture.supplyAsync(() -> operationTwo(),EXECUTOR_SERVICE);
        CompletableFuture<String> operationOneTask = CompletableFuture.supplyAsync(() -> operationOne(),EXECUTOR_SERVICE);
        String finalResult = operationOneTask.thenCombineAsync(operationTwoTask, (o1, o2) -> {
            System.out.println(o2);
            System.out.println(o1);
            StringBuilder stringBuilder = new StringBuilder();
            stringBuilder.append(o1).append(o2);
            System.out.println("合并运算一和运算二的结果:" + stringBuilder.toString());
            return stringBuilder.toString();
        },EXECUTOR_SERVICE).join();
        System.out.println(finalResult);
        EXECUTOR_SERVICE.shutdown();
    }
    private static String operationTwo() {
        System.out.println("运算二完成");
        return "运算二完成";
    }
    private static String  operationOne() {
        try {
            TimeUnit.SECONDS.sleep(4);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("运算一完成");
        return "运算一完成";
    }
}

有同学看到这里可能会问,那如果我只需要处理这两个运算值,只需要在回调里面处理不需要返回,那thenCombineAsync不是满足不了我的需求了吗?对啊,这个不满足你需求,你可以用thenAcceptBoth就行了,CompletableFuture合并结果的方法如下:

public <U,V> CompletableFuture<V> thenCombine(CompletionStage<? extends U> other,
BiFunction<? super T,? super U,? extends V> fn)
public <U,V> CompletableFuture<V> thenCombineAsync(CompletionStage<? extends U> other,
 BiFunction<? super T,? super U,? extends V> fn)
public <U,V> CompletableFuture<V> thenCombineAsync(
 CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn, Executor executor) 
public <U> CompletableFuture<Void> thenAcceptBoth(CompletionStage<? extends U> other,
BiConsumer<? super T, ? super U> action)    
public <U> CompletableFuture<Void> thenAcceptBothAsync(
 CompletionStage<? extends U> other,
BiConsumer<? super T, ? super U> action)    
public <U> CompletableFuture<Void> thenAcceptBothAsync(
CompletionStage<? extends U> other,
BiConsumer<? super T, ? super U> action, Executor executor)

如果不熟悉函数式编程,可能看这些方法签名会有些头疼,但是本篇只要求你懂一点Java中的Lambda函数的基本知识即可。第二个参数用于接两个运算的值,第一个参数给CompletableFuture的supplyAsync的返回就行。BiFunction和BiConsumer这两个函数式接口的区别在于,BitFunction要求给的Lambda函数有返回值,而BitConsumer是无返回值的。第三个参数如果不给的话,如果用的方法是带Async的,则会启用CompletableFuture内置的线程池,不推荐使用CompletableFuture内置的线程池来执行合并行为。不带Async的是用前两个运算中率先完成任务的线程来执行合并运算。

上面我们讲的是计算模型是CompletableFuture支持的一种模型: 二元依赖。事实上CompletableFuture可以支持零元依赖、一元依赖、二元依赖、多元依赖。依赖的意思是某个回调需要满足某些条件才能触发,零元依赖是不需要任何依赖,直接异步执行一个方法,像下面这样:

public static void main(String[] args) {
        // 线程池用的还是上文的线程池,这里不再贴出
        // 创建零元依赖的第一种方式
        // 第一个参数用的是Lambda表达式 传入的方法体就是打印并返回hello world
        CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> {
            System.out.println("hello world");
            return "hello world";
        }, EXECUTOR_SERVICE);
        // 方式二 通过静态方法创建
        CompletableFuture<String> cf2 = CompletableFuture.completedFuture("complete");
        System.out.println(cf2);
        // 方式三 常常被用来将一个普通的回调方法转换为CompletableFuture,纳入到CompletableFuture的编排中
        CompletableFuture cf3 = new CompletableFuture();
        cf3.complete("complete");
        System.out.println(cf3);
}

其实上周这篇文章没发的一个原因就是我不理解CompletableFuture的completedFuture、complete方法,我不理解这两个方法是做啥用的,因为我将其类比到了Stream流的收集操作,以为是最终的任务完成之后会触发这个方法,但这个方法接收的是人一个值,这里我就又想不通了。其实我们可以将complete 方法可以理解为任务的起点。前面的例子一个典型的例子就是我们知道任务的起点,所以根据任务的不同类型选择runAsync(无返回值给下个阶段做运算),supplyAsync(有返回值给下个阶段做运算), 但是有一种计算模型是被我们所忽略的,即外部通知我们开始计算:

private static void noticeCompute() {
   //EXECUTOR_SERVICE 是一个线程池
    CompletableFuture completableFuture = new CompletableFuture();
        completableFuture.thenAcceptAsync(lastResult->{
            System.out.println(lastResult);
    },EXECUTOR_SERVICE);
  // ompletableFuture.complete("hello world");  语句一
}

其实可以直接看出上面的代码什么也不会输出,原因在于thenAcceptAsync并未接收到值,我们解除语句一的注释,就能输出hello world。我将complete方法理解为通知,像是多米诺骨牌的第一张倒下的牌。上面我们介绍的是二元依赖,这是一个比较鲜明的例子,二元依赖的协作任务都能完成,那就意味着能完成一元依赖或者多元依赖运算。所谓一元依赖运算指的是我们有两个执行方法,姑且称之为方法一和方法二,方法一是任务的起点,方法二依赖方法一的结果,更为通俗一点的介绍是方法一的返回结果是方法二的入参,如下面的代码所示:

private static void oneDemo() {
    String methodOneResult = methodOne();
    methodTwo(methodOneResult);
 }

如果想将方法methodOne、methodTwo方法都各自委托给一个线程来完成上面的任务,我们用CompletableFuture可以这么写

private static void oneAsyncDemo() {
        CompletableFuture.supplyAsync(() -> methodOne(),EXECUTOR_SERVICE).thenAcceptAsync(lastResult-> methodTwo(lastResult),EXECUTOR_SERVICE);
    }

有同学看到这里可能会说,thenAcceptAsync接受的lambda表达式没有返回值,那如果我还想用methodTwo方法的返回值怎么办,CompletableFuture也早有准备,可以用thenSupplyAsync接口。其实看到这里我们已经可以大致总结出来CompletableFuture方法的规律了:

  • 带supply方法要求给的lambda表达式有返回值
  • 带accept方法要求给的lambda表达式无返回值
  • 方法名带async的会启用一个线程去执行传入的lambda表达式
  • 合并两个线程的计算结果, 带combine接受的lambda表达式的要求有返回值,带accept接受的lambda表达式不要求有返回值。
  • join 和 get方法都用来获取计算结果,get 方法有检查异常,要求开发者强制处理。
    下面我们来介绍多元依赖运算,所谓多元依赖运算通俗的讲就是我们四个方法: 方法一、方法二、方法三、方法四。方法四需要在方法一、方法二、方法三执行完毕之后才执行,同时根据三个方法的返回值进行运算, 同步的写法如下图所示:
private static void multiDemo() {
        String resultOne = methodOne();
        String resultTwo = methodTwo("");
        String resultThree = methodThree("");
        methodFour(resultOne,resultTwo,resultThree);  
 }

上面的methodFour方法是用三个方法的返回值作为入参,如果将上面四个方法都各自委托给一个线程进行计算,我们可以这么做:

private static void multiAsyncDemo() {
        CompletableFuture<String> oneFuture = CompletableFuture.supplyAsync(() -> methodOne(),EXECUTOR_SERVICE);
        CompletableFuture<String> twoFuture = CompletableFuture.supplyAsync(() -> methodTwo(""),EXECUTOR_SERVICE);
        CompletableFuture<String> threeFuture = CompletableFuture.supplyAsync(() -> methodThree(""),EXECUTOR_SERVICE);
        CompletableFuture<Void> cf = CompletableFuture.allOf(oneFuture, twoFuture, threeFuture);
        cf.thenAcceptAsync(o->{
            // thenAcceptAsync 在oneFuture、twoFuture、threeFuture计算完成之后,才会执行
            // 所以这里可以直接调用join方法拿结果
            String resultOne = oneFuture.join();
            String resultTwo = twoFuture.join();
            String resultThree = threeFuture.join();
            System.out.println(resultOne+resultTwo+resultThree);
        },EXECUTOR_SERVICE);
 }

那有同学看到这里就会问了,那如果我想要methodOne、methodTwo、methodThree任意一个返回结果,我该怎么写,CompletableFuture提供了anyOf,写法示例如下图所示:

private static void multiAsyncAnyDemo() {
        CompletableFuture<String> oneFuture = CompletableFuture.supplyAsync(() -> methodOne(),EXECUTOR_SERVICE);
        CompletableFuture<String> twoFuture = CompletableFuture.supplyAsync(() -> methodTwo(""),EXECUTOR_SERVICE);
        CompletableFuture<String> threeFuture = CompletableFuture.supplyAsync(() -> methodThree(""),EXECUTOR_SERVICE);
        CompletableFuture<Object> cf3 = CompletableFuture.anyOf(oneFuture, twoFuture, threeFuture);
        cf3.thenAcceptAsync(result ->{
            System.out.println("我是最后结果"+result);
        });
 }

其实上面的二元依赖还有一种模型被我们所忽略,如下图所示:

运算三接收运算一、运算二的哪一个结果都行,CompletableFuture的示例如下图所示:

private static void anyTwoResult() {
        CompletableFuture<String> oneFuture = CompletableFuture.supplyAsync(() -> methodOne(),EXECUTOR_SERVICE);
        CompletableFuture<String> twoFuture = CompletableFuture.supplyAsync(() -> methodTwo(""),EXECUTOR_SERVICE);
        oneFuture.acceptEither(twoFuture,anyResult->{
            System.out.println(anyResult);
        });
 }

异常处理

到现在为止我们只是在第一个例子中用CompletableFuture的exceptionally处理过协作过程中的异常,exceptionally的特点是处理过程中有异常就处理,无异常就不执行,那如果我们想让异常处理阶段必不可少呢,CompletableFuture提供了handle方法供我们使用, handle方法事无论异常是否发生都会执行,示例如下:

private static void handleDemo() {
        CompletableFuture.supplyAsync(()->{
            int i = 1 / 0;
            System.out.println("hello world");
            return "hello world";
        }).handle((result, exception)->{
            if (exception != null){
                System.out.println("果然出现了异常");
                return "发生了异常";
            }else {
                return result;
            }
        }).thenAccept(o->{
            System.out.println(o);
        });
}

那如果我并不想返回值呢,handle方法要求提供的lambda方法有返回值,如果不想给返回值可以用whenComplete, 如下图所示:

private static void whenCompleteDemo() {
    CompletableFuture.supplyAsync(()->{
        int i = 1 / 0;
        System.out.println("hello world");
        return "hello world";
    }).whenComplete((result,ex)->{
        if (ex != null){
            ex.printStackTrace();
        }else {
            System.out.println(result);
        }
    });
}

总结一下CompletableFuture

CompletableFuture  为我们提供的计算模型如下

  • 我们自己指定计算起点

  • 由外部选择开始执行计算模型的时机

  • 合并多个计算结果或选择任意一个计算结果

Semaphore 示例

Semaphore 意味信号量,我个人将其理解为通行证,我们可以用这个设置最大访问数,举个现实的例子就是节假日风景区会涌来很多人,但是景区的接待能力是有限的,所以景区会选择通过门票控制人数,门票卖光就需要等待其他人从风景区出来,这也类似于去饭店吃饭,饭店的人满了,就会发号,让人在饭店外等待。示例如下图所示:

private static void demo01() {
    // 一共只有两张票
    Semaphore semaphore = new Semaphore(2);
    for (int i = 0; i < 4 ; i++) {
        EXECUTOR_SERVICE.submit(()->{
            try {
                 // 申请票 申请一次票少一张
                // 如果票数为0,其他线程调用会陷入阻塞。
                semaphore.acquire();
                System.out.println(Thread.currentThread().getName()+"进入饭店吃饭");
                System.out.println(Thread.currentThread().getName()+"走出饭店 腾出一张桌子");
                // 释放通行证,释放通行证的时候会唤醒
                // 因调用acquire方法陷入阻塞的线程
                semaphore.release();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
    }
}

CyclicBarrier 示例

CountDownLatch中有两个示例: 一个是一个等多个,即线程在执行当前方法的时候,为了提升整体的运行速度,选择将一些任务提交给线程池处理,等线程池将任务处理完毕再接着往下执行。第二个则是多等一,多个线程先调用await方法,某个线程处理完成调用countDown。CyclicBarrier 提供的协作模型跟CountDownLatch略有不同,CyclicBarrier 实现的是类似于收集龙珠,每个人负责收集一颗龙珠,收集到龙珠之后,进行等待,所有人收集龙珠完毕,大家就都可以喊出召唤神龙了:

public class CyclicBarrierDemo {
    public static void main(String[] args) {
        CyclicBarrier cyclicBarrier = new CyclicBarrier(7);
        for(int i = 1;i <= 7; i++) {
            int finalI = i;
            EXECUTOR_SERVICE.submit(() -> {
                System.out.println(Thread.currentThread().getName() + "\t 收集到第" + finalI + "颗龙珠");
                try {
                    // 注意线程池的核心线程数要求是7个
                    // 此处需要七次调用。
                    // 召唤神龙才能输出
                    cyclicBarrier.await();
                    System.out.println("****召唤神龙");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (BrokenBarrierException e) {
                    e.printStackTrace();
                }
            });
        }
    }
}

写在最后

我们对比一下CountDownLatch 、Semaphore 、CyclicBarrier 。这三个都可以用来实现线程之间进行互相等待,在上面的示例中我们选择用CountDownLatch用作任务计数器,用构造函数指定总任务数,每当一个线程完成一个任务,我们调用countDown方法,对完成任务减一。如果所有任务没有完成,调用CountDownLatch的await方法的线程会陷入阻塞。而Semaphore 则是对某资源实现控制,我们在构造函数中指定的数字就是允许线程调用acquire方法的次数,调用一次许可证减一,调用次数用进,其他线程无法执行acquire方法之下的代码。而CyclicBarrier与CountDownLatch 类似,但是CyclicBarrier的await方法类似于CountDownLatch的countDown和await方法的结合体,先countDown再调用await,此时任务未完成,陷入阻塞。我们可以用CompletableFuture 来实现CountDownLatch类似的效果。其实本来还打算介绍一下JDK 7的Phaser,但是考虑到一篇里面塞太多东西也不利于吸收,就放到下一篇了。

参考资料

相关文章
|
5月前
|
数据处理
多线程与并发编程【线程对象锁、死锁及解决方案、线程并发协作、生产者与消费者模式】(四)-全面详解(学习总结---从入门到深化)
多线程与并发编程【线程对象锁、死锁及解决方案、线程并发协作、生产者与消费者模式】(四)-全面详解(学习总结---从入门到深化)
61 1
|
2月前
|
Java 开发者
解锁并发编程新姿势!深度揭秘AQS独占锁&ReentrantLock重入锁奥秘,Condition条件变量让你玩转线程协作,秒变并发大神!
【8月更文挑战第4天】AQS是Java并发编程的核心框架,为锁和同步器提供基础结构。ReentrantLock基于AQS实现可重入互斥锁,比`synchronized`更灵活,支持可中断锁获取及超时控制。通过维护计数器实现锁的重入性。Condition接口允许ReentrantLock创建多个条件变量,支持细粒度线程协作,超越了传统`wait`/`notify`机制,助力开发者构建高效可靠的并发应用。
76 0
|
3月前
|
安全 Java API
Java并发编程的艺术:解锁多线程同步与协作的秘密
【7月更文挑战第28天】在Java的世界中,并发编程如同一场精心编排的交响乐,每一个线程都是乐团中的乐手,而同步机制则是那指挥棒,确保旋律的和谐与统一。本文将深入探讨Java并发编程的核心概念,包括线程的创建、同步机制、以及线程间的通信方式,旨在帮助读者解锁Java多线程编程的秘密,提升程序的性能和响应性。
37 3
|
3月前
|
安全 Java 数据处理
Java并发编程:线程同步与协作的深度解析
在探索Java并发编程的海洋中,线程同步与协作的灯塔指引着航向。本文将深入挖掘线程同步机制的核心原理,揭示锁、条件变量等工具如何确保数据的一致性和线程间有序的通信。通过案例分析,我们将解码高效并发模式背后的设计哲学,并探讨现代Java并发库如何简化复杂的同步任务。跟随文章的步伐,您将获得提升多线程应用性能与可靠性的关键技能。 【7月更文挑战第24天】
34 5
|
5月前
|
Java
Java一分钟:线程协作:wait(), notify(), notifyAll()
【5月更文挑战第11天】本文介绍了Java多线程编程中的`wait()`, `notify()`, `notifyAll()`方法,它们用于线程间通信和同步。这些方法在`synchronized`代码块中使用,控制线程执行和资源访问。文章讨论了常见问题,如死锁、未捕获异常、同步使用错误及通知错误,并提供了生产者-消费者模型的示例代码,强调理解并正确使用这些方法对实现线程协作的重要性。
55 3
|
Java 数据处理 调度
Java线程并发协作与任务定时调度
Java线程并发协作与任务定时调度
82 1
|
存储 Java 调度
《Java-SE-第二十四章》之线程间协作
《Java-SE-第二十四章》之线程间协作
|
Go C++
通过哲学家进餐问题学习线程间协作(代码实现以leetcode1226为例)
通过哲学家进餐问题学习线程间协作(代码实现以leetcode1226为例)
113 0
为什么线程协作的 wait() 方法需要写在循环里,你有想过吗?
那么问题是为啥这里是 while 而不是 if 呢?这个问题我最开始也想了很久,按理来说已经在 synchronized 块里面了嘛,就不需要了。这个也是我前面一直是这么认为的,直到最近看了一个 Stackoverflow 上的问题才对这个问题有了比较深入的理解。 试想我们要试想一个有界的队列。那么常见的代码可以是这样:
|
Java 调度
69. 对并发熟悉吗?谈谈线程间的协作(wait/notify/sleep/yield/join)
69. 对并发熟悉吗?谈谈线程间的协作(wait/notify/sleep/yield/join)
53 1
69. 对并发熟悉吗?谈谈线程间的协作(wait/notify/sleep/yield/join)
下一篇
无影云桌面