使用Netty的DefaultPromise实现异步转同步遇到的

问题背景

在使用netty进行通信时涉及到了异步转同步的操作，服务端发送一条消息，要等待客户端返回结果才能进行下一步，为了省事直接使用了Netty自带的DefaultPromise来实现Future的相关操作。
经过测试两端可以正常通信，但是在服务端获取Future内部的结果后，紧接着就报了一个空指针异常

代码分析

异步转同步核心代码

public class OperationResultFuture extends DefaultPromise<BalanceMessage> {

}


public class RequestFuturePool {

    private static final Map<String, OperationResultFuture> futureMap = new ConcurrentHashMap<>();

    public static Future<Message> add(String key) {
        OperationResultFuture future = new OperationResultFuture();
        futureMap.put(key, future);
        return future;
    }

    public static void setResult(String key, Message result) {
        try {
            OperationResultFuture future = futureMap.get(key);
            if (future != null) {
                future.setSuccess(result);
            }
        } finally {
            futureMap.remove(key);
        }
    }

}

// 业务代码
 public static Message sendAndGetResponse(String channelId, Message message) {
        // 获取消息的唯一id和Future进行绑定
        String key = balanceMessage.getStreamId();
        channel.writeAndFlush(balanceMessage);
        Future<Message> future = RequestFuturePool.add(key);

        try {
            // 获取结果，超时时间3秒
            return future.get(3, TimeUnit.SECONDS);
        } catch (Exception e) {
            log.error("获取结果异常", e);
        }
 }

在上述代码中，获取结果可以正常获取，获取完成之后就抛出异常，于是进入DefaultPromise查看源代码
netty版本: netty-all-4.1.33.Final

源代码

// set结果的源代码
public Promise<V> setSuccess(V result) {
        if (this.setSuccess0(result)) {
            this.notifyListeners();
            return this;
        } else {
            throw new IllegalStateException("complete already: " + this);
        }
    }

// 异常代码在notifyListeners()方法中
private void notifyListeners() {
        // 1. 获取EventExecutor
        EventExecutor executor = this.executor();
        if (executor.inEventLoop()) {
            InternalThreadLocalMap threadLocals = InternalThreadLocalMap.get();
            int stackDepth = threadLocals.futureListenerStackDepth();
            if (stackDepth < MAX_LISTENER_STACK_DEPTH) {
                threadLocals.setFutureListenerStackDepth(stackDepth + 1);

                try {
                    this.notifyListenersNow();
                } finally {
                    threadLocals.setFutureListenerStackDepth(stackDepth);
                }

                return;
            }
        }

        safeExecute(executor, new 1(this));
}

// 在上面的第一处获取EventExecutor的代码中，executor()方法会返回当前类的EventExecutor
// 而EventExecutor是在构造方法中指定的，如下是构造方法，本项目中用的是默认构造方法，所以
// EventExecutor为空，在调用executor.inEventLoop()就会发生空指针异常
 public DefaultPromise(EventExecutor executor) {
     this.executor = (EventExecutor)ObjectUtil.checkNotNull(executor, "executor");
 }

 protected DefaultPromise() {
     this.executor = null;
 }

通过上面的debug成功找到了bug，于是netty的github上搜索了一下issues，也有人有类似问题。
于是将netty版本升级到了4.1.35.Final, 问题就解决了，看一下4.1.35的源码

 public Promise<V> setSuccess(V result) {
        if (setSuccess0(result)) {
            return this;
        }
        throw new IllegalStateException("complete already: " + this);
 }

 private boolean setSuccess0(V result) {
        return setValue0(result == null ? SUCCESS : 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;
 }

通过上述代码可以看到通知方法notifyListeners()移到了内部，也就是如果还有其他线程在等待结果返回的话就通过EventExecutor通知其他线程，成功避免了空指针的发生。

至此问题解决，记录一波