@Override
protected void doStop() {
    try {
        ChannelGroupFuture f = allChannels.close();
        f.addListener(new ChannelGroupFutureListener() {
            @Override
            public void operationComplete(ChannelGroupFuture future) throws Exception {
                if (future.isSuccess()) {
                    notifyStopped();
                } else {
                    notifyFailed(future.cause());
                }
            }
        });
    } catch (Throwable t) {
        notifyFailed(t);
        Throwables.propagate(t);
    }
}
 

public static CompletableFuture<Void> completable(ChannelGroupFuture future) {
  CompletableFuture<Void> cf = new CompletableFuture<>();
  future.addListener(
      (ChannelGroupFutureListener)
          future1 -> {
            if (future1.isSuccess()) {
              cf.complete(null);
            } else {
              cf.completeExceptionally(future1.cause());
            }
          });
  return cf;
}
 

/**
 * 在这里接收客户端的消息 在客户端和代理服务器建立连接时，也获得了代理服务器和目标服务器的通道outbound，
 * 通过outbound写入消息到目标服务器
 *
 * @param ctx
 * @param msg
 * @throws Exception
 */
@Override
public void channelRead0(final ChannelHandlerContext ctx, byte[] msg) throws Exception {

    log.info("客户端消息");
    allChannels.writeAndFlush(msg).addListener(new ChannelGroupFutureListener() {
        @Override
        public void operationComplete(ChannelGroupFuture future) throws Exception {
            //防止出现发送不成功造成的永久不读取消息的错误.
            ctx.channel().read();
        }
    });
}
 

public AsyncFuture<Void> stop() {
    final Channel server = this.server.getAndSet(null);

    if (server == null) {
        throw new IllegalStateException("server not started");
    }

    final ResolvableFuture<Void> serverClose = async.future();

    server.close().addListener((ChannelFutureListener) f -> {
        if (!f.isSuccess()) {
            serverClose.fail(f.cause());
            return;
        }

        serverClose.resolve(null);
    });

    final ResolvableFuture<Void> channelGroupClose = async.future();

    connected.close().addListener((ChannelGroupFutureListener) f -> {
        if (!f.isSuccess()) {
            channelGroupClose.fail(f.cause());
            return;
        }

        channelGroupClose.resolve(null);
    });

    return async.collectAndDiscard(
        ImmutableList.<AsyncFuture<Void>>of(serverClose, channelGroupClose));
}