takeaway
Original article, reprint please indicate the source.
This article source address: netty-source-code-Analysis
The version of Netty used in this article is version 4.1.6.final: annotated netty source code
In this “BIO vs NIO” article we show you a client called Hello World written using JDK native NIO. Remember the key steps? Let’s go over them again.
-
Create a SocketChannel
-
The server port is connected
-
Set SocketChannel to non-blocking
-
Registers a SocketChannel with a selector
Today we’re going to take a look at these key steps and see what netty does in between.
1 Client boot code
The following code bootstraps a client, which we refer to as “bootstrap code” for the rest of this article.
- To create a
EventLoopGroupUnlike the server, there is no need to createbossGroupSince the client does not need to handle new connections, theeventLoopGroupThe function of the server sideworkerGroup. - To create a
BootStrapAnd will beeventLoopGroupPass in, setchannelforNioSocketChannel(corresponding to the JDKSocketChannel). - Set up a
ChannelParameter and oneChannelProperties. - To configure a
handlerthehandlerWe did nothing but print some event logs. - call
bootstrap.connectMethod to connect to the server.
Running this program will print the following results in the console:
HandlerAdded
ChannelRegistered
ChannelActive
/** * Welcome to follow the public account "type code", Get blogger WeChat in-depth exchange * * @ author wangjianxin * / public class ty. Com.zhongdaima.net analysis. The bootstrap. ClientBoot {public static void main(String[] args) { EventLoopGroup eventLoopGroup = new NioEventLoopGroup(1); try { Bootstrap bootstrap = new Bootstrap(); Bootstrap.group (eventLoopGroup).channel(nioSocketChannel.class) // Set the channel parameter.option(channeloption.tcp_nodelay, Attr (attributeKey.valueof ("ChannelName"), "ClientChannel") .handler(new ChannelInboundHandlerAdapter(){ @Override public void channelRegistered(ChannelHandlerContext ctx) throws Exception { System.out.println("ChannelRegistered"); } @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { System.out.println("ChannelActive"); } @Override public void handlerAdded(ChannelHandlerContext ctx) throws Exception { System.out.println("HandlerAdded"); }}); ChannelFuture ChannelFuture = the bootstrap. Connect (" 127.0.0.1 ", 8000); Channel channel = channelFuture.syncUninterruptibly().channel(); channel.closeFuture().awaitUninterruptibly(); } finally { eventLoopGroup.shutdownGracefully(); }}}Copy the code2 Startup Process
We follow it from bootstrap.connect(“127.0.0.1”,8000), where the code is simple: encapsulates host and port into InetSocketAddress, and then calls doResolveAndConnect.
public ChannelFuture connect(String inetHost, int inetPort) {
return connect(InetSocketAddress.createUnresolved(inetHost, inetPort));
}
public ChannelFuture connect(SocketAddress remoteAddress) {
return doResolveAndConnect(remoteAddress, config.localAddress());
}
Copy the codeLet’s look at the oResolveAndConnect method. The key logic here is to call the initAndRegister and doResolveAndConnect0 methods.
private ChannelFuture doResolveAndConnect(final SocketAddress remoteAddress, final SocketAddress localAddress) {
// Key logic
final ChannelFuture regFuture = initAndRegister();
final Channel channel = regFuture.channel();
if (regFuture.isDone()) {
if(! regFuture.isSuccess()) {return regFuture;
}
// Key logic
return doResolveAndConnect0(channel, remoteAddress, localAddress, channel.newPromise());
} else {
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();
// Key logicdoResolveAndConnect0(channel, remoteAddress, localAddress, promise); }}});returnpromise; }}Copy the codeThere are three key steps in the initAndRegister method. 1 through the channelFactory. NewChannel () to create a Channel, the assignment of chnnelFactory here let’s have been in the “server startup process analysis, here no longer here; 2 is init(channel), where init is implemented in the Boostrap class. 3 is the config (.) group (). The register (channel).
final ChannelFuture initAndRegister(a) {
Channel channel = null;
try {
channel = channelFactory.newChannel();
init(channel);
} catch (Throwable t) {
}
ChannelFuture regFuture = config().group().register(channel);
return regFuture;
}
Copy the codeNext, let’s focus on the three key steps in the initAndRegister method and the doResolveAndConnect0 method.
2.1 Create a NioSocketChannel
ChannelFactory. NewChannel () call Channel the implementation class instance is created, no arguments constructor implementation classes for NioSocketChannel here, let’s with no arguments construction method to the class. Here with the server startup NioServerSocketChannel NioServerSocketChannel is invoked is used by the provider. OpenServerSocketChannel () to create a JDK ServerSocketCh Annel, while the client is to call the provider. OpenSocketChannel () to create a SocketChannel JDK. At this point, we see the first step “Create a SocketChannel” mentioned in the guide. Finally, we call the constructor of the parent class AbstractNioByteChannel.
public NioSocketChannel(a) {
this(DEFAULT_SELECTOR_PROVIDER);
}
public NioSocketChannel(SelectorProvider provider) {
this(newSocket(provider));
}
private static SocketChannel newSocket(SelectorProvider provider) {
try {
return provider.openSocketChannel();
} catch (IOException e) {
throw new ChannelException("Failed to open a socket.", e); }}public NioSocketChannel(SocketChannel socket) {
this(null, socket);
}
public NioSocketChannel(Channel parent, SocketChannel socket) {
super(parent, socket);
config = new NioSocketChannelConfig(this, socket.socket());
}
Copy the codeLet’s take a look at the constructor for AbstractNioByteChannel. It’s easy to continue calling the constructor for the parent class AbstractNioChannel.
protected AbstractNioByteChannel(Channel parent, SelectableChannel ch) {
super(parent, ch, SelectionKey.OP_READ);
}
Copy the codeCh.configureblocking (false) is called within the constructor of AbstractNioChannel to set the Channel to non-blocking, and it continues to call the constructor of the parent class AbstractChannel. Here we see step 3 “Set SocketChannel to non-blocking” mentioned in the guide.
protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
super(parent);
this.ch = ch;
this.readInterestOp = readInterestOp;
try {
// Set to non-blocking
ch.configureBlocking(false);
} catch (IOException e) {
}
}
Copy the codeAbstractChannel’s constructor assigns an ID to a Channel, creating an Unsafe and a PipeLine, and we’ll talk about that later.
protected AbstractChannel(Channel parent) {
this.parent = parent;
id = newId();
unsafe = newUnsafe();
pipeline = newChannelPipeline();
}
Copy the code2.2 Initializing a Channel
The init method is implemented in the Boostrap class, where we add the handler we set in the bootstrap code to the pipeline. Then set some parameters for the Channel (Option (channelOption.tcp_nodelay, true) in the boot code) and attributes (attr(AttributeKey.Valueof (“ChannelName”) in the boot code), “ClientChannel”)).
@Override @SuppressWarnings("unchecked") void init(Channel channel) throws Exception { ChannelPipeline p = channel.pipeline(); // Add the handler set in the boot code p.ddlast (config.handler()); Final Map<ChannelOption<? >, Object> options = options0(); synchronized (options) { for (Entry<ChannelOption<? >, Object> e: options.entrySet()) { try { if (! channel.config().setOption((ChannelOption<Object>) e.getKey(), e.getValue())) { logger.warn("Unknown channel option: " + e); } } catch (Throwable t) { logger.warn("Failed to set a channel option: " + channel, t); }}} final Map<AttributeKey<? >, Object> attrs = attrs0(); synchronized (attrs) { for (Entry<AttributeKey<? >, Object> e: attrs.entrySet()) { channel.attr((AttributeKey<Object>) e.getKey()).set(e.getValue()); }}Copy the code2.3 registered Channel
Let’s go back to AbstractBootstrap’s initAndRegister method and see ChannelFuture regFuture = config().group().register(channel), This is the place to register Channel. Let’s go in and have a look.
The return of config.group() is the eventLoopGroup we set in our bootstrap code.
Look at with to register (channel), the register method is abstract, concrete implementation in MultithreadEventLoopGroup, and go in.
@Override
public ChannelFuture register(Channel channel) {
return next().register(channel);
}
Copy the codeThe next() method calls the Next () method of EventExecutorChooser to select an EventLoop. EventExecutorChooser has two implementations, one is PowerOfTowEventExecutorChooser and GenericEventExecutorChooser, both with the Chooser is polling strategy, just different polling algorithm. If within EventLoopGroup EventLoop number is a power of 2, use PowerOfTowEventExecutorChooser, otherwise use GenericEventExecutorChooser.
PowerOfTowEventExecutorChooser use bit operation.
@Override
public EventExecutor next() {
return executors[idx.getAndIncrement() & executors.length - 1];
}
Copy the codeAnd GenericEventExecutorChooser use take over operation.
@Override
public EventExecutor next(a) {
return executors[Math.abs(idx.getAndIncrement() % executors.length)];
}
Copy the codeAs we can see from the selection algorithm of EventLoop, Netty does everything for performance.
Chooser attribute assignment within the constructor of MultithreadEventExecutorGroup through chooserFactory created.
protected MultithreadEventExecutorGroup(int nThreads, Executor executor,
EventExecutorChooserFactory chooserFactory, Object... args) {
chooser = chooserFactory.newChooser(children);
}
Copy the codeAnd chooserFactory assignment in MultithreadEventExecutorGroup another constructor. When we are in the boot code with new NioEventLoopGroup (1) create EventLoopGroup will call to the constructor, the default value is DefaultEventExecutorChooserFactory. The INSTANCE.
protected MultithreadEventExecutorGroup(int nThreads, Executor executor, Object... args) {
this(nThreads, executor, DefaultEventExecutorChooserFactory.INSTANCE, args);
}
Copy the codeThe EventLoop selected by the next() method is SingleThreadEventLoop, so we follow the register method for SingleThreadEventLoop, and finally call the register method for unsafe.
@Override
public ChannelFuture register(Channel channel) {
return register(new DefaultChannelPromise(channel, this));
}
@Override
public ChannelFuture register(final ChannelPromise promise) {
ObjectUtil.checkNotNull(promise, "promise");
promise.channel().unsafe().register(this, promise);
return promise;
}
Copy the codeUnsafe. The register method in io.net ty. Channel. AbstractChannel. AbstractUnsafe, we to go and see. The two main things to do in a register method are to bind an EventLoop and to call the register0 method. The calling thread is not an EventLoop thread and initiates an asynchronous task.
@Override
public final void register(EventLoop eventLoop, final ChannelPromise promise) {
/ / bind eventloop
AbstractChannel.this.eventLoop = eventLoop;
if (eventLoop.inEventLoop()) {
register0(promise);
// In this case, we are not in the EventLoop, that is, the current thread is not an EventLoop thread, will go to the branch
} else {
try {
eventLoop.execute(new Runnable() {
@Override
public void run(a) {
// Call the subclass register0 methodregister0(promise); }}); }catch (Throwable t) {
}
}
}
Copy the codeThere are three main steps within the register0 method.
The first step is doRegister(), which we’ll talk about later.
Step 2 is pipeline invokeHandlerAddedIfNeeded () this step is to complete the trigger to add handler before binding EventLoop operation, for example, we added a ChannelInitializer, The Handler added to the initChannel method of the ChannelInitalizer, which is called by the channelAdded method, which must be called inside an EventLoop, The call is encapsulated as an asynchronous task before an EventLoop is bound.
These operations are in the pendingHandlerCallbackHead in pipeline, is a two-way linked list, For details, see addLast(EventExecutorGroup Group, String Name, ChannelHandler Handler) of DefaultChannelPipeLine.
This step calls system.out.println (“HandlerAdded”) in our bootstrap and types “HandlerAdded” on the console.
Step 3 The ChannelRegistered event is triggered. This step calls system.out.println (“ChannelRegistered”) in our bootloader, typing “ChannelRegistered” on the console.
Ok, so now we know why our bootstrap is typing “HandlerAdded” and “ChannelRegistered” first.
Then we go down to isActive() and finally call the isOpen() and isConnected () methods of the JDK SocketChannel class. We’re not going to post the code, the reader is going to check it out, it’s easy, obviously we haven’t set up the connection yet, so the if branch code is not going to execute.
private void register0(ChannelPromise promise) {
try {
// Register the Channel with the Selector
doRegister();
/ / to do those in the binding EventLoop trigger to add handler before the operation, the operation is in the pendingHandlerCallbackHead in pipeline, a linked list, For details, see 'addLast(EventExecutorGroup Group, String Name, ChannelHandler Handler)' of DefaultChannelPipeLine.
pipeline.invokeHandlerAddedIfNeeded();
// Set the promise to be successful
safeSetSuccess(promise);
// The ChannelRegistered event is triggered
pipeline.fireChannelRegistered();
// There is no Active because the connection has not been established yet
if (isActive()) {
if (firstRegistration) {
pipeline.fireChannelActive();
} else if(config().isAutoRead()) { beginRead(); }}}catch (Throwable t) {
}
}
Copy the codeLet’s follow the doRegister method, which is an abstract method, and in this case the method is implemented in AbstractNioChannel. Ok, so this is where we finally get to step 4: registering a Channel with a Selector.
@Override
protected void doRegister(a) throws Exception {
boolean selected = false;
for (;;) {
try {
selectionKey = javaChannel().register(eventLoop().selector, 0.this);
return;
} catch (CancelledKeyException e) {
}
}
}
Copy the codeAt this point, out of the four steps that we talked about in the introduction, there’s a second step that we haven’t seen. Where is it? Let’s move on.
2.4 Connecting to the Server
If we go back to the doResolveAndConnect method of the Bootstrap class, we have already analyzed initAndRegister(), because initAndRegister is asynchronous and returns a Future. At this point, the Future is probably finished. It might not be done, but it’s a judgment call.
If the regFuture is complete, call doResolveAndConnect0 directly. Otherwise, put doResolveAndConnect0 in the regFuture Listener and wait until the regFuture is complete. It is called by the EventLoop thread.
private ChannelFuture doResolveAndConnect(final SocketAddress remoteAddress, final SocketAddress localAddress) {
final ChannelFuture regFuture = initAndRegister();
final Channel channel = regFuture.channel();
if (regFuture.isDone()) {
return doResolveAndConnect0(channel, remoteAddress, localAddress, channel.newPromise());
} else {
final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);
regFuture.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
if(cause ! =null) {
promise.setFailure(cause);
} else{ promise.registered(); doResolveAndConnect0(channel, remoteAddress, localAddress, promise); }}});returnpromise; }}Copy the codeBetween the time the if is completed and the time the Listener is added, the promise may have been completed, and the Listener may not have been called back. The mystery in DefaultPromise’s addListener(GenericFutureListener
> Listener), if the promise is completed, then the nofityListeners method is called directly to submit the asynchronous task to EventLoop (at this point, the EventLoop binding is complete). The asynchronous task is the Listener that the callback has just registered.
@Override
public Promise<V> addListener(GenericFutureListener<? extends Future<? super V>> listener) {
synchronized (this) {
addListener0(listener);
}
if (isDone()) {
notifyListeners();
}
return this;
}
Copy the codeLet’s look at the doResolveAndConnect0 method in the BootStrap class. The AddressResolver is used to resolve the SocketAddress. AddressResolver here. The default value is io.net ty resolver. DefaultAddressResolverGroup# INSTANCE, same here is asynchronous, and initAndRegister there is the same, let’s not much to discuss. We focused on the doConnect method.
private ChannelFuture doResolveAndConnect0(final Channel channel, SocketAddress remoteAddress,
final SocketAddress localAddress, final ChannelPromise promise) {
try {
final EventLoop eventLoop = channel.eventLoop();
final AddressResolver<SocketAddress> resolver = this.resolver.getResolver(eventLoop);
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) {
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;
}
Copy the codeIf we look at the doConnect method, in general, we don’t specify the localAddress of the client, so the localAddress is usually null, We follow the channel.connect(remoteAddress, connectPromise) method.
private static void doConnect( final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise connectPromise) { final Channel channel = connectPromise.channel(); 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); }}); }Copy the codeHere’s the overall architecture of Netty again.channel.connect(remoteAddress, connectPromise)The method is implemented inAbstractChannelClass, this is calledpipeline.connect(remoteAddress, promise)“Is called againtail.connect(remoteAddress, promise), the call will eventually be made fromtailPassed to thehead(refer to the above netty overall architecture diagram), how to pass the specific, let’s studypipeLineWe’ll talk about it when we talk about it. Let’s go straight toHeadContext.
@Override
public ChannelFuture connect(SocketAddress remoteAddress, ChannelPromise promise) {
return pipeline.connect(remoteAddress, promise);
}
@Override
public final ChannelFuture connect(SocketAddress remoteAddress, ChannelPromise promise) {
return tail.connect(remoteAddress, promise);
}
Copy the codeUnsafe.connect (remoteAddress, localAddress, promise) ¶ The implementation of this method is in the AbstractNioUnsafe class.
@Override
public void connect( ChannelHandlerContext ctx, SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise) throws Exception {
unsafe.connect(remoteAddress, localAddress, promise);
}
Copy the codeWe go to the Connect method of AbstractNioUnsafe, where the doConnect(remoteAddress, localAddress) method is called.
@Override
public final void connect(
final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) {
try {
boolean wasActive = isActive();
if (doConnect(remoteAddress, localAddress)) {
fulfillConnectPromise(promise, wasActive);
} else{}}catch(Throwable t) { promise.tryFailure(annotateConnectException(t, remoteAddress)); closeIfClosed(); }}Copy the codeThe doConnect method is implemented in NioSocketChannel. We said that the localAddress is usually null for active connection, so we won’t talk about doBind0 here. For those interested, go back to the “Server startup process” article to see the doBind0 method.
JavaChannel ().connect(remoteAddress) is called next. Since the Channel has been set to non-blocking, the connect operation is asynchronous and done asynchronously by the operating system. So the connect method here could either return true or it could return false. At this point, we have seen the second step mentioned in the guidance to connect to the server port. By this point, all of the operations mentioned in the guide have been found in Netty.
Take a look at the JDK comments for the connect method
If this channel is in non-blocking mode then an invocation of this method initiates a non-blocking connection operation. If the connection is established immediately, as can happen with a local connection, then this method returns true. Otherwise this method returns false and the connection operation must later be completed by invoking the finishConnect method.
translation
If the channel is in non-blocking mode, calling this method initializes a non-blocking join operation. This method returns true if the connection completes immediately, as might happen with a local connection, for example. Otherwise, the method returns false and must then call the finishConnect method to complete the connection.
That is, if connect returns true, the connection is complete. If false is returned, you need to call finishConnect to complete the connection.
If the connect method returns false and you need to call finishConnect to complete the connection, where is that call? The mystery is on the next line. If false is returned, add the selectionKey.op_connect interest event to the selectionKey. Call finishConnect when the bound EventLoop finds a selectionkey. OP_CONNECT event on the Channel, which we’ll look at later. If true is returned, the connection was successful and we continue back to AbstractNioUnsafe’s connect method
@Override
protected boolean doConnect(SocketAddress remoteAddress, SocketAddress localAddress) throws Exception {
if(localAddress ! =null) {
doBind0(localAddress);
}
boolean success = false;
try {
boolean connected = javaChannel().connect(remoteAddress);
// In most cases connected is false
if(! connected) { selectionKey().interestOps(SelectionKey.OP_CONNECT); } success =true;
return connected;
} finally {
if(! success) { doClose(); }}}Copy the codeAnd if doConnect returns true, then we call the fulfillConnectPromise method, just as the name suggests, and set the promise to fulfill. let’s just follow through and see what happens with doConnect returning false and we’ll see in just a second.
public final void connect(
final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) {
try {
boolean wasActive = isActive();
if (doConnect(remoteAddress, localAddress)) {
fulfillConnectPromise(promise, wasActive);
} else{}}catch(Throwable t) { promise.tryFailure(annotateConnectException(t, remoteAddress)); closeIfClosed(); }}Copy the codeIn the fulfillConnectPromise method, I first call the promise.trySuccess() method to set the promise to complete, and I call the pipeline().firechannelActive () method, So this will eventually call system.out.println (“ChannelActive”) in our bootstrap code and print out ChannelActive on the console.
private void fulfillConnectPromise(ChannelPromise promise, boolean wasActive) {
boolean active = isActive();
boolean promiseSet = promise.trySuccess();
if(! wasActive && active) { pipeline().fireChannelActive(); }if (!promiseSet) {
close(voidPromise());
}
}
Copy the codeIf the doConnect method returns false, first assign the connectPromise in the AbstractNioChannel class to the promise passed in the parameter. Why do we assign this value? A scheduled task has been added that will set the connectPromise to fail when the connection timeout period arrives.
public final void connect(
final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) {
try {
boolean wasActive = isActive();
if (doConnect(remoteAddress, localAddress)) {
fulfillConnectPromise(promise, wasActive);
} else {
int connectTimeoutMillis = config().getConnectTimeoutMillis();
if (connectTimeoutMillis > 0) {
connectTimeoutFuture = eventLoop().schedule(new Runnable() {
@Override
public void run(a) {
ChannelPromise connectPromise = AbstractNioChannel.this.connectPromise;
ConnectTimeoutException cause =
new ConnectTimeoutException("connection timed out: " + remoteAddress);
if(connectPromise ! =null && connectPromise.tryFailure(cause)) {
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(); }}Copy the codeOk, so we said that the doConnect method will most likely return false, and the connection is not complete, so where is the actual connection completion? We already said that, if the connect method returns false then we have this line selectionKey().interestOps(selectionKey.op_connect); Add a selectionKey. OP_CONNECT interest event.
This is something that we haven’t done yet, so I’m just going to give you, and how we get there, we’ll talk about it later. Go straight to NioEventLoop’s processSelectedKey method, which is so long that we’re just going to take a little bit of it. OP_CONNECT = selectionKey.op_connect = selectionkey.op_connect = selectionkey.op_connect = selectionkey.op_connect = selectionKey.op_connect = selectionKey.op_connect = selectionKey.op_connect = selectionKey.op_connect = selectionKey.op_connect = selectionKey.op_connect = selectionKey.op_connect The unsafe.finishConnect() method is then called.
if((readyOps & SelectionKey.OP_CONNECT) ! =0) {
// remove OP_CONNECT as otherwise Selector.select(..) will always return without blocking
// See https://github.com/netty/netty/issues/924
int ops = k.interestOps();
ops &= ~SelectionKey.OP_CONNECT;
k.interestOps(ops);
unsafe.finishConnect();
}
Copy the codeLooking at the unsafe.finishConnect() method, which is implemented in AbstractNioUnsafe, calling doFinishFinishConnect() is the key to this method. FulfillConnectPromise (connectPromise, wasActive), which we just talked about earlier, and I’m just not going to analyze anymore. However, remember that we assigned the promise passed in the connect method of AbstractNioUnsafe to the connectPromise property of AbstractNioChannel? And here’s why, if you don’t save that promise in your attribute, you just can’t pass the connectPromise to the fulfilltPromise (connectPromise, wasActive) method.
@Override
public final void finishConnect(a) {
assert eventLoop(a).inEventLoop(a);
try {
boolean wasActive = isActive();
doFinishConnect();
fulfillConnectPromise(connectPromise, wasActive);
} catch (Throwable t) {
} finally{}}Copy the codeThe implementation of doFinishFinishConnect() in the NioSocketChannel class is quite simple, calling the finishConnect method of the JDK’s NioSocketChannel class. Remember the question we left you with? Where is the finishiConnect() method called? The answer is already there, right here.
@Override
protected void doFinishConnect() throws Exception {
if (!javaChannel().finishConnect()) {
throw new Error();
}
}
Copy the codeAt this point, we have analyzed the client startup process.
3 summary
Netty client startup process:
-
Creating a Channel instance makes the Channel non-blocking, creating a PipeLine and an Unsafe instance for the Channel.
-
Initialize the Channel, add the handler set in the bootstrap code to the Channel, and set the parameters and properties.
-
Register the Channel, bind an EventLoop to the Channel, and register the Channel with the Selector.
-
Call the connect method to connect to the server. If connect returns false, call the finishiConnect method to complete the connection when the Channel OP_CONNECT event occurs.
About the author
Wang Jianxin, senior Java engineer of architecture department, mainly responsible for service governance, RPC framework, distributed call tracking, monitoring system, etc. Love technology, love learning, welcome to contact and exchange.