netty的Promise源码分析

发表于 2021-02-03 分类于 netty Waline：阅读次数：

netty 使用 DefaultPromise 完成异步操作，它对 jdk 的 Future 进行了扩展，提供了更丰富的功能。

DefaultPromise 的结构

DefaultPromise 的类继承结构如下图所示

最顶层的接口是 jdk 的 Future，它用于代表异步操作的结果，netty 对其进行了扩展

io.netty.util.concurrent.Future

io.netty.util.concurrent.Future 对 jdk 的 Future 进行了增强，它增加了如下方法

主要可以分为以下几类：

获取执行的结果状态。是否执行成功、是否取消、获取异常信息
添加、移除 Listener，用于监听执行成功事件并做响应处理
同步等待执行完成
getNow 获取执行结果，非阻塞操作，若无结果则返回 null

io.netty.util.concurrent.AbstractFuture

AbstractFuture 是 Future 的一个抽象实现类，它提供了 get 方法的实现

public V get() throws InterruptedException, ExecutionException {
    // 等待执行完成
    await();

    // 获取异常信息
    Throwable cause = cause();
    if (cause == null) {
        // 无异常，直接获取结果
        return getNow();
    }
  
    // 返回异常信息
    if (cause instanceof CancellationException) {
        throw (CancellationException) cause;
    }
    throw new ExecutionException(cause);
}

public V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
    // 带超时的等待
    if (await(timeout, unit)) {
        Throwable cause = cause();
        if (cause == null) {
            return getNow();
        }
        if (cause instanceof CancellationException) {
            throw (CancellationException) cause;
        }
        throw new ExecutionException(cause);
    }
    // 超时异常
    throw new TimeoutException();
}

io.netty.util.concurrent.Promise

Promise 继承了 Future，它用于表示支持写操作的 Future

它相对于 Future 主要增加了设置执行结果的方法

设置执行成功的结果。setSuccess、trySuccess
设置执行失败的结果。setFailure、tryFailure
标记 Future 不能取消。setUncancellable

io.netty.util.concurrent.DefaultPromise

DefaultPromise 是 Promise 的一个默认实现类，它的主要属性如下

// 执行的结果，使用volatile修饰，保证可见性
private volatile Object result;
// 当Promise执行完成时需要通知Listener，此时就使用这个executor
private final EventExecutor executor;
// 要通知的listener，因为它可能是不同的类型，所以定义为Object类型，使用时再判断类型
private Object listeners;
// 要使用wait()/notifyAll()机制，这个变量记录了waiter的数量
private short waiters;
// 是否正在通知listener
private boolean notifyingListeners;

从属性中可以看出，DefaultPromise 使用了 Object 的 wait/notiry 机制

DefaultPromise 分析

接下来我们来看下它是怎么工作的，如下是一段示例代码

// 创建一个 Promise
Promise<?> future = new DefaultPromise(GlobalEventExecutor.INSTANCE);

// 添加一个listener
future.addListener(new GenericFutureListener() {
    @Override
    public void operationComplete(Future future) throws Exception {
        // 当future代表的异步操作执行完成后，输出日志
        System.out.println("operationComplete");
    }
});

// 新启动一个线程，用于完成future的异步执行
new Thread(new Runnable() {
    @Override
    public void run() {
        // 输出日志
        System.out.println("before thread start");
      
        // 设置future为执行成功
        future.trySuccess(null);
      
        // 输出日志
        System.out.println("after thread start");
    }
}).start();

这里创建了一个 DefaultPromise 示例，并且在一个异步线程中设置执行成功。此外往 DefaultPromise 中添加了一个 listener，当其执行完之后输出一行日志。如下是执行的结果

1
2
3

before thread start
after thread start
operationComplete

可以看到，当异步线程执行为 future 设置完执行结果之后，才触发了 listener 的执行

调用链路

当执行完 future.trySuccess(null) 这行代码时，最终会往 GlobalEventExecutor.INSTANCE 中添加一个任务，它的调用栈如下

// 1. DefaultPromise#trySuccess
public boolean trySuccess(V result) {
    return setSuccess0(result);
}

// 2. DefaultPromise#setSuccess0
private boolean setSuccess0(V result) {
    return setValue0(result == null ? SUCCESS : result);
}

// 3. DefaultPromise#setValue0
private boolean setValue0(Object objResult) {
    if (RESULT_UPDATER.compareAndSet(this, null, objResult) ||
        RESULT_UPDATER.compareAndSet(this, UNCANCELLABLE, objResult)) {
        if (checkNotifyWaiters()) {
            // 这里
            notifyListeners();
        }
        return true;
    }
    return false;
}

// 4.DefaultPromise#notifyListeners
private void notifyListeners() {
    EventExecutor executor = executor();
    if (executor.inEventLoop()) {
        final InternalThreadLocalMap threadLocals = InternalThreadLocalMap.get();
        final int stackDepth = threadLocals.futureListenerStackDepth();
        if (stackDepth < MAX_LISTENER_STACK_DEPTH) {
            threadLocals.setFutureListenerStackDepth(stackDepth + 1);
            try {
                notifyListenersNow();
            } finally {
                threadLocals.setFutureListenerStackDepth(stackDepth);
            }
            return;
        }
    }

    // 这里
    safeExecute(executor, new Runnable() {
        @Override
        public void run() {
            notifyListenersNow();
        }
    });
}

// 5. DefaultPromise#safeExecute
private static void safeExecute(EventExecutor executor, Runnable task) {
    try {
        executor.execute(task);
    } catch (Throwable t) {
        rejectedExecutionLogger.error("Failed to submit a listener notification task. Event loop shut down?", t);
    }
}

// 6. GlobalEventExecutor#execute
public void execute(Runnable task) {
    if (task == null) {
        throw new NullPointerException("task");
    }

    // 添加任务
    addTask(task);
    if (!inEventLoop()) {
        // 启动线程
        startThread();
    }
}

// 7. GlobalEventExecutor#addTask
private void addTask(Runnable task) {
    if (task == null) {
        throw new NullPointerException("task");
    }
    // 添加到队列
    taskQueue.add(task);
}

注意第 6 步，添加完任务后，因为当前并不是在 eventloop 线程中执行，所以会执行 startThread 启动线程。

此时该线程会从任务队列中获取任务执行，因而会获取到第 4 步添加到 exector 中 Runnable 任务 notifyListenersNow，它的调用栈如下：

notifyListenersNow 最终调用 notifyListener0 通知注册的 listener

// DefaultPromise#notifyListener0
private static void notifyListener0(Future future, GenericFutureListener l) {
    try {
        l.operationComplete(future);
    } catch (Throwable t) {
        if (logger.isWarnEnabled()) {
            logger.warn("An exception was thrown by " + l.getClass().getName() + ".operationComplete()", t);
        }
    }
}

总结

DefaultPromise 通过 Object 的 wait/notify 机制实现线程间的同步，通过 volatile 属性保证线程间的可见性

DefaultPromise 支持注册 listener，当任务执行完成时通知 listener，这采用的是设计模式中的观察者模式。

DefaultPromise 需设置 Executor，当通知 listener 时，不是在主线程中执行，而是在 Executor 中执行