概述

Future代表了异步操作的结果，Netty的Future对JDK的Future进行了扩展。

JDK.Future

仅定义了是否完成以及是否取消的状态判断以及get方法。

public interface Future<V> {boolean cancel(boolean mayInterruptIfRunning);boolean isCancelled();boolean isDone();V get() throws InterruptedException, ExecutionException;V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;
}

Netty.Future

增加了：

结果状态
监听器管理
sync
await


public interface Future<V> extends java.util.concurrent.Future<V> {//statusboolean isSuccess();boolean isCancellable();Throwable cause();//listenerFuture<V> addListener(GenericFutureListener<? extends Future<? super V>> listener);Future<V> addListeners(GenericFutureListener<? extends Future<? super V>>... listeners);Future<V> removeListener(GenericFutureListener<? extends Future<? super V>> listener);Future<V> removeListeners(GenericFutureListener<? extends Future<? super V>>... listeners);//syncFuture<V> sync() throws InterruptedException;Future<V> syncUninterruptibly();//awaitFuture<V> await() throws InterruptedException;Future<V> awaitUninterruptibly();boolean await(long timeout, TimeUnit unit) throws InterruptedException;boolean await(long timeoutMillis) throws InterruptedException;boolean awaitUninterruptibly(long timeout, TimeUnit unit);boolean awaitUninterruptibly(long timeoutMillis);V getNow();@Overrideboolean cancel(boolean mayInterruptIfRunning);
}

ChannelFuture

增加了相关channel信息。

状态信息：

public interface ChannelFuture extends Future<Void> {Channel channel();boolean isVoid();
}

Promise

Promise相对Future增加了设置状态的方法。


public interface Promise<V> extends Future<V> {//设置操作状态方法Promise<V> setSuccess(V result);boolean trySuccess(V result);Promise<V> setFailure(Throwable cause);boolean tryFailure(Throwable cause);boolean setUncancellable();@OverridePromise<V> addListener(GenericFutureListener<? extends Future<? super V>> listener);@OverridePromise<V> addListeners(GenericFutureListener<? extends Future<? super V>>... listeners);@OverridePromise<V> removeListener(GenericFutureListener<? extends Future<? super V>> listener);@OverridePromise<V> removeListeners(GenericFutureListener<? extends Future<? super V>>... listeners);@OverridePromise<V> await() throws InterruptedException;@OverridePromise<V> awaitUninterruptibly();@OverridePromise<V> sync() throws InterruptedException;@OverridePromise<V> syncUninterruptibly();
}

继承关系

源码分析

AbstractFuture<V>

get()方法调用await()方法。

public abstract class AbstractFuture<V> implements Future<V> {@Overridepublic 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);}@Overridepublic 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();}
}

CompleteFuture<V>

代表一个Future已经完成。

内部包含一个EventExecutor，用于注册事件监听器

public abstract class CompleteFuture<V> extends AbstractFuture<V> {private final EventExecutor executor;protected CompleteFuture(EventExecutor executor) {this.executor = executor;}@Overridepublic Future<V> addListener(GenericFutureListener<? extends Future<? super V>> listener) {if (listener == null) {throw new NullPointerException("listener");}DefaultPromise.notifyListener(executor(), this, listener);return this;}@Overridepublic Future<V> addListeners(GenericFutureListener<? extends Future<? super V>>... listeners) {if (listeners == null) {throw new NullPointerException("listeners");}for (GenericFutureListener<? extends Future<? super V>> l: listeners) {if (l == null) {break;}DefaultPromise.notifyListener(executor(), this, l);}return this;}
}

DefaultPromise<V>

设置状态内部使用了一个AtomicReferenceFieldUpdater

private static final AtomicReferenceFieldUpdater<DefaultPromise, Object> RESULT_UPDATER =
            AtomicReferenceFieldUpdater.newUpdater(DefaultPromise.class, Object.class, "result");

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;
    }

Netty--Future和Promise相关推荐

netty中的future和promise源码分析（二）
前面一篇netty中的future和promise源码分析(一)中对future进行了重点分析,接下来讲一讲promise. promise是可写的future,从future的分析中可以发现在其中没 ...
netty channelinactive触发条件_Netty的Future和Promise
在应用中以任务的方式实现功能时,需要考虑到任务间的依赖关系.当任务强依赖上一个任务的结果时,任务间的消息通知就需要重视起来.任务间的消息通知分为同步和异步方式,同步就是消息的发送方要等待消息返回才能继 ...
给Future一个Promise
对java开发者来说,经常需要在一个线程中另起一个线程来异步干其他事,就涉及到熟悉的Thread和Runnable.使用方式如下: System.out.println("Do someth ...
future promise java_第四章 Future和Promise
Netty是一个异步网络处理框架,在实现中大量使用了Future机制,并在Java自带Future的基础上,增加了Promise机制.这两者的目的都是使异步编程更加方便使用.在阅读源码之前,我们需要对 ...
【netty】Netty并发工具-Promise
1.概述转载并且补充:从源码上理解Netty并发工具-Promise 2.前提最近一直在看Netty相关的内容,也在编写一个轻量级的RPC框架来练手,途中发现了Netty的源码有很多亮点,某些实现 ...
Scala教程之:Future和Promise
文章目录定义返回Future的方法阻塞方式获取Future的值非阻塞方式获取Future的值 Future链 flatmap VS map Future.sequence() VS Future ...
C++11多线程之future和promise
std::future和promise的作用是在不同线程之间传递数据.使用指针也可以完成数据的传递,但是指针非常危险,因为互斥量不能阻止指针的访问:而且指针的方式传递的数据是固定的,如果更改数据类型, ...
C++中Future和Promise的一种简单实现
女主宣言由于工作需求,笔者最近在阅读 Pulsar C++ 客户端的实现,发现该客户端虽然是基于 C++11 编写的,但却自己编写了 Future 和 Promise 类,随着阅读的深入,也体会到了 ...
中的listeners_C++中Future和Promise的一种简单实现
女主宣言由于工作需求,笔者最近在阅读 Pulsar C++ 客户端的实现,发现该客户端虽然是基于 C++11 编写的,但却自己编写了 Future 和 Promise 类,随着阅读的深入,也体会到了 ...
C++11多线程---future和promise
future和promise的作用是在不同线程之间传递数据.使用指针也可以完成数据的传递,但是指针非常危险,因为互斥量不能阻止指针的访问:而且指针的方式传递的数据是固定的,如果更改数据类型,那么还需要 ...

Netty--Future和Promise

概述