Future 与 ChannelFuture 异步处理探究

Future 是Java原生API中JUC包下提供的接口。JUC是Java用来处理并发内容的集合,简单来说Future就是用来处理在同步执行代码中获取异步执行结果的上层接口,其下有多种实现类用于不同并发场景。
原生Future中的get()方法,会阻塞当前线程,知道Future中执行的异步任务完成后。Netty在原生基础上又增加了监听器(Listener)接口,用来更精确的控制异步任务执行时间。

Future常用实现类

FutureTask

  • 主要继承关系:实现Runnable接口和Future接口
  • 主要功能:可以被看作Runnable对象提交线程池执行,可以看作Future对象获取异步执行结果
简单使用示例
package com.ht.actuatorlearn.curr.future;

import java.util.concurrent.*;

/**
 * FutureTask 类示例
 * 线程池异步执行任务,在同步代码块中阻塞获取异步执行结果
 *
 * @author: lht
 * @date: 2023-03-02
 */
public class FutureTaskTest {
    /**
     * 声明线程池
     */
    private static ExecutorService EXECUTOR = null;

    public static void main(String[] args) {
        EXECUTOR = Executors.newSingleThreadExecutor();
        
        // 创建异步执行Future对象
        FutureTask<String> futureTask = new FutureTask<>(() -> {
            // 注意这里使用的 Callable 类异步执行
            // 休眠5s 代替耗时操作
            TimeUnit.SECONDS.sleep(5);
            return "hello";
        });
        try {
            // 提交执行任务
            EXECUTOR.execute(futureTask);

            // 同步代码块1
            System.out.println("Waiting async Results....");

            // 获取异步执行结果 线程阻塞
            System.out.println("Result is: " + futureTask.get());

            // 同步代码块2
            System.out.println("Completed...");

            EXECUTOR.shutdown();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

ScheduledFuture

  • 主要功能:用于获取在程序中计划定期执行任务的结果
简单使用示例
package com.ht.actuatorlearn.curr.future;

import java.util.concurrent.*;

/**
 * 延迟执行任务,并在接收到特定结果下执行一系列操作(关闭任务)
 *
 * @author: lht
 * @date: 2023-03-03
 */
public class ScheduledFutureTest {

    public static void main(String[] args) {
        // 声明工作线程池
        ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();

        // 创建随机数任务
        Callable<Integer> task = () -> {
            return (int) (Math.random() * 10);
        };

        // 延时任务提交线程池
        ScheduledFuture<Integer> schedule = executorService.schedule(task, 5, TimeUnit.SECONDS);

        try {
            Integer result = schedule.get();
            System.out.println("Task Result: " + result);

            // 关闭线程池
            executorService.shutdown();
        }catch (Exception e) {
            e.printStackTrace();
        }
    }
}

ChannelFuture

Netty中,ChannelFuture类似于Future,表示异步的I/O操作结果。当创建新通道或使用Channel进行发送或接收数据时,这些耗时操作在Netty的底层都是异步进行处理的。如果实际业务逻辑中需要获取上述异步的执行结果,那就需要使用到ChannelFuture.

常用示例
  1. 添加监听器(最常用处理):ChannelFuture基于Future的阻塞模型上又实现了监听器回调模式,可以更精确的获取异步执行结果和操作信息,并且不阻塞主线程
        Object message = new Object();

        // channel 异步发送数据
        ChannelFuture future = ctx.writeAndFlush(message);
        future.addListener((ChannelFutureListener) channelFuture -> {
            if (future.isSuccess()) {
                // 消息发送成功 回调
                
            } else {
                // 消息发送失败 回调

            }
        });

2.同步等待操作完成:使用sync()方法可以时耗时异步动作变为同步动作,同步等待耗时操作完成再执行后续代码

        // 同步推送数据
        ChannelFuture channelFuture = ctx.writeAndFlush(message).sync();
        System.out.println(channelFuture.isSuccess());

3.异步等待: 使用ChannelFuture 的 await() 异步等待I/O动作完成,等待中将阻塞当前线程,只到操作完成或中断

        // 阻塞当前线程 异步等待
        ChannelFuture channelFuture = ctx.writeAndFlush(message).await();
        System.out.println(channelFuture.isSuccess());

实际上:从ChannelFuture的源码中我们可以看到,它其实是Future的拓展,并且不具有返回值的异步调用,同时和一个Channel进行绑定。ChannelPromise接口又在其上继承了ChannelFuture类和Promise类,使之既可以写异步执行结果,又具备监听通道的功能。

最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
平台声明:文章内容(如有图片或视频亦包括在内)由作者上传并发布,文章内容仅代表作者本人观点,简书系信息发布平台,仅提供信息存储服务。

推荐阅读更多精彩内容