【并发编程十：CompletableFuture的应用】

快来打我* 2023-10-09 14:25 59阅读 0赞

上一篇  
[【并发编程九：线程安全问题分析及锁的介绍（2）synchronized】][2_synchronized]

## 一、CompletableFuture ##

CompletableFuture实现了CompletionStage接口和Future接口，前者是对后者的一个扩展，增加了异步回调、流式处理、多个Future组合处理的能力，使Java在处理多任务的协同工作时更加顺畅便利。

*  1、supplyAsync方法：供给一个对象
 *  2、thenAccept方法：然后同步消费掉这个对象
 *  3、runAfterBoth方法：和执行另一个消费方法，
 *      执行完毕后都去执行一个Runnable
 *  4、whenComplete方法：执行结束后的消费方法
 *      调用此方法的线程会阻塞到有返回值为止。
 *  5、allOf方法：
 *  6、anyOf方法：

/**
     * CompletableFuture的应用
     * 1、supplyAsync方法：供给一个对象
     * 2、thenAccept方法：然后同步消费掉这个对象
     * 3、runAfterBoth方法：和执行另一个消费方法，
     * 		执行完毕后都去执行一个Runnable
     * 4、whenComplete方法：执行结束后的消费方法
     * 		调用此方法的线程会阻塞到有返回值为止。
     * ---------------本类到此为止---------------------------
     * 5、allOf方法：
     * 6、anyOf方法：
     * @author Peter
     */
    public class CompletableFutureTest02 {
        
    	
    	/**
    	 * 消耗3秒的方法
    	 * @param str
    	 */
    	private  static void oneStrPrintln(String str){
        
    		String threadName = Thread.currentThread().getName() + "：";
    		System.out.println(threadName+"开始消费字符串："+str+"……");
    		try {
        
    			Thread.sleep(3_000);
    		} catch (InterruptedException e) {
        
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    		}
    		System.out.println(threadName+"消费字符串："+str+"完毕！");
    	}
    	
    	
    	/**
    	 * 消耗1秒的方法
    	 * @param str
    	 */
    	private  static void twoStrPrintln(String str){
        
    		System.out.println("开始消费字符串："+str+"……");
    		try {
        
    			Thread.sleep(1_000);
    		} catch (InterruptedException e) {
        
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    		}
    		System.out.println("消费字符串："+str+"完毕！");
    	}
    
    	public static void main(String[] args) throws Exception {
        
    		ExecutorService threadPool = Executors.newCachedThreadPool();
    		//runAfterBoth、runAfterBothAsync
    		CompletableFuture.supplyAsync(() ->"我是第一个被供给的字符串！")
    			.thenAccept(CompletableFutureTest02::oneStrPrintln)
    			.runAfterEither(
    				CompletableFuture.supplyAsync(() -> 
    					Thread.currentThread().getName() + "：我是第二个被供给的字符串！")
    				.thenAccept(CompletableFutureTest02::twoStrPrintln), 
    				() ->{
        
    				System.out.println(Thread.currentThread().getName() +
    						"：这句话是在runAfterBoth方法的"
    						+ "第一个参数逻辑执行完之后输出的");
    			}/*,threadPool*/)/*.whenComplete((result,throwable) ->{
    				System.out.println("返回值是："+result);
    				System.out.println("抛出的东西是："+throwable);
    			})*/;
    		
    		//来一个死循环等待的特殊写法
    		Thread.currentThread().join();
    	}
    
    }

### 二、CompletableFuture的应用 ###

