Java并发编程异步操作Future和FutureTask

码农在囧途

生活是一个洗礼自己的过程,这个洗礼并不是传统意义上的洗礼,传统意义上的洗礼通常认为这个人的思想得到洗礼,灵魂得到洗礼,十分的清新脱俗,不世故,不圆滑,而现实的洗礼实则是让一个人褪去幼稚,褪去无知,让你变得点头哈腰,圆滑世故,我们都是动物,需要物质满足,更需要欲望填补,所以,变成自己小时候唾骂的对象也是可以理解,不过这是一个选择,你可以进行选择,只是在物欲横流的时代,多数人没有这种选择的权力!

Future和FutureTask

Future是一个接口,FutureTask是一个类,实现RunnableFuture接口,RunnableFuture接口继承Future接口。

Future接口的方法

V get() :获取异步执行的结果,如果没有返回结果,此方法会阻塞直到异步计算完成。

V get(Long timeout , TimeUnit unit) :获取异步执行结果,如果没有结果可用,此方法会阻塞,但是会有时间限制,如果阻塞时间超过设定的timeout时间,该方法将抛出异常。

boolean isDone() :如果任务执行结束,无论是正常结束或是中途取消还是发生异常,都返回true。

boolean isCancelled() :如果任务完成前被取消,则返回true。

boolean cancel(boolean mayInterruptRunning) :如果任务还没开始,执行cancel(…)方法将返回false;如果任务已经启动, 执行cancel(true)方法将以中断执行此任务线程的方式来试图停止任务,如果停止成功,返回true;当任务已经启动, 执行cancel(false)方法将不会对正在执行的任务线程产生影响(让线程正常执行到完成),此时返回false;当任务已经完成, 执行cancel(…)方法将返回false。mayInterruptRunning参数表示是否中断执行中的线程。

Future是一个接口,因此我们不能直接创建对象,需要配合线程池一起使用,FutureTask我们可以直接创建对象。

Future的使用

Future代表异步执行的结果,也就是说异步执行完毕后,结果保存在Future里, 我们在使用线程池submit()时需要传入Callable接口,线程池的返回值为一个Future,而Future则保存了执行的结果 ,可通过Futureget()方法取出结果,如果线程池使用的是execute()方法,则传入的是Runnable接口无返回值。

如下我们使用Future模拟下单操作,用户下单后保存订单信息扣减库存增加积分发送短信通知,这么多个任务如果使用同步执行,那么效率就会 比较低,用户体验不好,一般我们会采用消息队列来达到异步的效果,今天我们就不用消息队列,而是使用Future接口来实现异步。

public class FutureTest {
    final static ExecutorService threadPool = Executors.newCachedThreadPool();
​
    //保存订单任务
    public static Future<R> saveOrderTask(OrderInfo orderInfo) {
        return threadPool.submit(new Callable<R>() {
            @Override
            public R call() throws Exception {
                return saveOrder(orderInfo);
            }
        });
    }
​
    //扣减库存任务
    public static Future<R> decreaseStockTask(OrderInfo orderInfo) {
        return threadPool.submit(new Callable<R>() {
            @Override
            public R call() throws Exception {
                return decreaseStockByCommodityId(orderInfo);
            }
        });
    }
​
    //增加积分任务
    public static Future<R> increaseIntegralTask(OrderInfo orderInfo) {
        return threadPool.submit(new Callable<R>() {
            @Override
            public R call() throws Exception {
                return increaseIntegralByUserId(orderInfo);
            }
        });
    }
​
    public static void sendMsgToPhone(OrderInfo orderInfo) {
        threadPool.execute(new Runnable() {
            @Override
            public void run() {
                System.out.println("用户【" + orderInfo.getUserId() + "】,你已下单成功~~~~~~~~");
            }
        });
    }
​
    //增加积分rpc接口
    public static R increaseIntegralByUserId(OrderInfo orderInfo) {
        System.out.println("增加积分~~~~~~~~");
        integralService.increaseIntegralByUserId(orderInfo.getUserId(),20);
        return new R(200, "增加积分成功", null);
    }
​
    //扣减库存rpc接口
    public static R decreaseStockByCommodityId(OrderInfo orderInfo) {
        System.out.println("扣减库存~~~~~~~~");
        stockService.decreaseStockByCommodityId(orderInfo.getCommodityId());
        return new R(200, "扣减库存成功", null);
    }
​
    //保存订单rpc接口
    public static R saveOrder(OrderInfo orderInfo) throws InterruptedException {
        System.out.println("保存订单~~~~~~~~");
        Thread.sleep(2000);
        orderService.insert(orderInfo);
        return new R(200, "保存订单成功", null);
    }
​
​
    public static void main(String[] args) throws ExecutionException, InterruptedException, TimeoutException {
        OrderInfo orderInfo = new OrderInfo().setId("123455").setUserId("111111").setCommodityId("123321");
        Future<R> orderTask = saveOrderTask(orderInfo);
        Future<R> stockTask = decreaseStockTask(orderInfo);
        Future<R> integralTask = increaseIntegralTask(orderInfo);
        sendMsgToPhone(orderInfo);
        if (orderTask.get().getCode() == 200 && orderTask.isDone()) 
            System.out.println(orderTask.get().getMsg());
        if (stockTask.get().getCode() == 200 && stockTask.isDone()) 
            System.out.println(stockTask.get().getMsg());
        if (integralTask.get().getCode() == 200 && integralTask.isDone()) 
            System.out.println(integralTask.get().getMsg());
        threadPool.shutdownNow();
    }
}

