【相關(guān)源碼都是出自4.1.55.Final-SNAPSHOT版本】
了解Netty服務(wù)端的啟動(dòng)過(guò)程后，現(xiàn)在換個(gè)角度，客戶(hù)端是怎樣啟動(dòng)呢？這一篇文章基于【Netty源碼系列】服務(wù)端啟動(dòng)流程的解析，如果有相似的代碼我會(huì)簡(jiǎn)單帶過(guò)，不會(huì)再贅述。
現(xiàn)在咱們先看下官方example是怎樣啟動(dòng)客戶(hù)端的，為了使讀者更加專(zhuān)注客戶(hù)端啟動(dòng)的流程，所以我刪去非啟動(dòng)不要的流程和加上一些注釋?zhuān)M量精簡(jiǎn)代碼，使我們能夠快速的掌握Netty客戶(hù)端啟動(dòng)的流程。
public final class EchoClient {
// Netty服務(wù)端的主機(jī)名
static final String HOST = System.getProperty("host", "127.0.0.1");
// Netty服務(wù)端的端口
static final int PORT = Integer.parseInt(System.getProperty("port", "8007"));
public static void main(String[] args) throws Exception {
// 配置線(xiàn)程池組
EventLoopGroup group = new NioEventLoopGroup();
try {
Bootstrap b = new Bootstrap();
b.group(group)
.channel(NioSocketChannel.class) // SocketChannel類(lèi)型為NioSocketChannel
.option(ChannelOption.TCP_NODELAY, true) // SocketChannel相關(guān)TCP參數(shù)配置
.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception { // 配置 SocketChannel 處理器
ChannelPipeline p = ch.pipeline();
p.addLast(new EchoClientHandler());
}
});
// 啟動(dòng)客戶(hù)端
ChannelFuture f = b.connect(HOST, PORT).sync();
// 等待直到客戶(hù)端channel關(guān)閉
f.channel().closeFuture().sync();
} finally {
// 優(yōu)雅關(guān)閉線(xiàn)程池組
group.shutdownGracefully();
}
}
}
和Netty服務(wù)端啟動(dòng)的代碼大部分都是相似的，無(wú)非都是要一開(kāi)始啟動(dòng)線(xiàn)程池組，然后配置channel的類(lèi)型，相關(guān)TCP參數(shù)和處理器，當(dāng)配置完之后阻塞啟動(dòng)，最后優(yōu)雅關(guān)閉線(xiàn)程池組。除了客戶(hù)端啟動(dòng)時(shí)調(diào)用connect方法，其它基本與Netty服務(wù)端啟動(dòng)的流程基本一樣，所以這篇文章主要就是分析Netty客戶(hù)端調(diào)用connect的過(guò)程。
客戶(hù)端啟動(dòng)——connect
public ChannelFuture connect(String inetHost, int inetPort) {
return connect(InetSocketAddress.createUnresolved(inetHost, inetPort));
}
// JDK原生方法，根據(jù)主機(jī)名和端口號(hào)創(chuàng)建未解析的套接字地址
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并注冊(cè)到SocketChannel上
final ChannelFuture regFuture = initAndRegister();
final Channel channel = regFuture.channel();
// 因?yàn)閕nitAndRegister方法是異步的，所以regFuture有可能是還未注冊(cè)，因此程序需要作出判斷
if (regFuture.isDone()) {
if (!regFuture.isSuccess()) {
return regFuture;
}
return doResolveAndConnect0(channel, remoteAddress, localAddress, channel.newPromise());
} else { // 如果regFuture還未完成注冊(cè)，則加上監(jiān)聽(tīng)器，當(dāng)注冊(cè)完成后回調(diào)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;
}
}
執(zhí)行initAndRegister方法的流程，在Netty服務(wù)端啟動(dòng)流程系列已重點(diǎn)分析，不太清楚的讀者可參考【Netty源碼系列】服務(wù)端啟動(dòng)流程（三）綁定端口并啟動(dòng)
當(dāng)NioSocketChannel初始化和注冊(cè)完成后，下一步就會(huì)調(diào)用doResolveAndConnect0方法
/**
* 解析遠(yuǎn)程地址并進(jìn)行連接
*/
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;
}
// 解析遠(yuǎn)程地址
final Future<SocketAddress> resolveFuture = resolver.resolve(remoteAddress);
if (resolveFuture.isDone()) {
final Throwable resolveFailureCause = resolveFuture.cause();
if (resolveFailureCause != null) {
// 異常處理
channel.close();
promise.setFailure(resolveFailureCause);
} else {
// 連接遠(yuǎn)程地址
doConnect(resolveFuture.getNow(), localAddress, promise);
}
return promise;
}
// 解析沒(méi)完成時(shí)等待解析完成
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 {
// 解析成功開(kāi)始連接
doConnect(future.getNow(), localAddress, promise);
}
}
});
} catch (Throwable cause) {
promise.tryFailure(cause);
}
return promise;
}
調(diào)用doResolveAndConnect0是為了連接服務(wù)器，該方法中對(duì)遠(yuǎn)程地址的解析，最終都會(huì)調(diào)用doConnect方法
private static void doConnect(
final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise connectPromise) {
final Channel channel = connectPromise.channel();
// 根據(jù)channel綁定的eventLoop線(xiàn)程執(zhí)行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方法實(shí)際上是在AbstractChannel類(lèi)聲明的
@Override
public ChannelFuture connect(SocketAddress remoteAddress, ChannelPromise promise) {
return pipeline.connect(remoteAddress, promise);
}
/**
* DefaultChannelPipeline類(lèi)
*/
@Override
public final ChannelFuture connect(SocketAddress remoteAddress, ChannelPromise promise) {
return tail.connect(remoteAddress, promise);
}
/**
* AbstractChannelHandlerContext類(lèi)
*/
@Override
public ChannelFuture connect(SocketAddress remoteAddress, ChannelPromise promise) {
return connect(remoteAddress, null, promise);
}
/**
* AbstractChannelHandlerContext類(lèi)
*/
@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;
}
通過(guò)findContextOutbound方法，從TailContext向前找到第一個(gè)OutBound類(lèi)型為true，實(shí)際上也就是HeadContext，然后調(diào)用當(dāng)前類(lèi)的invokeConnect方法，調(diào)用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類(lèi)
*/
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類(lèi)
*/
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類(lèi)
*/
@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();
}
}
}
經(jīng)過(guò)漫長(zhǎng)的debug，終于到達(dá)客戶(hù)端啟動(dòng)連接服務(wù)端的最底層，實(shí)際上就是通過(guò)JDK原生類(lèi)SocketUtils調(diào)用connect方法完成服務(wù)端的連接。為了梳理整個(gè)客戶(hù)端啟動(dòng)連接服務(wù)端的流程，特意畫(huà)出時(shí)序圖幫助理解。
以上就是Netty客戶(hù)端的啟動(dòng)連接服務(wù)端過(guò)程，那么當(dāng)Netty客戶(hù)端發(fā)起請(qǐng)求的時(shí)候，Netty服務(wù)端會(huì)有哪些操作和改變呢？接下來(lái)請(qǐng)看【Netty源碼系列】服務(wù)端接收請(qǐng)求流程
