首页 > 其他分享 >CompletableFuture异步回调

CompletableFuture异步回调

时间：2023-12-01 18:23:24浏览次数：30  
标签：异步 10 System num CompletableFuture println 回调 out

CompletableFuture异步回调

CompletableFuture简介

CompletableFuture被用于异步编程，异步通常意味着非阻塞，可以使得任务单独允许在与主线程分离的其他线程中，并且通过回调可以在主线程中得到异步任务的执行状态，是否完成，和是否异常信息。
CompletableFuture实现了Future，CompletionStage接口，实现了Future接口可以兼容线程池框架，而CompletionStage接口才是异步编程的接口抽象，里面定义多种异步方法，通过这两者集合，从而打造出了强大的CompletableFuture类。

Futrue和CompletableFuture

Future在Java里面，通过用来表示一个异步任务的引用，比如我们将任务提交到线程池里面，然后我么会得到一个Future，在Future里面有isDone方法来判断任务是否处理结束，该有get方法可以一直阻塞直到任务结束然后获取结果，但整体来说这种方式，还是同步的，因为需要客户端不断阻塞等待或者不断轮询才能知道任务是否完成。

Futrue缺点

1.不支持手动完成。2.不支持进一步的非阻塞调用。3.不支持链式调用。4.不支持多个Future合并。5.不支持异步处理。

CompletableFuture类的使用案例

CompletableFuture01
package com.shaonian.juc.completable;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

/**
 * 演示CompletableFuture
 * @author 长名06
 * @version 1.0
 */
public class CompletableFuture01 {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<String> future = new CompletableFuture<>();

        new Thread(() -> {
            System.out.println("子线程开始干活");
            try {
                //子线程沉睡3s
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            //完成future任务
            future.complete("success");
        },"A").start();

        System.out.println("主线程调用get方法获取结果为:" + future.get());
        System.out.println("主线程完成，阻塞结束");
    }
}
CompletableFuture02
package com.shaonian.juc.completable;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

/**
 * @author 长名06
 * @version 1.0
 */
public class CompletableFuture02 {
    public static void main(String[] args) throws ExecutionException, InterruptedException {

        //异步调用，无返回值
        CompletableFuture<Void> completableFuture1 = CompletableFuture.runAsync(() -> {
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread().getName() + "执行runSync()");
        });
        completableFuture1.get();

        //异步调用，有返回值
        CompletableFuture<Integer> completableFuture2 = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + "执行supplyAsync()");
//            int i = 1/0;
            return 1024;
        });
        completableFuture2.whenComplete((t, u) -> {
            System.out.println("----t=" + t);//t参数，是执行的返回值
            System.out.println("----u=" + u);//异常信息
        }).get();
//        System.out.println(Runtime.getRuntime().availableProcessors());

    }
}
CompletableFuture03
package com.shaonian.juc.completable;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

/**
 * 演示线程依赖，执行api thenApply()
 * 一个任务，依赖于另一个任务可以使用thenApply()将两个任务(线程)串行化
 * 对一个数先加10 再平方
 * @author 长名06
 * @version 1.0
 */
public class CompletableFuture03 {

    public static Integer num = 10;

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        System.out.println(Thread.currentThread().getName() + "主线程开始");


        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("加10任务开启");
            num += 10;
            return num;
        }).thenApply(i -> num * num);
        Integer integer = future.get();
        System.out.println("主线程结束，子线程的结果为" + integer);
    }
}
CompletableFuture04
package com.shaonian.juc.completable;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.function.Consumer;
import java.util.function.Function;

/**
 * 消费处理结果
 * thenAccept()方法，接收任务的处理结果，并消费结果，不返回结果了
 * @author 长名06
 * @version 1.0
 */
public class CompletableFuture04 {
    public static Integer num = 10;

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        System.out.println(Thread.currentThread().getName() + "主线程开始");


        CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("加10任务开启");
            num += 10;
            return num;
        }).thenApply(new Function<Integer, Integer>() {
            @Override
            public Integer apply(Integer integer) {
                return num * num;
            }
        }).thenAccept(new Consumer<Integer>() {
            @Override
            public void accept(Integer i) {
                System.out.println("子线程全部处理完成，最后调用了accept方法，消费了结果" + i);
            }
        });
    }
}
CompletableFuture05
package com.shaonian.juc.completable;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.function.Consumer;
import java.util.function.Function;

/**
 * 异常处理
 * @author 长名06
 * @version 1.0
 */
public class CompletableFuture05 {
    public static Integer num = 10;

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        System.out.println(Thread.currentThread().getName() + "主线程开始");


        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
            int i = 1/0;//模拟异常
            System.out.println("加10任务开启");
            num += 10;
            return num;
        }).exceptionally(new Function<Throwable, Integer>() {
            @Override
            public Integer apply(Throwable ex) {
                System.out.println(ex.getMessage());
                return -1;
            }
        });
    }
}
CompletableFuture06
package com.shaonian.juc.completable;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.function.BiFunction;
import java.util.function.Function;

