Shulou(Shulou.com)06/01 Report--

This article mainly introduces "what is the Netty server startup source code". In the daily operation, I believe that many people have doubts about what the Netty server startup source code is. Xiaobian consulted all kinds of materials and sorted out a simple and easy-to-use method of operation. I hope it will be helpful to answer the doubt of "what is the Netty server startup source code?" Next, please follow the editor to study!

First, start with the example of EchoServer

Where do the examples come from? Any open source framework will have its own sample code, and the Netty source code is no exception. For example, the most common EchoServer example is included in the module netty-example. Let's use this example to enter the server startup process chapter.

Public final class EchoServer {static final boolean SSL = System.getProperty ("ssl")! = null; static final int PORT = Integer.parseInt (System.getProperty ("port", "8007")); public static void main (String [] args) throws Exception {/ / Configure SSL. Final SslContext sslCtx; if (SSL) {SelfSignedCertificate ssc = new SelfSignedCertificate (); sslCtx = SslContextBuilder.forServer (ssc.certificate (), ssc.privateKey ()). Build ();} else {sslCtx = null;} / / 1. Declare the Main-Sub Reactor mode thread pool: EventLoopGroup / / Configure the server. EventLoopGroup bossGroup = new NioEventLoopGroup (1); EventLoopGroup workerGroup = new NioEventLoopGroup (); / / create EchoServerHandler object final EchoServerHandler serverHandler = new EchoServerHandler (); try {/ / 2. Declare that the server launches the bootstrap and sets the related property ServerBootstrap b = new ServerBootstrap () B.group (bossGroup, workerGroup) .channel (NioServerSocketChannel.class) .option (ChannelOption.SO_BACKLOG, 100) .handler (new LoggingHandler (LogLevel.INFO)) .childHandler (new ChannelInitializer () {@ Override public void initChannel (SocketChannel ch) throws Exception {ChannelPipeline p = ch.pipeline () If (sslCtx! = null) {p.addLast (sslCtx.newHandler (ch.alloc ();} / p.addLast (new LoggingHandler (LogLevel.INFO)); p.addLast (serverHandler);}}); / / 3. Bind the port to start the server and wait for / / Start the server synchronously. ChannelFuture f = b.bind (PORT). Sync (); / / 4 The listening server shuts down and blocks waiting for / / Wait until the server socket is closed. F.channel (). CloseFuture (). Sync ();} finally {/ / 5. Gracefully close both EventLoopGroup thread pools / / Shut down all event loops to terminate all threads. BossGroup.shutdownGracefully (); workerGroup.shutdownGracefully ();}

[lines 18, 19] declare the Main-Sub Reactor mode thread pool: EventLoopGroup

Create two EventLoopGroup objects. Among them, bossGroup is used for the server to accept the connection of the client, and workerGroup is used to read and write the data of the SocketChannel of the client.

(EventLoopGroup is not the focus of this article, so it will be analyzed in subsequent articles.)

[line 23-39] declares that the server starts the bootstrap and sets the relevant properties

AbstractBootstrap is a helper class that provides an easy-to-use way to configure and launch a Channel through the method chain (method chaining). Io.netty.bootstrap.ServerBootstrap, which implements the AbstractBootstrap abstract class, is used for the initiator implementation class of Server. Io.netty.bootstrap.Bootstrap, which implements the AbstractBootstrap abstract class, is used for the initiator implementation class of Client. The following class diagram shows:

! [AbstractBootstrap class inherits .png] (/ / p3-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/d877706356d4427a9afd4b13d7177142~tplv-k3u1fbpfcp-zoom-1.image)

(in the EchoServer sample code, we see that ServerBootstrap's group, channel, option, childHandler and other attribute chain settings are described in detail in the code about the AbstractBootstrap system.)

[code line 43] bind the port to start the server and wait synchronously

First call the # bind (int port) method, bind the port, and then call the ChannelFuture#sync () method, blocking and waiting for success. For bind operation, it is the "server startup process" that this article will describe in detail.

[code line 47] the listening server shuts down and blocks waiting

First call the # closeFuture () method, shut down the listening server, and then call the ChannelFuture#sync () method, blocking and waiting for success. Note that instead of shutting down the server, the snooping of channel is turned off.

[lines 51, 52] gracefully close two EventLoopGroup thread pools

The execution instructions in the finally code block indicate that the server will eventually shut down, so the EventLoopGroup#shutdownGracefully () method is called to close the two EventLoopGroup objects respectively, terminating all threads.

II. Service start-up process

Before analyzing the source code of the service startup process, let's review the initial code we started on the server side through JDK NIO programming:

ServerSocketChannel = ServerSocketChannel.open (); serverSocketChannel.configureBlocking (false); serverSocketChannel.socket (). Bind (new InetSocketAddress (port), 1024); selector = Selector.open (); serverSocketChannel.register (selector, SelectionKey.OP_ACCEPT)

These five lines of code mark the most familiar process:

Open serverSocketChannel

Configure non-blocking mode

Bind a listening port for the socket of channel

Create Selector

Register serverSocketChannel with selector

Later, after analyzing the startup process of Netty, you will have a new understanding of these steps. In the EchoServer example, when you enter the # bind (int port) method, AbstractBootstrap#bind () actually has multiple methods to facilitate the passing of different address parameters. The actual method called is the AbstractBootstrap#doBind (final SocketAddress localAddress) method, as shown below:

Private ChannelFuture doBind (final SocketAddress localAddress) {final ChannelFuture regFuture = initAndRegister (); final Channel channel = regFuture.channel (); if (regFuture.cause ()! = null) {return regFuture;} if (regFuture.isDone ()) {/ / At this point we know that the registration was complete and successful. ChannelPromise promise = channel.newPromise (); doBind0 (regFuture, channel, localAddress, promise); return promise;} else {/ / Registration future is almost always fulfilled already, but just in case it's not. 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) {/ / Registration on the EventLoop failed so fail the ChannelPromise directly to not cause an / / IllegalStateException once we try to access the EventLoop of the Channel. Promise.setFailure (cause);} else {/ / Registration was successful, so set the correct executor to use. / / See https://github.com/netty/netty/issues/2586 promise.registered (); doBind0 (regFuture, channel, localAddress, promise);}); return promise;}}

[line 2]: call the # initAndRegister () method to initialize and register a Channel object. Because registration is an asynchronous process, a ChannelFuture object is returned. For detailed analysis, see "initAndRegister ()".

[line 4-6]: if an exception occurs, return it directly.

[code line 9-34]: because registration is an asynchronous process, it may or may not have been completed. So the implementation code is divided into [lines 10 to 14] and [lines 15 to 36] to deal with completed and unfinished cases, respectively.

The core is in [lines 11, 29], calling the # doBind0 (final ChannelFuture regFuture, final Channel channel, final SocketAddress localAddress, final ChannelPromise promise) method, binding the port of Channel, and registering Channel with SelectionKey.

If the corresponding ChanelFuture is not completed asynchronously, the ChannelFuture#addListener (ChannelFutureListener) method is called to add the listener. After the registration is completed, the callback executes # doBind0 (...). The logic of the method

Through the doBind method, you can know that the server startup process is roughly as follows:

1. Create Channel

Enter from # doBind (final SocketAddress localAddress) to initAndRegister ():

Final ChannelFuture initAndRegister () {Channel channel = null; try {channel = channelFactory.newChannel (); init (channel);} catch (Throwable t) {if (channel! = null) {/ / channel can be null if newChannel crashed (eg SocketException ("too many open files")) channel.unsafe () .closeForcibly () / / as the Channel is not registered yet we need to force the usage of the GlobalEventExecutor return new DefaultChannelPromise (channel, GlobalEventExecutor.INSTANCE) .setFailure (t);} / as the Channel is not registered yet we need to force the usage of the GlobalEventExecutor return new DefaultChannelPromise (new FailedChannel (), GlobalEventExecutor.INSTANCE) .setFailure (t);} ChannelFuture regFuture = config () .group () .register (channel) If (regFuture.cause ()! = null) {if (channel.isRegistered ()) {channel.close ();} else {channel.unsafe (). CloseForcibly ();}} return regFuture;}

[line 4] calls the ChannelFactory#newChannel () method to create a Channel object. The ChannelFactory class inherits as follows:

You can see @ deprecated Use {@ link io.netty.channel.ChannelFactory} instead., in the ChannelFactory comment, which is just the adjustment of the package name and remains the same for the inheritance structure. Netty uses ReflectiveChannelFactory by default, and we can see the overloading method:

@ Overridepublic T newChannel () {try {return constructor.newInstance ();} catch (Throwable t) {throw new ChannelException ("Unable to create Channel from class" + constructor.getDeclaringClass (), t);}}

Obviously, as its name suggests, the Channel object instance is constructed through the reflection mechanism. Constructor is initialized in its constructor: this.constructor = clazz.getConstructor (); this clazz is supposed to be the Class object of the Channel we are going to create. So what is the Class object? Let's move on to see how channelFactory is initialized.

First, find the following code in AbstractBootstrap:

@ Deprecatedpublic B channelFactory (ChannelFactory, Object > e: attrs.entrySet ()) {@ SuppressWarnings ("unchecked") AttributeKey key = (AttributeKey) e.getKey (); channel.attr (key) .set (e.getValue ());} ChannelPipeline p = channel.pipeline (); final EventLoopGroup currentChildGroup = childGroup; final ChannelHandler currentChildHandler = childHandler; final Entry, Object > [] currentChildAttrs Synchronized (childOptions) {currentChildOptions = childOptions.entrySet () .toArray (newOptionArray (0));} synchronized (childAttrs) {currentChildAttrs = childAttrs.entrySet () .toArray (newAttrArray (0));} p.addLast (new ChannelInitializer () {@ Override public void initChannel (final Channel ch) throws Exception {final ChannelPipeline pipeline = ch.pipeline (); ChannelHandler handler = config.handler () If (handler! = null) {pipeline.addLast (handler);} ch.eventLoop () .execute (new Runnable () {@ Override public void run () {pipeline.addLast (ch (currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs) });});}

[lines 3-6]: the options returned by the options0 () method saves the collection of options that the user sets customization in EchoServer, so that ServerBootstrap sets the set of configured options to the collection of options in Channel.

[lines 8-15]: the attrs returned by the attrs0 () method saves the custom property set that the user sets in EchoServer, so that ServerBootstrap sets the configured property set to the property set of Channel.

[lines 21-28]: user-defined childOptions and childAttrs are saved through the local variables currentChildOptions and currentChildAttrs for the [line 43] ServerBootstrapAcceptor constructor.

[code 30-47]]: create a ChannelInitializer object and add it to pipeline for subsequent initialization of ChannelHandler to pipeline, including the LoggingHandler configured by the user in EchoServer and the created ServerBootstrapAcceptor object.

[line 34-37]: add the LoggingHandler of the initiator configuration to the pipeline.

[lines 39-45]: create a ServerBootstrapAcceptor object and add it to the pipeline. As can be seen from the name, ServerBootstrapAcceptor is also a ChannelHandler implementation class, which is specially used to accept new connection requests from clients and throw new requests to an event circulator. Let's not do too much analysis. We found that we were using EventLoop.execute to perform the add process. Why? Also record the problem (Problem-3)

It should be noted that when pipeline introduced the core components of Netty, he mentioned that it was a bi-directional linked list containing ChannelHandlerContext. Each context is unique to a ChannelHandler. After initialization, the ChannelPipeline contains the following structure:

3. Register for Channel

After initializing some basic configuration and properties of Channel, go back to the initAndRegister () entry method that created the Channel in the first place, and then [line 17] ChannelFuture regFuture = config (). Group (). Register (channel) immediately after initializing the Channel; obviously you enter the registration process through EventLoopGroup (the EventLoopGroup system will be explained in a later article)

In EchoServer, the initiator also sets NioEventLoopGroup through ServerBootstrap#group (), which inherits from MultithreadEventLoopGroup, so the registration process goes into the register (Channel channel) method overloaded by MultithreadEventLoopGroup, as follows:

@ Overridepublic ChannelFuture register (Channel channel) {return next () .register (channel);}

Here will call the next () method to select an EventLoop to register Channel, which actually uses something called EventExecutorChooser to choose, it actually has two ways to implement-- PowerOfTwoEventExecutorChooser and GenericEventExecutorChooser, essentially select an EventExecutor from the EventExecutor array, we here is NioEventLoop, so, what's the difference between them? (Problem-4: it will be explained in more detail in subsequent articles introducing the EventLoopGroup system, which is simply mentioned here that the essence is to take the remainder according to the length of the array, but the N-th power of 2 is more efficient. )

Then, when you come to NioEventLoop's register (channel) method, will you ask if you can't find it? Prompt NioEventLoop to inherit SingleThreadEventLoop, so the parent class method:

@ Overridepublic ChannelFuture register (Channel channel) {return register (new DefaultChannelPromise (channel, this));} @ Overridepublic ChannelFuture register (final ChannelPromise promise) {ObjectUtil.checkNotNull (promise, "promise"); promise.channel () .unsafe () .register (this, promise); return promise;}

As you can see, you first created something called ChannelPromise, which is a subclass of ChannelFuture. [line 9] calls back the register () method of Channel's Unsafe, where the first parameter is this, that is, NioEventLoop, and the second parameter is the ChannelPromise you just created.

Click the AbstractUnsafe#register (EventLoop eventLoop, final ChannelPromise promise) method to enter, and the code is as follows:

Public final void register (EventLoop eventLoop, final ChannelPromise promise) {if (eventLoop = = null) {throw new NullPointerException ("eventLoop");} if (isRegistered ()) {promise.setFailure (new IllegalStateException ("registered to an event loop already")); return } if (! isCompatible (eventLoop)) {promise.setFailure (new IllegalStateException ("incompatible event loop type:" + eventLoop.getClass (). GetName ()); return;} AbstractChannel.this.eventLoop = eventLoop; if (eventLoop.inEventLoop ()) {register0 (promise) } else {try {eventLoop.execute (new Runnable () {@ Override public void run () {register0 (promise);}}) } catch (Throwable t) {logger.warn ("Force-closing a channel whose registration task was not accepted by an event loop: {}", AbstractChannel.this, t); closeForcibly (); closeFuture.setClosed (); safeSetFailure (promise, t);}

[code line 15] this line of code sets the eventLoop property of Channel. The code in front of this line is mainly used to verify that the incoming eventLoop parameter is not null, to verify that it has been registered, and to verify that the Channel and eventLoop types match.

[code 18, 24] next, trace to the AbstractUnsafe#register0 (ChannelPromise promise) method:

Private void register0 (ChannelPromise promise) {try {/ / check if the channel is still open as it could be closed in the mean time when the register / / call was outside of the eventLoop if (! promise.setUncancellable () | |! ensureOpen (promise)) {return;} boolean firstRegistration = neverRegistered; doRegister (); neverRegistered = false; registered = true; / / Ensure we call handlerAdded (...) Before we actually notify the promise. This is needed as the / / user may already fire events through the pipeline in the ChannelFutureListener. Pipeline.invokeHandlerAddedIfNeeded (); safeSetSuccess (promise); pipeline.fireChannelRegistered (); / / Only fire a channelActive if the channel has never been registered. This prevents firing / / multiple channel actives if the channel is deregistered and re-registered. If (isActive ()) {if (firstRegistration) {pipeline.fireChannelActive ();} else if (config (). IsAutoRead ()) {/ / This channel was registered before and autoRead () is set. This means we need to begin read / / again so that we process inbound data. / See https://github.com/netty/netty/issues/4805 beginRead ();}} catch (Throwable t) {/ / Close the channel directly to avoid FD leak. CloseForcibly (); closeFuture.setClosed (); safeSetFailure (promise, t);}}

[line 9] enter the AbstractNioChannel#doRegister () method:

Protected void doRegister () throws Exception {boolean selected = false; for (;;) {try {selectionKey = javaChannel (). Register (eventLoop (). UnwrappedSelector (), 0, this); return } catch (CancelledKeyException e) {if (! selected) {/ / Force the Selector to select now as the "canceled" SelectionKey may still be / / cached and not removed because no Select.select (..) Operation was called yet. EventLoop (). SelectNow (); selected = true;} else {/ / We forced a select operation on the selector before but the SelectionKey is still cached / / for whatever reason. JDK bug? Throw e;}}

[code line 5] A key line of code that binds the Java native NIO Selector to the Channel object (ServerSocketChannel) of the Java native NIO, and binds the Channel of the current Netty to the SelectionKey in the form of attachment:

Call the # unwrappedSelector () method, which returns the Java native NIO Selector object, and each NioEventLoop corresponds uniquely to the Selector.

Call the SelectableChannel#register (Selector sel, int ops, Object att) method to register the Channel object of the Java native NIO to the NIO Selector object.

Through the above analysis of the registration channel source code, the sequence diagram of the summary process is as follows:

4. Bind port

After registering Channel, I finally go back to the AbstractBootstrap#doBind () method to analyze the port binding logic of Channel. The code for entering doBind0 is as follows:

Private static void doBind0 (final ChannelFuture regFuture, final Channel channel, final SocketAddress localAddress, final ChannelPromise promise) {/ / This method is invoked before channelRegistered () is triggered. Give user handlers a chance to set up / / the pipeline in its channelRegistered () implementation. Channel.eventLoop () .execute (new Runnable () {@ Override public void run () {if (regFuture.isSuccess ()) {channel.bind (localAddress, promise) .addListener (ChannelFutureListener.CLOSE_ON_FAILURE);} else {promise.setFailure (regFuture.cause ());});}

[line 7]: under the condition that Channel is registered successfully, call EventLoop to execute the port binding logic of Channel. However, in fact, the current thread is already the same thread as EventLoop, so why do you want to do this? The answer is in the English notes [lines 5 to 6], which are remembered here as a question (Problem-5).

[code line 11]: enter AbstractChannel#bind (SocketAddress localAddress, ChannelPromise promise), and immediately return asynchronously and add ChannelFutureListener.CLOSE_ON_FAILURE listening events.

[code line 13]: if the operation before binding the port is not successful, the port binding operation cannot be performed naturally, and the cause of the exception can be recorded through promise.

The AbstractChannel#bind (SocketAddress localAddress, ChannelPromise promise) method is as follows:

Public ChannelFuture bind (SocketAddress localAddress, ChannelPromise promise) {return pipeline.bind (localAddress, promise);}

Pipeline is the DefaultChannelPipeline created when you created the channel. Enter this method:

Public final ChannelFuture bind (SocketAddress localAddress, ChannelPromise promise) {return tail.bind (localAddress, promise);}

[draw an internal structure of DefaultChannelPipeline when analyzing the initialization process, so that you can easily enter a series of bind methods of DefaultChannelPipeline after analysis.]

First, tail represents TailContext and enters the AbstractChannelHandlerContext# bind (final SocketAddress localAddress, final ChannelPromise promise) method:

Public ChannelFuture bind (final SocketAddress localAddress, final ChannelPromise promise) {/ / omit part of the code final AbstractChannelHandlerContext next = findContextOutbound (MASK_BIND); EventExecutor executor = next.executor (); if (executor.inEventLoop ()) {next.invokeBind (localAddress, promise) } else {safeExecute (executor, new Runnable () {@ Override public void run () {next.invokeBind (localAddress, promise);}, promise, null);} return promise;}

[line 3]: the findContextOutbound method mainly executes ctx = ctx.prev;, then the resulting next is the context that binds the LoggingHandler

[code line 6]: enter the invokeBind (localAddress, promise) method and directly execute LoggingHandler#bind (this, localAddress, promise). The method after entering is as follows:

Public void bind (ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise) throws Exception {if (logger.isEnabled (internalLevel)) {logger.log (internalLevel, format (ctx, "BIND", localAddress);} ctx.bind (localAddress, promise);}

After the log basic level of LoggingHandler is set to the default INFO, the information of the binding operation is printed. Next, continue to loop to the AbstractChannelHandlerContext# bind (final SocketAddress localAddress, final ChannelPromise promise) method to execute ctx = ctx.prev to take the HeadContext into the bind method:

Public void bind (ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise) {unsafe.bind (localAddress, promise);}

Walk around and finally jump out of the pipeline wheel and return to the AbstractUnsafe#bind (final SocketAddress localAddress, final ChannelPromise promise) method, the port binding logic of Channel. The code is as follows:

Public final void bind (final SocketAddress localAddress, final ChannelPromise promise) {/ / is omitted here. Boolean wasActive = isActive (); try {doBind (localAddress);} catch (Throwable t) {safeSetFailure (promise, t); closeIfClosed (); return;} / omitted here.}

The doBind entry method for doing practical things is as follows:

At this point, the Java native NIO ServerSocketChannel on the server is finally bound to the port.

III. Problem induction

Problem-1: how is the algorithm for allocating ID to channel in the AbstractChannel constructor in the Channel creation process implemented?

Problem-2: what is the function of AbstractChannel inner class AbstractUnsafe?

Problem-3: the process of adding ServerBootstrapAcceptor to pipeline in the initialization channel process is performed through EventLoop.execute. Why?

Problem-4: what are the differences and optimization principles between PowerOfTwoEventExecutorChooser and GenericEventExecutorChooser in the registration channel process?

Problem-5: EventLoop is called in the binding port process to execute the port binding logic of Channel. However, in fact, the current thread is already the same thread as EventLoop, so why do you want to do this?

At this point, the study of "what is the Netty server startup source code" is over. I hope to be able to solve your doubts. The collocation of theory and practice can better help you learn, go and try it! If you want to continue to learn more related knowledge, please continue to follow the website, the editor will continue to work hard to bring you more practical articles!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.