Future异步回调详解

青旅半醒 2022-11-21 11:40 218阅读 0赞

### 文章目录 ###

*   *  案例
     *  join实现
     *  FutureTask
     *  Guava 的异步回调
     *  扩展

## 案例 ##

在深入理解 异步回调模式前，我们以一个经典案例来说明，即  
数学家华罗庚先生的文章《统筹方法》，介绍了一个烧水泡茶的例子，文中提到最优的工序应该是下面这样

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNjUxOTA0_size_16_color_FFFFFF_t_70_pic_center]

用两个线程 T1 和 T2 来完成烧水泡茶程序，T1 负责洗水壶、烧开水、泡茶这三道工序，T2 负责洗茶壶、洗茶杯、拿茶叶三道工序，其中 T1 在执行泡茶这道工序时需要等待 T2 完成拿茶叶的工序

这里实现方式有多中，我们先一最简单的线程汇总的join来实现

## join实现 ##

在编写程序前我们先要了解一下join的基础知识

*  A线程调用B线程的join方法，在B线程没有执行完成钱，A线程一直处于阻塞状态
 *  join是实例方法，需要用线程对象去调用
 *  使用join线程合并线程无法获取到合并线程的返回值，即无法知道烧水线程执行的结果。只能一直阻塞等待烧水线程结束

public class JoinDemo { 
        public static final int SLEEP_TIME = 1000;
    
        public static void main(String[] args) { 
          Thread hThread = new Thread(() -> { 
                try { 
                    Thread.currentThread().setName("烧水线程");
                    System.out.println("洗好水壶");
                    System.out.println("灌好凉水");
                    System.out.println("放在火上");
                    Thread.sleep(SLEEP_TIME);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
                System.out.println("水烧开了");
            });
          hThread.start();
    
            Thread wThread = new Thread(() -> { 
                try { 
                    Thread.currentThread().setName("清洗线程");
                    System.out.println("洗茶壶");
                    System.out.println("洗茶杯");
                    System.out.println("拿茶叶");
                    Thread.sleep(SLEEP_TIME);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
                System.out.println("洗茶叶完成");
            });
            wThread.start();
    
            // 主线程 1. 合并烧水线程
            try { 
                hThread.join();
                wThread.join();
                System.out.println("泡泡茶喝");
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
        }
    }

运行结果：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNjUxOTA0_size_16_color_FFFFFF_t_70_pic_center 1]  
结果很完美，可惜我们无法获取到烧水和洗茶叶线程的执行结果，如果要获取到他们的执行结果我们就需要使用jdk1.5版本后提供的新的多线程类：  
FutureTask

## FutureTask ##

jdk为了弥补Runnable接口没有返回值的问题，jdk提供了Callable接口  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNjUxOTA0_size_16_color_FFFFFF_t_70_pic_center 2]  
Callable接口是一个泛型接口，也是函数式接口，异常为Exception，不用强制捕获，Callable接口和Runnable接口相比功能更强大，但是有个问题就是Callable接口实例不能作为Thread线程实例的target来使用，而Runnable接口可以直接作为Thread的构造参数传入开启线程。  
而且Java线程类中只有Thread，为了解决这个问题，就有了我们今天的FutureTask类  
FutureTask 的构造方法

public FutureTask(Callable<V> callable) { 
            if (callable == null)
                throw new NullPointerException();
            this.callable = callable;
            this.state = NEW;       // ensure visibility of callable
        }

FutureTask 接口主要的功能有如下三个：

1.  判断并发任务是否执行完成
2.  获取并发任务完成后的结果
3.  取消并发执行中的任务  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNjUxOTA0_size_16_color_FFFFFF_t_70_pic_center 3]

// 获取任务执行结果，如果任务没有执行完成，调用次方法线程会一直阻塞，知道任务完成
    V get()
    // 和上面一样功能，不同的是多了阻塞时间限制，在限制时间内没有完成则抛出异常
    public V get(long timeout, TimeUnit unit)
    // 获取任务执行状态，执行完成返回true
    public boolean isDone()
    // 获取任务的取消状态，如果任务完成前被取消，则返回true
    public boolean isCancelled()
    // 取消任务执行
    public boolean cancel(boolean mayInterruptIfRunning)

在介绍完 FutureTask 类后我们来基于FutureTask 实现上面泡茶喝案例

public class FutureTaskDemo { 
    
        public static final int SLEEP_TIME = 1000;
    
        public static void main(String[] args) { 
            Callable<Boolean> hotWaterJob = new HotWaterJob();
            FutureTask<Boolean> hotWaterTask = new FutureTask<>(hotWaterJob);
            Thread hotWaterThread = new Thread(hotWaterTask, "烧水线程");
    
            Callable<Boolean> washJob = new WashJob();
            FutureTask<Boolean> washTask = new FutureTask<>(washJob);
            Thread washThread = new Thread(washTask, "清洗线程");
            hotWaterThread.start();
            washThread.start();
    
            try { 
                Boolean hotWaterFlag = hotWaterTask.get();
                Boolean washFlag = washTask.get();
                drinkTea(hotWaterFlag, washFlag);
            } catch (Exception e) { 
                e.printStackTrace();
            }
        }
    
        private static void drinkTea(Boolean hotWaterFlag, Boolean washFlag) { 
            if (hotWaterFlag && washFlag) { 
                System.out.println("喝茶");
            }
        }
    
    
        static class HotWaterJob implements Callable<Boolean> { 
    