/**
     * CompletableFuture的应用
     *  1、whenCompleteAsync方法：
     * 		调用此方法的返回的Future的get方法的线程不会阻塞，
     * 		而是直接返回结果。
     * whenComplete 和 whenCompleteAsync 的区别：
     * whenComplete：
     * 	是执行当前任务的线程执行继续执行 whenComplete 的任务。
     * whenCompleteAsync：
     * 	是执行把 whenCompleteAsync 这个任务继续提交给线程池来进行执行。
     * 
     * 2、thenApply方法：处理之后再转换成其它类型的一个对象不处理异常
     * 3、handle方法：处理之后再转换成其它类型的一个对象会处理异常
     * 4、exceptionally方法：处理中出现异常后对其进行处理，入参是
     * 		一个异常的父类
     * @author Peter
     */
    public class CompletableFutureTest04 {
        
    	
    	/**
    	 * 消耗3秒的方法
    	 * @param str
    	 */
    	private  static void oneStrPrintln(String str){
        
    		System.out.println("开始消费字符串："+str+"……");
    		try {
        
    			Thread.sleep(3_000);
    		} catch (InterruptedException e) {
        
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    		}
    		System.out.println("消费字符串："+str+"完毕！");
    	}
    	
    	/**
    	 * 消耗1秒的方法
    	 * @param str
    	 */
    	private  static void twoStrPrintln(String str){
        
    		System.out.println("开始消费字符串："+str+"……");
    		try {
        
    			Thread.sleep(1_000);
    		} catch (InterruptedException e) {
        
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    		}
    		System.out.println("消费字符串："+str+"完毕！");
    	}
    	
    	
    	/**
    	 * 消耗1秒的供给方法
    	 * @param str
    	 */
    	private  static String oneStrSupply(){
        
    		String threadName = Thread.currentThread().getName() + "：";
    		System.out.println(threadName+"开始供给……");
    		try {
        
    			Thread.sleep(1_000);
    		} catch (InterruptedException e) {
        
    			e.printStackTrace();
    		}
    		String str = "我是一个花费1秒才能供给的字符串";
    		System.out.println(threadName+"供给字符串："+str+"----完毕！");
    		return str;
    	}
    	
    	/**
    	 * 消耗5秒的供给方法
    	 * @param str
    	 */
    	private  static String twoStrSupply(){
        
    		System.out.println("开始供给……");
    		try {
        
    			Thread.sleep(5_000);
    		} catch (InterruptedException e) {
        
    			e.printStackTrace();
    		}
    		String str = "我是一个花费5秒才能供给的字符串";
    		System.out.println("供给字符串："+str+"-----完毕！");
    		return str;
    	}
    
    	public static void main(String[] args) throws Exception {
        
    		
    		ExecutorService threadPool = Executors.newCachedThreadPool();
    		
    		CompletableFuture<String> tmpFuture 
    		
    				= CompletableFuture.supplyAsync(CompletableFutureTest04::oneStrSupply,threadPool);
    		
    		//whenCompleteAsync、whenComplete
    		tmpFuture.whenCompleteAsync((result,throwable) ->{
        
    			String threadName = Thread.currentThread().getName() + "：";
    				//假设这里需要3秒
    				try {
        
    					Thread.sleep(3_000);
    				} catch (Exception e) {
        
    					e.printStackTrace();
    				}
    				System.out.println(threadName+"返回值是："+result);
    				System.out.println(threadName+"抛出的东西是："+throwable);
    			});
    		
    		
    //		System.out.println("最终get到的结果是："+tmpFuture.get());
    		System.out.println("输出这句话表示没有被阻塞……");
    		
    		
    		
    //		CompletableFuture<Integer> thenApplyFuture = 
    //					CompletableFuture.supplyAsync(CompletableFutureTest04::oneStrSupply)
    //					.thenApply(String::length);
    //		
    //		System.out.println("最终get到的结果是："+thenApplyFuture.get());
    		
    		
    //		CompletableFuture<Integer> thenApplyFuture = 
    //				CompletableFuture.supplyAsync(CompletableFutureTest04::oneStrSupply)
    //				.handle((result,throwable) ->result.length());
    //		
    //		
    //		
    //		
    //	
    //	System.out.println("最终get到的结果是："+thenApplyFuture.get());
    		
    
    		
    		//来一个死循环等待的特殊写法
    		Thread.currentThread().join();
    	}
    
    }

/**
     * CompletableFuture的应用
     * 1、whenCompleteAsync方法：调用此方法的线程不会阻塞，
     * 		而是直接返回结果。
     * 2、thenApply方法：处理之后再转换成其它类型的一个对象不处理异常
     * 3、handle方法：处理之后再转换成其它类型的一个对象会处理异常
     * 4、exceptionally方法：处理中出现异常后对其进行处理，入参是
     * 		一个异常的父类
     * @author Peter
     */
    public class CompletableFutureTest05 {
        
    	
    	/**
    	 * 消耗3秒的消费方法
    	 * @param str
    	 */
    	private  static void oneStrPrintln(String str){
        
    		System.out.println("开始消费字符串："+str+"……");
    		try {
        
    			Thread.sleep(3_000);
    			throw new RuntimeException("xixixixi");
    		} catch (InterruptedException e) {
        
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    		}
    		System.out.println("消费字符串："+str+"完毕！");
    	}
    	
    	/**
    	 * 消耗1秒的消费方法
    	 * @param str
    	 */
    	private  static void twoStrPrintln(String str){
        
    		System.out.println("开始消费字符串："+str+"……");
    		try {
        
    			Thread.sleep(1_000);
    		} catch (InterruptedException e) {
        
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    		}
    		System.out.println("消费字符串："+str+"完毕！");
    	}
    	
    	
    	/**
    	 * 消耗1秒的供给方法
    	 * @param str
    	 */
    	private  static String oneStrSupply(){
        
    		System.out.println("开始供给……");
    		try {
        
    			Thread.sleep(1_000);
    		} catch (InterruptedException e) {
        
    			e.printStackTrace();
    		}
    		String str = "我是一个花费1秒才能供给的字符串";
    		System.out.println("供给字符串："+str+"----完毕！");
    		return str;
    	}
    	
    	/**
    	 * 消耗3秒的供给方法
    	 * @param str
    	 */
    	private  static String twoStrSupply(){
        
    		System.out.println("开始供给……");
    		try {
        
    			Thread.sleep(3_000);
    		} catch (InterruptedException e) {
        
    			e.printStackTrace();
    		}
    		String str = "我是一个花费3秒才能供给的字符串";
    		System.out.println("供给字符串："+str+"-----完毕！");
    		return str;
    	}
    	
    	/**
    	 * 立即供给的方法
    	 * @param str
    	 */
    	private  static String nowStrSupply(){
        
    		System.out.println("开始供给……");
    		
    		String str = "我是一个立即供给的字符串";
    	
    		System.out.println("供给字符串："+str+"-----完毕！");
    		return str;
    	}
    	
    
    	public static void main(String[] args) throws Exception {
        
    		
    //		//先立即提供一个字符串
    //		CompletableFuture.supplyAsync(CompletableFutureTest05::nowStrSupply)
    //			//同步调用消费方法
    			.thenAccept(CompletableFutureTest05::oneStrPrintln);
    //			//异步调用消费方法
    //			.thenAcceptAsync(CompletableFutureTest05::oneStrPrintln);
    		
    		
    //		System.out.println("主线程在调用方法后的打印！");
    		
    		
    		//先立即提供一个字符串
    		CompletableFuture<String> supplyAsyncFuture 
    				= CompletableFuture.supplyAsync(CompletableFutureTest05::nowStrSupply);
    
    		
    		//同步地调用需要消耗3秒的消费方法消费掉这个字符串
    		supplyAsyncFuture.thenAccept(CompletableFutureTest05::oneStrPrintln)
    		.exceptionally(throwable -> {
        
    			System.out.println("得到的异常信息是："+throwable.getMessage());
    			return null;
    		});
    		
    		
    		
    		//来一个死循环等待的特殊写法
    		Thread.currentThread().join();
    		
    		
    		
    		
    	}
    
    }