输出

保存订单~~~~~~~~
扣减库存~~~~~~~~
增加积分~~~~~~~~
用户【111111】,你已下单成功~~~~~~~~
保存订单成功
扣减库存成功
增加积分成功

我们在保存订单接口模拟处理业务操作,花费了2s,从输出结果可以看出,其他rpc接口并没有在保存订单时而阻塞,而是同时执行,就达到了异步的效果。

不过我们发现了一个问题,那就是异步返回结果被阻塞了,我明明我扣减库存和增加积分接口很快就返回,但是从输出中却发现扣减库存和增加积分在保存 订单后输出,由此我们可看出Future会阻塞返回结果。

上面我们发送短信到用户手机并没有获取返回结果,所以没有使用Future,使用线程池我们就没有使用Callable接口,而是使用Runnable接口, 方法就是execute(),而不是submit()

execute()和submit()区别

1.execute无返回值,这样就无法知道任务是否执行成功,而submit有返回值。 2.execute抛出异常后无法处理,不能捕捉异常,而submit可以捕获异常;

FutureTask的使用

FutureTaskFuture接口的实现类,我们可以直接创建一个FutureTask对象,下面我们对上面的下单流程就行改造,使用FutureTask 来实现。

/**
 * @author 刘牌
 * @date 2022/3/2617:34
 */
public class PlaceOrderFutureTaskTest {
    final static ExecutorService threadPool = Executors.newCachedThreadPool();
​
    //保存订单任务
    public static FutureTask<R> saveOrderTask(OrderInfo orderInfo){
        return new FutureTask<>(new Callable<R>() {
            @Override
            public R call() throws Exception {
                return saveOrder(orderInfo);
            }
        });
    }
​
    //扣减库存任务
    public static FutureTask<R> decreaseStockTask(OrderInfo orderInfo){
        return new FutureTask<>(new Callable<R>() {
            @Override
            public R call() throws Exception {
                return decreaseStockByCommodityId(orderInfo);
            }
        });
    }
​
    //增加积分任务
    public static FutureTask<R> increaseIntegralTask(OrderInfo orderInfo){
        return new FutureTask<>(new Callable<R>() {
            @Override
            public R call() throws Exception {
                return increaseIntegralByUserId(orderInfo);
            }
        });
    }
​
    public static void sendMsgToPhone(OrderInfo orderInfo){
        threadPool.execute(new Runnable() {
            @Override
            public void run() {
                System.out.println("用户【"+orderInfo.getUserId()+"】,你已下单成功~~~~~~~~");
            }
        });
    }
​
    //增加积分rpc接口
    public static R increaseIntegralByUserId(OrderInfo orderInfo){
        System.out.println("增加积分~~~~~~~~");
        integralService.increaseIntegralByUserId(orderInfo.getUserId(),20);
        return new R(200,"增加积分成功",null);
    }
​
    //扣减库存rpc接口
    public static R decreaseStockByCommodityId(OrderInfo orderInfo){
        System.out.println("扣减库存~~~~~~~~");
        stockService.decreaseStockByCommodityId(orderInfo.getCommodityId());
        return new R(200,"扣减库存成功",null);
    }
​
    //保存订单rpc接口
    public static R saveOrder(OrderInfo orderInfo) throws InterruptedException {
        System.out.println("保存订单~~~~~~~~");
        Thread.sleep(2000);
        orderService.insert(orderInfo);
        return new R(200,"保存订单成功",null);
    }
    
    public static void main(String[] args) throws ExecutionException, InterruptedException, TimeoutException {
        OrderInfo orderInfo = new OrderInfo().setId("123455").setUserId("111111").setCommodityId("123321");
        FutureTask<R> orderTask = saveOrderTask(orderInfo);
        FutureTask<R> stockTask = decreaseStockTask(orderInfo);
        FutureTask<R> integralTask = increaseIntegralTask(orderInfo);
        threadPool.submit(orderTask);
        threadPool.submit(stockTask);
        threadPool.submit(integralTask);
        sendMsgToPhone(orderInfo);
        if (orderTask.get().getCode() == 200 && orderTask.isDone()) 
            System.out.println(orderTask.get().getMsg());
        if (stockTask.get().getCode() == 200 && stockTask.isDone()) 
            System.out.println(stockTask.get().getMsg());
        if (integralTask.get().getCode() == 200 && integralTask.isDone()) 
            System.out.println(integralTask.get().getMsg());
        threadPool.shutdownNow();
    }
}

输出

保存订单~~~~~~~~
扣减库存~~~~~~~~
增加积分~~~~~~~~
用户【111111】,你已下单成功~~~~~~~~
保存订单成功
扣减库存成功
增加积分成功

额~~~,从代码中我们看出其实没啥区别,就是一个接口和实现类的不同写法而已,从输入也可以看出和上面的Future一样,由此可知FutureTask获取结果也是 阻塞的。

总结

从上面的流程中可以看出,FutureFutureTask能够实现异步,但是获取结果却是同步的,这缺陷也是显而易见,如果遇到耗时的任务,那么获取返回值的时候 其他任务就会被阻塞,只能排队慢慢来,在高并发的场景下不适合,那有没有解决方案呢,肯定有的,那就是CompletableFuture,我们后面继续介绍,本章我们 就不对其进行介绍。

今天的分享就到这里,感谢你的观看,我们下期见。