            @Override
            public Boolean call() throws Exception { 
                try { 
                    Thread.currentThread().setName("烧水线程");
                    System.out.println("洗好水壶");
                    System.out.println("灌好凉水");
                    System.out.println("放在火上");
                    Thread.sleep(SLEEP_TIME);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                    return false;
                }
                System.out.println("水烧开了");
                return true;
            }
        }
    
        static class WashJob implements Callable<Boolean> { 
            @Override
            public Boolean call() throws Exception { 
                try { 
                    Thread.currentThread().setName("清洗线程");
                    System.out.println("洗茶壶");
                    System.out.println("洗茶杯");
                    System.out.println("拿茶叶");
                    Thread.sleep(SLEEP_TIME);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                    return false;
                }
                System.out.println("洗茶叶完成");
                return true;
            }
        }
    }

上面 通过 FutureTask.get处于异步阻塞状态，在清洗烧水没有完成前，主线程是做不了任何其他事情的，而异步阻塞效率往往比较低下，只是我们考虑使用异步回调机制，异步回调我们就需要使用一些三方jar了，比如常用的Guava FutureCallback

## Guava 的异步回调 ##

Guava对Java 的异步回调做了增强

1.  引入新的接口ListenableFuture,继承于Future接口，使得Java的Future异步任务再Guava中能被监控和获取非阻塞的异步执行结果
2.  引入新的接口 FutureCallback，更具异步结果完成不同的回调处理
3.  要获取 ListenableFuture实例必须通过线程池（Guava的线程池）来提交Callable任务来获取

我们来基于Guava来实现烧水泡茶喝案例：

public class GuavaFutureDemo { 
        public static final int SLEEP_TIME = 1000;
    
    
        public static void main(String[] args) { 
            MainJob mainJob = new MainJob();
            new Thread(mainJob).start();
            // 烧水
            Callable<Boolean> hotWater = new HotWaterJob();
            Callable<Boolean> washJob = new WashJob();
            ExecutorService pool = Executors.newFixedThreadPool(5);
            // 构造guava线程池
            ListeningExecutorService guavaPool = MoreExecutors.listeningDecorator(pool);
            // 提交烧水任务
            ListenableFuture<Boolean> hotFuture = guavaPool.submit(hotWater);
            // 绑定异步回调，烧水完成后，将烧水任务的 waterFlag 设置为true
           Futures.addCallback(hotFuture, new FutureCallback<Boolean>() { 
               @Override
               public void onSuccess(@Nullable Boolean aBoolean) { 
                   if (aBoolean) { 
                       mainJob.waterFlag = true;
                   }
    
               }
    
               @Override
               public void onFailure(Throwable throwable) { 
                   System.out.println("烧水失败");
               }
           }, guavaPool);
    
           // 提交清洗茶杯任务
            ListenableFuture<Boolean> washFuture = guavaPool.submit(washJob);
            Futures.addCallback(washFuture, new FutureCallback<Boolean>() { 
                @Override
                public void onSuccess(@Nullable Boolean aBoolean) { 
                    if (aBoolean) { 
                        mainJob.washFlag = true;
                    }
    
                }
    
                @Override
                public void onFailure(Throwable throwable) { 
                    System.out.println("清洗茶壶失败");
                }
            }, guavaPool);
    
        }
    
    
    
    
        static class HotWaterJob implements Callable<Boolean> { 
    
            @Override
            public Boolean call() throws Exception { 
                try { 
                    Thread.currentThread().setName("烧水线程");
                    System.out.println("洗好水壶");
                    System.out.println("灌好凉水");
                    System.out.println("放在火上");
                    Thread.sleep(SLEEP_TIME);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                    return false;
                }
                System.out.println("水烧开了");
                return true;
            }
        }
    
        static class WashJob implements Callable<Boolean> { 
            @Override
            public Boolean call() throws Exception { 
                try { 
                    Thread.currentThread().setName("清洗线程");
                    System.out.println("洗茶壶");
                    System.out.println("洗茶杯");
                    System.out.println("拿茶叶");
                    Thread.sleep(SLEEP_TIME);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                    return false;
                }
                System.out.println("洗茶叶完成");
                return true;
            }
        }
    
    
        static class MainJob implements Runnable { 
    
         public static boolean waterFlag = false;
         public static boolean washFlag = false;
    
            @Override
            public void run() { 
                while (!waterFlag && !washFlag) { 
                    try { 
                        System.out.println("等待烧水洗杯中");
                        Thread.sleep(1000);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                    if (waterFlag && washFlag) { 
                        drinkTea(washFlag, waterFlag);
                    }
    
                }
    
            }
    
            private void drinkTea(boolean washFlag, boolean waterFlag) { 
                if (washFlag && waterFlag) { 
                    System.out.println("烧水洗杯完成，开始喝茶");
                } else if(!washFlag){ 
                    System.out.println("杯子清洗失败");
                } else if(!waterFlag){ 
                    System.out.println("烧是失败");
                }
    
            }
        }
    
    }

## 扩展 ##

实际Netty中也扩展了实现了自己的异步回调机制，感兴趣的可以自己去了解

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNjUxOTA0_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/891f9113fcc54a128bd652d67754f9e5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNjUxOTA0_size_16_color_FFFFFF_t_70_pic_center 1]: https://img-blog.csdnimg.cn/20201101144806905.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNjUxOTA0,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNjUxOTA0_size_16_color_FFFFFF_t_70_pic_center 2]: https://img-blog.csdnimg.cn/20201101145051191.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNjUxOTA0,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNjUxOTA0_size_16_color_FFFFFF_t_70_pic_center 3]: https://img-blog.csdnimg.cn/20201101145748396.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNjUxOTA0,size_16,color_FFFFFF,t_70#pic_center