java - Netty ChannelFuture通知如何不造成死锁

Question

我读过《Netty Guide》，它对 ChannelFuture 没有太多解释。我很困惑为什么它不会导致僵局。

1.它教我如何启动这样的服务器。

ServerBootstrap sb = new ServerBootstrap();
        sb.group(bossGroup, workerGroup);
        sb.channel(NioServerSocketChannel.class);
        sb.localAddress(new InetSocketAddress(port));
        sb.childHandler(new ChannelInitializer<SocketChannel>() {

                @Override
                public void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new BigGameMessageDecoder(msgRegistry));
                        ch.pipeline().addLast(BIG_MSG_ENCODER);
                    if (isDebug) {
                        ch.pipeline().addLast(MSG_LOGGER);
                    }

                    ch.pipeline().addLast(new GameMessageHandler<>(msgRegistry,
                            sessionFactory.createGameSession(), event, false));
                } 

        });

        ChannelFuture cf = sb.bind().sync();
        logger.error("start server at port: {}", port);
        if (sync) {
            cf.channel().closeFuture().sync();
        }

行内:

ChannelFuture cf = sb.bind().sync();

sb.bind() 返回一个 ChannelFuture 并且 sync() 将等待，直到这个 future isDone。

我阅读了 DefaultChannelGroupFuture 代码，它确实向我展示了 sync() 调用 await() 。 而await()会锁定自己，等待其他人的通知。

在 ChannelFuture 的函数 setSuccess 中，它会尝试再次获取该锁。所以我的问题是 sync() 是否先获取锁然后等待，然后 ChannelFuture 尝试通知但它无法获取锁。会造成死锁吗？

如果没有，ChannelFuture 如何通知其他监听器？

其他书籍告诉我不要在 ChannelHandler 中使用 sync() 或 await() 因为它可能会导致死锁。为什么？问题1和问题3有什么区别？

<小时/>

public DefaultChannelGroupFuture sync() throws InterruptedException {
    super.sync();
    return this;
}

  public Promise<V> sync() throws InterruptedException {
    await();
    rethrowIfFailed();
    return this;
}


  public Promise<V> await() throws InterruptedException {
    if (isDone()) {
        return this;
    }

    if (Thread.interrupted()) {
        throw new InterruptedException(toString());
    }

    synchronized (this) {
        while (!isDone()) {
            checkDeadLock();
            incWaiters();
            try {
                wait();
            } finally {`enter code here`
                decWaiters();
            }
        }
    }
    return this;
}

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



  private boolean setSuccess0(V result) {
    if (isDone()) {
        return false;
    }

    synchronized (this) {
        // Allow only once.
        if (isDone()) {
            return false;
        }
        if (result == null) {
            this.result = SUCCESS;
        } else {
            this.result = result;
        }
        if (hasWaiters()) {
            notifyAll();
        }
    }
    return true;
}

Answer 1

在上面的代码中:

public Promise<V> await() throws InterruptedException {
if (isDone()) {
    return this;
}

if (Thread.interrupted()) {
    throw new InterruptedException(toString());
}

synchronized (this) {
    while (!isDone()) {
        checkDeadLock();
        incWaiters();
        try {
            wait();
        } finally {`enter code here`
            decWaiters();
        }
    }
}
return this;

sync方法代码checkDeadLock();会检查当前线程是否是用于处理io事件的内部线程，如果不是，则会发生死锁绑定(bind)操作将分派(dispatch)到等待锁的同一线程。然后，wait(); 将释放 this 的锁并等待某个线程获取锁并通知它。当IO线程调用setSuccess时，它可以获得锁，因为没有人持有锁。