三、CompletableFuture的应用

/**
     * CompletableFuture的应用
     * 1、thenAcceptBoth方法：组合，将两个逻辑组合执行，并行执行。
     * 2、acceptEither方法：提供两个逻辑供给（供给对象类型必须相同），
     * 		 然后取其中执行快的去消费。
     * 	总结：acceptXXX都会去接收上一步执行逻辑的返回值
     * 3、runAfterBoth方法：组合，将两个逻辑组合执行，并行执行，
     * 		两个逻辑都执行完毕再执行第二个参数的逻辑（Runable）但是不接收上一步结果。
     * 4、runAfterEither方法：任意一个执行完了就开始执行第二个参数的逻辑（Runable）
     * 		但是不接收上一步结果。
     * 5、thenCombine同thenAcceptBoth方法，
     * 		不同的是可以对两个逻辑的结果还可以进行一个处理逻辑。
     * 6、thenCompose上一个逻辑的出参作为当前逻辑的入参，
     * 		然后再去执行一个消费逻辑。
     * @author Peter
     */
    public class CompletableFutureTest06 {
        
    	
    	/**
    	 * 消耗3秒的方法
    	 * @param str
    	 */
    	private  static void oneStrPrintln(String str){
        
    		System.out.println("开始消费字符串："+str+"……");
    		try {
        
    			Thread.sleep(3_000);
    		} catch (InterruptedException e) {
        
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    		}
    		System.out.println("消费字符串："+str+"完毕！");
    	}
    	
    	/**
    	 * 消耗1秒的方法
    	 * @param str
    	 */
    	private  static void twoStrPrintln(String str){
        
    		System.out.println("开始消费字符串："+str+"……");
    		try {
        
    			Thread.sleep(1_000);
    		} catch (InterruptedException e) {
        
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    		}
    		System.out.println("消费字符串："+str+"完毕！");
    	}
    	
    	
    	/**
    	 * 消耗1秒的供给方法
    	 * @param str
    	 */
    	private  static String oneStrSupply(){
        
    		System.out.println("开始供给……");
    		try {
        
    			Thread.sleep(1_000);
    		} catch (InterruptedException e) {
        
    			e.printStackTrace();
    		}
    		String str = "我是一个花费1秒才能供给的字符串";
    		System.out.println("供给字符串："+str+"----完毕！");
    		return str;
    	}
    	
    	/**
    	 * 消耗5秒的供给方法
    	 * @param str
    	 */
    	private  static String twoStrSupply(){
        
    		System.out.println("开始供给……");
    		try {
        
    			Thread.sleep(5_000);
    		} catch (InterruptedException e) {
        
    			e.printStackTrace();
    		}
    		String str = "我是一个花费5秒才能供给的字符串";
    		System.out.println("供给字符串："+str+"-----完毕！");
    		return str;
    	}
    
    	public static void main(String[] args) throws Exception {
        
    		
    //		CompletableFuture.supplyAsync(null).thenCompose(fn)
    
    		
    		
    		//来一个死循环等待的特殊写法
    		Thread.currentThread().join();
    	}
    
    }

[2_synchronized]: https://blog.csdn.net/weixin_43333483/article/details/124558495?spm=1001.2014.3001.5501