io.netty.channel.ChannelFuture#cancel ( )源码实例Demo

public void testConnectCancellation(Bootstrap cb) throws Throwable {
    cb.handler(new TestHandler()).option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 4000);
    ChannelFuture future = cb.connect(BAD_HOST, BAD_PORT);
    try {
        if (future.await(1000)) {
            if (future.isSuccess()) {
                fail("A connection attempt to " + BAD_HOST + " must not succeed.");
            } else {
                throw future.cause();
            }
        }

        if (future.cancel(true)) {
            assertThat(future.channel().closeFuture().await(500), is(true));
            assertThat(future.isCancelled(), is(true));
        } else {
            // Cancellation not supported by the transport.
        }
    } finally {
        future.channel().close();
    }
}
 

private void waitWriteDone(ChannelFuture future, int timeout, RequestProtocol request, boolean needRemoveTrace)
		throws SailfishException {
	boolean done = future.awaitUninterruptibly(timeout);
	if (!done) {
		// useless at most of time when do writeAndFlush(...) invoke
		future.cancel(true);
		if (needRemoveTrace) {
			getTracer().remove(request.packetId());
		}
		throw new SailfishException(ExceptionCode.WRITE_TIMEOUT,
				String.format("write to remote[%s] timeout, protocol[%s]", channel.remoteAddress(), request));
	}
	if (!future.isSuccess()) {
		if (needRemoveTrace) {
			getTracer().remove(request.packetId());
		}
		throw new SailfishException(ExceptionCode.CHANNEL_WRITE_FAIL,
				String.format("write to remote[%s] fail, protocol[%s]", channel.remoteAddress(), request),
				future.cause());
	}
}
 

public void testConnectCancellation(Bootstrap cb) throws Throwable {
    cb.handler(new TestHandler()).option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 4000);
    ChannelFuture future = cb.connect(BAD_HOST, BAD_PORT);
    try {
        if (future.await(1000)) {
            if (future.isSuccess()) {
                fail("A connection attempt to " + BAD_HOST + " must not succeed.");
            } else {
                throw future.cause();
            }
        }

        if (future.cancel(true)) {
            assertThat(future.channel().closeFuture().await(500), is(true));
            assertThat(future.isCancelled(), is(true));
        } else {
            // Cancellation not supported by the transport.
        }
    } finally {
        future.channel().close();
    }
}
 

@Override
public void operationComplete(final ChannelFuture future) throws Exception {
    if (future.isCancelled()) {
        LOG.debug("Connection {} cancelled!", future);
    } else if (future.isSuccess()) {
        LOG.debug("Connection {} succeeded!", future);
        future.channel().closeFuture().addListener((ChannelFutureListener) channelFuture -> scheduleConnect());
    } else {
        if (this.delay > MAXIMUM_BACKOFF) {
            LOG.warn("The time of maximum backoff has been exceeded. No further connection attempts with BMP "
                    + "router {}.", this.remoteAddress);
            future.cancel(false);
            return;
        }
        final EventLoop loop = future.channel().eventLoop();
        loop.schedule(() -> this.bootstrap.connect().addListener(this), this.delay, TimeUnit.MILLISECONDS);
        LOG.info("The connection try to BMP router {} failed. Next reconnection attempt in {} milliseconds.",
                this.remoteAddress, this.delay);
        this.delay *= 2;
    }
}
 

/**
 * 安静的关闭future，不产生任何影响
 */
public static void closeQuietly(@Nullable ChannelFuture channelFuture) {
    if (null != channelFuture) {
        try {
            channelFuture.cancel(true);
            channelFuture.channel().close();
        } catch (Throwable ignore) {

        }
    }
}
 

@Override
protected void doConnect() {
    ChannelFuture future = null;
    if (isClosed()) {
        logger.warn("{} closed, won't do channel connect", this.toString());
    }
    try {
        future = bootstrap.connect(getRemoteAddress());
        future.awaitUninterruptibly();
        if (future.isCancelled()) {
            throw new TransportException("Failed connect to server " + getRemoteAddress() + " from " + getClass().getName() + ", cause it be cancelled");
        } else if (!future.isSuccess()) {
            String errorMsg = future.cause() != null ? future.cause().getMessage() : "";
            throw new TransportException("Failed connect to server " + getRemoteAddress() + " from " + getClass().getName() + "[timeout:" + connTimeout + "ms], cause:" + errorMsg, future.cause());
        }
        NettyChannel oldChannel = this.channel;
        NettyChannel newChannel = ChannelManager.getOrAddChannel(future.channel());
        this.channel = newChannel;

        // 关闭旧的连接
        if (oldChannel != null) {
            if (oldChannel.isConnected()) {
                logger.info("Close old netty channel:{} and create new netty channel:{}", oldChannel, newChannel);
                oldChannel.close();
            }
        }

    } catch (Exception e) {
        if (e instanceof TransportException) {
            throw e;
        }
        throw new TransportException("Failed to connect to server " + getRemoteAddress(), e);
    } finally {
        if (future != null && !isConnected()) {
            future.cancel(true);
        }
    }
}
 

public void destroy() {
  LOG.info("Destroying server on address(es) {}", addresses);
  for (ChannelFuture channelFuture : channelFutures) {
    if (channelFuture != null && channelFuture.isCancellable()) {
      channelFuture.cancel(true);
    }
  }
  try {
    shutdownGroups();
  } finally {
    channelFutures.clear();
  }
}
 

@Override
protected void doConnect() throws Throwable {
    long start = System.currentTimeMillis();
    ChannelFuture future = bootstrap.connect(getConnectAddress()).sync();
    try{
        boolean ret = future.awaitUninterruptibly(getConnectTimeout(), TimeUnit.MILLISECONDS);
        if (ret && future.isSuccess()) {
            io.netty.channel.Channel newChannel = future.channel();
            try {
                // 关闭旧的连接
                io.netty.channel.Channel oldChannel = Netty4Client.this.channel; // copy reference
                if (oldChannel != null) {
                    try {
                        if (logger.isInfoEnabled()) {
                            logger.info("Close old netty channel " + oldChannel + " on create new netty channel " + newChannel);
                        }
                        oldChannel.close();
                    } finally {
                        Netty4Channel.removeChannelIfDisconnected(oldChannel);
                    }
                }
            } finally {
                if (Netty4Client.this.isClosed()) {
                    try {
                        if (logger.isInfoEnabled()) {
                            logger.info("Close new netty channel " + newChannel + ", because the client closed.");
                        }
                        newChannel.close();
                    } finally {
                        Netty4Client.this.channel = null;
                        Netty4Channel.removeChannelIfDisconnected(newChannel);
                    }
                } else {
                    Netty4Client.this.channel = newChannel;
                }
            }
        } else if (future.cause() != null) {
            throw new RemotingException(this, "client(url: " + getUrl() + ") failed to connect to server "
                    + getRemoteAddress() + ", error message is:" + future.cause().getMessage(), future.cause());
        } else {
            throw new RemotingException(this, "client(url: " + getUrl() + ") failed to connect to server "
                    + getRemoteAddress() + " client-side timeout "
                    + getConnectTimeout() + "ms (elapsed: " + (System.currentTimeMillis() - start) + "ms) from netty client "
                    + NetUtils.getLocalHost() + " using dubbo version " + Version.getVersion());
        }
    }finally{
        if (! isConnected()) {
            future.cancel(true);
        }
    }
}