/**
 * 消费结果，同时处理异常
 * handle类似与thenAccept/thenRun方法，是最后一步结果的调用，但是同时可以处理异常
 * @author 长名06
 * @version 1.0
 */
public class CompletableFuture06 {
    public static Integer num = 10;

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        System.out.println(Thread.currentThread().getName() + "主线程开始");


        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
                int i = 1/0;
                System.out.println("加10任务开启");
                num += 10;
            return num;
        }).handle(new BiFunction<Integer, Throwable, Integer>() {
            @Override
            public Integer apply(Integer i, Throwable ex) {
                System.out.println("进入了handle方法");
                if(ex != null){
                    System.out.println("发生了异常,内容为" + ex.getMessage());
                    return -1;
                }else{
                    System.out.println("正常执行,结果为" + i);
                    return i;
                }
            }
        });
    }
}
CompletableFuture07
package com.shaonian.juc.completable;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ExecutionException;
import java.util.function.BiFunction;
import java.util.function.Function;

/**
 * 两个CompletableFuture结果的合并
 * @author 长名06
 * @version 1.0
 */
public class CompletableFuture07 {
    public static Integer num = 10;

    public static void main(String[] args) throws ExecutionException, InterruptedException {

        //有依赖关系的合并
        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("加10任务开启");
            num += 10;
            return num;
        });
        //合并
        CompletableFuture<Integer> future2 = future.thenCompose(new Function<Integer, CompletionStage<Integer>>() {
            @Override
            public CompletionStage<Integer> apply(Integer i) {
                return CompletableFuture.supplyAsync(() -> {
                    return i + 1;
                });
            }
        });
        System.out.println(future.get());
        System.out.println(future2.get());

        //无依赖的任务合并
        CompletableFuture<Integer> job1 = CompletableFuture.supplyAsync(() -> {
            System.out.println("加10任务开启");
            num += 10;
            return num;
        });

        CompletableFuture<Integer> job2 = CompletableFuture.supplyAsync(() -> {
            System.out.println("乘10任务开启");
            num *= 10;
            return num;
        });

        //合并两个结果
        CompletableFuture<Object> future3 = job1.thenCombine(job2, new BiFunction<Integer, Integer, List<Integer>>() {
            @Override
            public List<Integer> apply(Integer result1, Integer result2) {
                ArrayList<Integer> list = new ArrayList<>();
                list.add(result1);
                list.add(result2);
                return list;
            }
        });
        System.out.println("合并结果为" + future3.get());
    }
}
CompletableFuture08
package com.shaonian.juc.completable;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

/**
 * 多个独立任务的合并 allOf
 * @author 长名06
 * @version 1.0
 */
public class CompletableFuture08 {

    public static Integer num = 10;

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        List<CompletableFuture<Integer>> list = new ArrayList<>();

        CompletableFuture<Integer> job1 = CompletableFuture.supplyAsync(() -> {
            System.out.println("加10任务开启");
            num += 10;
            return num;
        });
        list.add(job1);

        CompletableFuture<Integer> job2 = CompletableFuture.supplyAsync(() -> {
            System.out.println("乘10任务开启");
            num *= 10;
            return num;
        });
        list.add(job2);

        CompletableFuture<Integer> job3 = CompletableFuture.supplyAsync(() -> {
            System.out.println("减10任务开启");
            num -= 10;
            return num;
        });
        list.add(job3);

        CompletableFuture<Integer> job4 = CompletableFuture.supplyAsync(() -> {
            System.out.println("除10任务开启");
            num /= 10;
            return num;
        });
        list.add(job4);
        //使用allOf需注意，输入也会执行任务，但是无法获取到结果
        //allOf需要等所有的任务执行完毕
        /**
         * 返回值是CompletableFuture<Void>类型
         *  public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs) {
         *         return andTree(cfs, 0, cfs.length - 1);
         *  }
         */
//        CompletableFuture<Void> allJob = CompletableFuture.allOf(list.toArray(new CompletableFuture[0]));
//        System.out.println(allJob.get());
        //也可以使用 join的形式，执行，可以获取结果
        List<Integer> allResult = list.stream().map(CompletableFuture::join)
                .collect(Collectors.toList());
        System.out.println(allResult);
    }
}
CompletableFuture09
package com.shaonian.juc.completable;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

/**
 * anyOf
 * @author 长名06
 * @version 1.0
 */
public class CompletableFuture09 {
    public static Integer num = 10;

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        List<CompletableFuture<Integer>> list = new ArrayList<>();

