【相关源码都是出自4.1.55.Final-SNAPSHOT版本】
了解Netty服务端的启动过程后，现在换个角度，客户端是怎样启动呢？这一篇文章基于【Netty源码系列】服务端启动流程的解析，如果有相似的代码我会简单带过，不会再赘述。
现在咱们先看下官方example是怎样启动客户端的，为了使读者更加专注客户端启动的流程，所以我删去非启动不要的流程和加上一些注释，尽量精简代码，使我们能够快速的掌握Netty客户端启动的流程。
public final class EchoClient {
// Netty服务端的主机名
static final String HOST = System.getProperty("host", "127.0.0.1");
// Netty服务端的端口
static final int PORT = Integer.parseInt(System.getProperty("port", "8007"));
public static void main(String[] args) throws Exception {
// 配置线程池组
EventLoopGroup group = new NioEventLoopGroup();
try {
Bootstrap b = new Bootstrap();
b.group(group)
.channel(NioSocketChannel.class) // SocketChannel类型为NioSocketChannel
.option(ChannelOption.TCP_NODELAY, true) // SocketChannel相关TCP参数配置
.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception { // 配置 SocketChannel 处理器
ChannelPipeline p = ch.pipeline();
p.addLast(new EchoClientHandler());
}
});
// 启动客户端
ChannelFuture f = b.connect(HOST, PORT).sync();
// 等待直到客户端channel关闭
f.channel().closeFuture().sync();
} finally {
// 优雅关闭线程池组
group.shutdownGracefully();
}
}
}
和Netty服务端启动的代码大部分都是相似的，无非都是要一开始启动线程池组，然后配置channel的类型，相关TCP参数和处理器，当配置完之后阻塞启动，最后优雅关闭线程池组。除了客户端启动时调用connect方法，其它基本与Netty服务端启动的流程基本一样，所以这篇文章主要就是分析Netty客户端调用connect的过程。
客户端启动——connect
public ChannelFuture connect(String inetHost, int inetPort) {
return connect(InetSocketAddress.createUnresolved(inetHost, inetPort));
}
// JDK原生方法，根据主机名和端口号创建未解析的套接字地址
public static InetSocketAddress createUnresolved(String host, int port) {
return new InetSocketAddress(checkPort(port), checkHost(host));
}
public ChannelFuture connect(SocketAddress remoteAddress) {
ObjectUtil.checkNotNull(remoteAddress, "remoteAddress");
validate();
return doResolveAndConnect(remoteAddress, config.localAddress());
}
private ChannelFuture doResolveAndConnect(final SocketAddress remoteAddress, final SocketAddress localAddress) {
// 初始化channel并注册到SocketChannel上
final ChannelFuture regFuture = initAndRegister();
final Channel channel = regFuture.channel();
// 因为initAndRegister方法是异步的，所以regFuture有可能是还未注册，因此程序需要作出判断
if (regFuture.isDone()) {
if (!regFuture.isSuccess()) {
return regFuture;
}
return doResolveAndConnect0(channel, remoteAddress, localAddress, channel.newPromise());
} else { // 如果regFuture还未完成注册，则加上监听器，当注册完成后回调doResolveAndConnect0方法
final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);
regFuture.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
Throwable cause = future.cause();
if (cause != null) {
promise.setFailure(cause);
} else {
promise.registered();
doResolveAndConnect0(channel, remoteAddress, localAddress, promise);
}
}
});
return promise;
}
}
执行initAndRegister方法的流程，在Netty服务端启动流程系列已重点分析，不太清楚的读者可参考【Netty源码系列】服务端启动流程（三）绑定端口并启动
当NioSocketChannel初始化和注册完成后，下一步就会调用doResolveAndConnect0方法
/**
* 解析远程地址并进行连接
*/
private ChannelFuture doResolveAndConnect0(final Channel channel, SocketAddress remoteAddress,
final SocketAddress localAddress, final ChannelPromise promise) {
try {
// 获取channel绑定的eventLoop
final EventLoop eventLoop = channel.eventLoop();
AddressResolver<SocketAddress> resolver;
try {
resolver = this.resolver.getResolver(eventLoop);
} catch (Throwable cause) {
channel.close();
return promise.setFailure(cause);
}
if (!resolver.isSupported(remoteAddress) || resolver.isResolved(remoteAddress)) {
doConnect(remoteAddress, localAddress, promise);
return promise;
}
// 解析远程地址
final Future<SocketAddress> resolveFuture = resolver.resolve(remoteAddress);
if (resolveFuture.isDone()) {
final Throwable resolveFailureCause = resolveFuture.cause();
if (resolveFailureCause != null) {
// 异常处理
channel.close();
promise.setFailure(resolveFailureCause);
} else {
// 连接远程地址
doConnect(resolveFuture.getNow(), localAddress, promise);
}
return promise;
}
// 解析没完成时等待解析完成
resolveFuture.addListener(new FutureListener<SocketAddress>() {
@Override
public void operationComplete(Future<SocketAddress> future) throws Exception {
if (future.cause() != null) { // 异常处理
channel.close();
promise.setFailure(future.cause());
} else {
// 解析成功开始连接
doConnect(future.getNow(), localAddress, promise);
}
}
});
} catch (Throwable cause) {
promise.tryFailure(cause);
}
return promise;
}
调用doResolveAndConnect0是为了连接服务器，该方法中对远程地址的解析，最终都会调用doConnect方法
private static void doConnect(
final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise connectPromise) {
final Channel channel = connectPromise.channel();
// 根据channel绑定的eventLoop线程执行connect方法
channel.eventLoop().execute(new Runnable() {
@Override
public void run() {
if (localAddress == null) {
channel.connect(remoteAddress, connectPromise);
} else {
channel.connect(remoteAddress, localAddress, connectPromise);
}
connectPromise.addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
}
});
}
channel.connect方法实际上是在AbstractChannel类声明的
@Override
public ChannelFuture connect(SocketAddress remoteAddress, ChannelPromise promise) {
return pipeline.connect(remoteAddress, promise);
}
/**
* DefaultChannelPipeline类
*/
@Override
public final ChannelFuture connect(SocketAddress remoteAddress, ChannelPromise promise) {
return tail.connect(remoteAddress, promise);
}
/**
* AbstractChannelHandlerContext类
*/
@Override
public ChannelFuture connect(SocketAddress remoteAddress, ChannelPromise promise) {
return connect(remoteAddress, null, promise);
}
/**
* AbstractChannelHandlerContext类
*/
@Override
public ChannelFuture connect(
final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) {
ObjectUtil.checkNotNull(remoteAddress, "remoteAddress");
if (isNotValidPromise(promise, false)) {
return promise;
}
final AbstractChannelHandlerContext next = findContextOutbound(MASK_CONNECT);
EventExecutor executor = next.executor();
if (executor.inEventLoop()) {
next.invokeConnect(remoteAddress, localAddress, promise);
} else {
safeExecute(executor, new Runnable() {
@Override
public void run() {
next.invokeConnect(remoteAddress, localAddress, promise);
}
}, promise, null, false);
}
return promise;
}
通过findContextOutbound方法，从TailContext向前找到第一个OutBound类型为true，实际上也就是HeadContext，然后调用当前类的invokeConnect方法，调用connect方法
private void invokeConnect(SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise) {
if (invokeHandler()) {
try {
((ChannelOutboundHandler) handler()).connect(this, remoteAddress, localAddress, promise);
} catch (Throwable t) {
notifyOutboundHandlerException(t, promise);
}
} else {
connect(remoteAddress, localAddress, promise);
}
}
/**
* DefaultChannelPipeline类
*/
final class HeadContext extends AbstractChannelHandlerContext implements ChannelOutboundHandler, ChannelInboundHandler {
@Override
public void connect(
ChannelHandlerContext ctx,
SocketAddress remoteAddress, SocketAddress localAddress,
ChannelPromise promise) {
unsafe.connect(remoteAddress, localAddress, promise);
}
}
/**
* AbstractNioChannel类
*/
protected abstract class AbstractNioUnsafe extends AbstractUnsafe implements NioUnsafe {
@Override
public final void connect(
final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) {
if (!promise.setUncancellable() || !ensureOpen(promise)) {
return;
}
try {
if (connectPromise != null) {
throw new ConnectionPendingException();
}
boolean wasActive = isActive();
if (doConnect(remoteAddress, localAddress)) {
fulfillConnectPromise(promise, wasActive);
} else {
connectPromise = promise;
requestedRemoteAddress = remoteAddress;
int connectTimeoutMillis = config().getConnectTimeoutMillis();
if (connectTimeoutMillis > 0) {
connectTimeoutFuture = eventLoop().schedule(new Runnable() {
@Override
public void run() {
ChannelPromise connectPromise = AbstractNioChannel.this.connectPromise;
if (connectPromise != null && !connectPromise.isDone()
&& connectPromise.tryFailure(new ConnectTimeoutException(
"connection timed out: " + remoteAddress))) {
close(voidPromise());
}
}
}, connectTimeoutMillis, TimeUnit.MILLISECONDS);
}
promise.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
if (future.isCancelled()) {
if (connectTimeoutFuture != null) {
connectTimeoutFuture.cancel(false);
}
connectPromise = null;
close(voidPromise());
}
}
});
}
} catch (Throwable t) {
promise.tryFailure(annotateConnectException(t, remoteAddress));
closeIfClosed();
}
}
}
/**
* NioSocketChannel类
*/
@Override
protected boolean doConnect(SocketAddress remoteAddress, SocketAddress localAddress) throws Exception {
if (localAddress != null) {
doBind0(localAddress);
}
boolean success = false;
try {
boolean connected = SocketUtils.connect(javaChannel(), remoteAddress);
if (!connected) {
selectionKey().interestOps(SelectionKey.OP_CONNECT);
}
success = true;
return connected;
} finally {
if (!success) {
doClose();
}
}
}
经过漫长的debug，终于到达客户端启动连接服务端的最底层，实际上就是通过JDK原生类SocketUtils调用connect方法完成服务端的连接。为了梳理整个客户端启动连接服务端的流程，特意画出时序图帮助理解。
以上就是Netty客户端的启动连接服务端过程，那么当Netty客户端发起请求的时候，Netty服务端会有哪些操作和改变呢？接下来请看【Netty源码系列】服务端接收请求流程
网友评论