        CompletableFuture<Integer> job1 = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("加10任务开启");
            num += 10;
            return num;
        });
        list.add(job1);

        CompletableFuture<Integer> job2 = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(4000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("乘10任务开启");
            num *= 10;
            return num;
        });
        list.add(job2);

        CompletableFuture<Integer> job3 = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("减10任务开启");
            num -= 10;
            return num;
        });
        list.add(job3);

        CompletableFuture<Integer> job4 = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("除10任务开启");
            num /= 10;
            return num;
        });
        list.add(job4);
        //anyOf，这里只要有一个job执行完毕，就结束所有的任务执行，不需要等待所有的job执行完毕
        //但是这个很鸡肋，因为如果不要执行所有的任务，就没必要开启一个CompletableFuture了
        //也可以适用于竞争的场景，先执行成功的获取结果，其他的不再竞争了
        CompletableFuture<Object> allJob = CompletableFuture.anyOf(list.toArray(new CompletableFuture[0]));
        System.out.println(allJob.get());
    }
}

只是为了记录自己的学习历程，且本人水平有限，不对之处，请指正。

标签：异步,10,System,num,CompletableFuture,println,回调,out
From： https://www.cnblogs.com/changming06/p/17870668.html

相关文章

  • 7.2 Windows驱动开发：内核注册并监控对象回调
    在笔者上一篇文章《内核枚举进程与线程ObCall回调》简单介绍了如何枚举系统中已经存在的进程与线程回调，本章LyShark将通过对象回调实现对进程线程的句柄监控，在内核中提供了ObRegisterCallbacks回调，使用这个内核回调函数，可注册一个对象回调，不过目前该函数只能监控进程与线程句柄操......
  • 7.3 Windows驱动开发：内核监视LoadImage映像回调
    在笔者上一篇文章《内核注册并监控对象回调》介绍了如何运用ObRegisterCallbacks注册进程与线程回调，并通过该回调实现了拦截指定进行运行的效果，本章LyShark将带大家继续探索一个新的回调注册函数，PsSetLoadImageNotifyRoutine常用于注册LoadImage映像监视，当有模块被系统加载时则可......
  • 7.6 Windows驱动开发：内核监控FileObject文件回调
    本篇文章与上一篇文章《内核注册并监控对象回调》所使用的方式是一样的都是使用ObRegisterCallbacks注册回调事件，只不过上一篇博文中LyShark将回调结构体OB_OPERATION_REGISTRATION中的ObjectType填充为了PsProcessType和PsThreadType格式从而实现监控进程与线程，本章我们需要将该......
  • 6.8 Windows驱动开发：内核枚举Registry注册表回调
    在笔者上一篇文章《内核枚举LoadImage映像回调》中LyShark教大家实现了枚举系统回调中的LoadImage通知消息，本章将实现对Registry注册表通知消息的枚举，与LoadImage消息不同Registry消息不需要解密只要找到CallbackListHead消息回调链表头并解析为_CM_NOTIFY_ENTRY结构即可实现枚举......
  • Python爬虫异步与缓存技巧浅析
    Python爬虫异步与缓存技巧浅析在Python爬虫中，异步和缓存是两个非常重要的概念。异步可以显著提高爬虫的效率，而缓存则可以帮助我们避免重复抓取网页，节省时间和资源。一、异步爬虫Python的异步爬虫通常使用asyncio库来实现。这个库提供了异步I/O、事件循环、协程和任务等功能。下面是......
  • C++异步编程
    关键词//都会返回一个std::future,用来提供执行结果std::async、std::paclaged_task、std::promise知识点1、std::future<>():a).std::future的临时对象在调用析构函数时会阻塞当前线程。注意在局部区域内的std::future的析构。b).std::future::get()和std::future::wait()......
  • 优化系统性能：同步与异步操作的巧妙平衡
     在今天的数字化环境中，优化系统性能是任何技术团队不可忽视的重要任务。在这一过程中，合理地利用同步和异步操作扮演着至关重要的角色，直接影响着系统的响应速度、资源利用率以及用户体验。同步操作：简单直观但潜藏风险同步操作按照顺序执行，其优点在于逻辑清晰、易于理解......
  • 如何编写优雅的异步代码 — CompletableFuture
    如何编写优雅的异步代码—CompletableFuture Java实现异步编程的8种方式  ......
  • Django中实现事务的几种方式、事物的回滚和保存点、事务提交后，执行某个回调函数、Djan
    Django中实现事务的几种方式#https://zhuanlan.zhihu.com/p/622987268Django是支持事务操作的，它的默认事务行为是自动提交，具体表现形式为：每次数据库操作(比如调用save()方法)会立即被提交到数据库中。但是如果你希望把连续的SQL操作包裹在一个事务里，就需要手动开启事务#......
  • NET 6 WebAPI 如何接受微信支付回调
    [HttpPost("WxNotify")]publicasyncTaskWxNotify(){HttpContextAccessorcontext=new();varrequest=context.HttpContext?.Request;if(request!=null){varbuffer=newMemoryStream();awaitrequest.Body.Co......

赞助商

阅读排行

网站主页关于我们联系我们网站地图

本网站内容转载自其他媒体，侵权联系[admin##ips99.com]。

Copyright © 2020-2023 IPS99 版权所有 IPS99