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

This article is about how Apache Tomcat handles requests concurrently. The editor thinks it is very practical, so share it with you as a reference and follow the editor to have a look.

Introduction

As a commonly used http protocol server, tomcat is widely used. Tomcat also follows the Servelt protocol, which decouples the server from the real service logic code. Each only needs to pay attention to the Servlet protocol.

How does tomcat act as a high-performance server? Do you have the same question?

How does tomcat receive network requests?

How to achieve high-performance http protocol server?

Tomcat has used the NIO non-blocking io model since 8.0, which improves the throughput. The source code of this article is tomcat 9.0.48.

Receive Socket request

Org.apache.tomcat.util.net.Acceptor implements the Runnable interface to listen to socket in an endless loop in a single thread.

The initialization and startup of the thread is performed in the method org.apache.tomcat.util.net.AbstractEndpoint#startAcceptorThread

A very important attribute, org.apache.tomcat.util.net.AbstractEndpoint;, also implements the run method, which mainly has the following functions:

Limit on the maximum number of connections requested: up to 8: 1024; please note that when the maximum number of connections is reached, the bottom layer of the operating system will still receive client connections, but the user layer will no longer receive them

Get socketChannel

Public void run () {int errorDelay = 0; try {/ / Loop until we receive a shutdown command while (! stopCalled) {. If (stopCalled) {break;} state = AcceptorState.RUNNING; try {/ / if we have reached max connections, wait / / if the connection exceeds 8cm 1024, the thread blocks waiting Whether the sharing lock is implemented using the org.apache.tomcat.util.threads.LimitLatch class (AbstractQueuedSynchronizer is internally implemented) / / Please note that the bottom layer of the operating system will still receive client connections after reaching the maximum number of connections, but the user layer will no longer receive them. Endpoint.countUpOrAwaitConnection (); / / Endpoint might have been paused while waiting for latch / / If that is the case, don't accept new connections if (endpoint.isPaused ()) {continue;} U socket = null Try {/ / Accept the next incoming connection from the server / / socket / / abstract methods, different endPoint have different implementation methods. NioEndPoint as an example, the implementation method is serverSock.accept (). This method mainly depends on whether the serverSock instantiation is blocking and the accept method is blocking. If there is no socket link, it will be null socket = endpoint.serverSocketAccept ();} catch (Exception ioe) {/ / We didn't get a socket endpoint.countDownConnection () If (endpoint.isRunning ()) {/ / Introduce delay if necessary errorDelay = handleExceptionWithDelay (errorDelay); / / re-throw throw ioe;} else {break }} / / Successful accept, reset the error delay errorDelay = 0 / / Configure the socket if (! stopCalled & &! endpoint.isPaused ()) {/ / setSocketOptions () will hand the socket off to / / an appropriate processor if successful / / endPoint class abstract methods, different endPoint have different implementations. Process the acquired socketChannel link. If the socket link can be handled normally, the method will return true, otherwise it will be false if (! endpoint.setSocketOptions (socket)) {endpoint.closeSocket (socket);}} else {endpoint.destroySocket (socket) }} catch (Throwable t) {...} finally {stopLatch.countDown ();} state = AcceptorState.ENDED;}

Let's take a look at the specific implementation of the org.apache.tomcat.util.net.NioEndpoint#setSocketOptions method (NioEndpoint as an example)

The main things to do in this method:

Create NioChannel

Set socket to non-blocking

Add socket to the queue of Poller

Protected boolean setSocketOptions (SocketChannel socket) {NioSocketWrapper socketWrapper = null; try {/ / Allocate channel and wrapper / / give priority to the existing cache nioChannel NioChannel channel = null; if (nioChannels! = null) {channel = nioChannels.pop () } if (channel = = null) {SocketBufferHandler bufhandler = new SocketBufferHandler (socketProperties.getAppReadBufSize (), socketProperties.getAppWriteBufSize (), socketProperties.getDirectBuffer ()); if (isSSLEnabled ()) {channel = new SecureNioChannel (bufhandler, this) } else {channel = new NioChannel (bufhandler);}} / / Packaging nioEndpoint and NioChannel NioSocketWrapper newWrapper = new NioSocketWrapper (channel, this); channel.reset (socket, newWrapper); connections.put (socket, newWrapper); socketWrapper = newWrapper / / Set socket properties / / Disable blocking, polling will be used / / sets the socket of the current link to non-blocking socket.configureBlocking (false); if (getUnixDomainSocketPath () = = null) {socketProperties.setProperties (socket.socket ());} socketWrapper.setReadTimeout (getConnectionTimeout ()); socketWrapper.setWriteTimeout (getConnectionTimeout ()) SocketWrapper.setKeepAliveLeft (NioEndpoint.this.getMaxKeepAliveRequests ()); / / register the wrapped nioChannel with nioEndpoint, register with Poller, add the corresponding socket wrapper class to the queue of Poller, and wake up selector poller.register (socketWrapper); return true;} catch (Throwable t) {ExceptionUtils.handleThrowable (t) Try {log.error (sm.getString ("endpoint.socketOptionsError"), t);} catch (Throwable tt) {ExceptionUtils.handleThrowable (tt);} if (socketWrapper = = null) {destroySocket (socket);}} / / Tell to close the socket if needed return false } Socket request polling

The previous section received the socket request, wrapped it, added socket to the queue of Poller, and may have awakened Selector. In this section, let's take a look at how Poller polls socket.

First of all, org.apache.tomcat.util.net.NioEndpoint.Poller also implements the Runnable interface, which is a thread that can be started separately.

Initialization and startup are performed in org.apache.tomcat.util.net.NioEndpoint#startInternal

Important attributes:

Java.nio.channels.Selector: the poller is started when the Poller object is initialized

SynchronizedQueue: synchronized event queue

Let's take a look at the specific processing logic, the source code of the run method.

Public void run () {/ / Loop until destroy () is called while (true) {boolean hasEvents = false Try {if (! close) {/ / pull to the SynchronizedQueue event queue to see if there is already an event. If so, return true / / if the event is pulled from the queue (that is, the NioSocketWrapper was encapsulated as PollerEvent and added to the secondary queue in the previous step), register the socketChannel on the Selector Mark it as SelectionKey.OP_READ and add the handler function attachment (/ / NioSocketWrapper when Accetpor is added to Poller) hasEvents = events () If (wakeupCounter.getAndSet (- 1) > 0) {/ If we are here, means we have other stuff to do / / Do a non blocking select keyCount = selector.selectNow () } else {keyCount = selector.select (selectorTimeout);} wakeupCounter.set (0);} if (close) {events (); timeout (0, false) Try {selector.close ();} catch (IOException ioe) {log.error (sm.getString ("endpoint.nio.selectorCloseFail"), ioe);} break } / / Either we timed out or we woke up, process events first if (keyCount = = 0) {hasEvents = (hasEvents | events ());}} catch (Throwable x) {ExceptionUtils.handleThrowable (x) Log.error (sm.getString ("endpoint.nio.selectorLoopError"), x); continue;} Iterator iterator = keyCount > 0? Selector.selectedKeys (). Iterator (): null; / / Walk through the collection of ready keys and dispatch / / any active event / / selector polls to get registered events. If any events are ready, you can get the corresponding event while (iterator! = null & & iterator.hasNext ()) {SelectionKey sk = iterator.next () through the selectKeys method. / / after getting the event, delete the event from the iterator to prevent it from repeatedly polling iterator.remove (); / / get the handler of the event, this attachment is registered in the event () method, and the subsequent handling of this event is handed over to the wrapper to handle NioSocketWrapper socketWrapper = (NioSocketWrapper) sk.attachment () / Attachment may be null if another thread has called / / cancelledKey () if (socketWrapper! = null) {processKey (sk, socketWrapper);}} / / Process timeouts timeout (keyCount,hasEvents) } getStopLatch () .countDown ();

Here, there is a very important method, org.apache.tomcat.util.net.NioEndpoint.Poller#events (), which fetches the available socket received by Acceptor from the event queue of Poller and registers it with Selector

/ * Processes events in the event queue of the Poller. * * @ return true if some events were processed, * false if queue was empty * / public boolean events () {boolean result = false; PollerEvent pe = null / / if Acceptor adds socket to the queue, then the events.poll () method can get the corresponding event, otherwise it will return false for (int I = 0, size = events.size (); I < size & & (pe = events.poll ())! = null; iTunes +) {result = true; NioSocketWrapper socketWrapper = pe.getSocketWrapper () SocketChannel sc = socketWrapper.getSocket (). GetIOChannel (); int interestOps = pe.getInterestOps (); if (sc = = null) {log.warn (sm.getString ("endpoint.nio.nullSocketChannel")); socketWrapper.close () } else if (interestOps = = OP_REGISTER) {/ / if it is an event that Acceptor has just added to the queue, then the ops at this time is OP_REGISTER try {, / / register the secondary socket on the selector and mark it as the OP_READ event Add the handler function socketWrapper sc.register (getSelector (), SelectionKey.OP_READ, socketWrapper) when the event is triggered } catch (Exception x) {log.error (sm.getString ("endpoint.nio.registerFail"), x);}} else {/ /? The logic here is not clear under what circumstances final SelectionKey key = sc.keyFor (getSelector ()); if (key = = null) {/ / The key was cancelled (e.g. Due to socket closure) / / and removed from the selector while it was being / / processed. Count down the connections at this point / / since it won't have been counted down when the socket / / closed. SocketWrapper.close ();} else {final NioSocketWrapper attachment = (NioSocketWrapper) key.attachment (); if (attachment! = null) {/ / We are registering the key to start with, reset the fairness counter. Try {int ops = key.interestOps () | interestOps; attachment.interestOps (ops); key.interestOps (ops);} catch (CancelledKeyException ckx) {cancelledKey (key, socketWrapper) }} else {cancelledKey (key, socketWrapper);} if (running & &! paused & & eventCache! = null) {pe.reset () EventCache.push (pe);}} return result;}

Another important method is org.apache.tomcat.util.net.NioEndpoint.Poller#processKey. The previous method is to get event and register to selector. This method is to prepare the event through the data obtained by Selector, and begin to encapsulate it into the corresponding business processing thread SocketProcessorBase, and throw it into the thread pool to start processing.

Protected void processKey (SelectionKey sk, NioSocketWrapper socketWrapper) {try {if (close) {cancelledKey (sk, socketWrapper) } else if (sk.isValid ()) {if (sk.isReadable () | | sk.isWritable ()) {if (socketWrapper.getSendfileData ()! = null) {processSendfile (sk, socketWrapper, false) } else {unreg (sk, socketWrapper, sk.readyOps ()); boolean closeSocket = false / / Read goes before write if (sk.isReadable ()) {/ / if it is an asynchronous operation here This is where if (socketWrapper.readOperation! = null) {if (! socketWrapper.readOperation.process ()) {closeSocket = true }} else if (socketWrapper.readBlocking) {/ / readBlocking defaults to false synchronized (socketWrapper.readLock) {socketWrapper.readBlocking = false SocketWrapper.readLock.notify ();}} else if (! processSocket (socketWrapper, SocketEvent.OPEN_READ, true)) {/ / handles normal events, and the processSocket here is about to start processing requests. / / encapsulate the corresponding event into the corresponding thread, and then hand it over to the thread pool to handle the formal request business closeSocket = true }} if (! closeSocket & & sk.isWritable ()) {if (socketWrapper.writeOperation! = null) {if (! socketWrapper.writeOperation.process ()) { CloseSocket = true }} else if (socketWrapper.writeBlocking) {synchronized (socketWrapper.writeLock) {socketWrapper.writeBlocking = false; socketWrapper.writeLock.notify () }} else if (! processSocket (socketWrapper, SocketEvent.OPEN_WRITE, true)) {closeSocket = true }} if (closeSocket) {cancelledKey (sk, socketWrapper) }} else {/ / Invalid key cancelledKey (sk, socketWrapper);}} catch (CancelledKeyException ckx) {cancelledKey (sk, socketWrapper) } catch (Throwable t) {ExceptionUtils.handleThrowable (t); log.error (sm.getString ("endpoint.nio.keyProcessingError"), t);}} request specific processing

In the previous step, Selector obtained the ready request socket, then encapsulated the data according to the trigger processing functions registered by socket, threw it into the thread pool, and started specific business logic processing. This section starts from the worker thread encapsulation. Org.apache.tomcat.util.net.SocketProcessorBase is the abstract class of the worker thread class, which implements the Runnable interface. Different Endpoint implements specific processing logic. This section takes NioEndpoint as an example.

The following is the source code of the org.apache.tomcat.util.net.AbstractEndpoint#processSocket method

/ * Process the given SocketWrapper with the given status. Used to trigger * processing as if the Poller (for those endpoints that have one) * selected the socket. * * @ param socketWrapper The socket wrapper to process * @ param event The socket event to be processed * @ param dispatch Should the processing be performed on a new * container thread * * @ return if processing was triggered successfully * / public boolean processSocket (SocketWrapperBase socketWrapper, SocketEvent event Boolean dispatch) {try {if (socketWrapper = = null) {return false } / / give priority to the existing thread SocketProcessorBase sc = null; if (processorCache! = null) {sc = processorCache.pop ();} if (sc = = null) {sc = createSocketProcessor (socketWrapper, event) } else {sc.reset (socketWrapper, event);} / / get the thread pool. The initialization of the thread pool is to create / / tomcat using a custom org.apache.tomcat.util.threads.TaskQueue before the two separate threads of Acceptor and Poller are started. This tomcat is also adapted to the development of 10 core threads, with a maximum of 200 threads Executor executor = getExecutor () If (dispatch & & executor! = null) {executor.execute (sc);} else {sc.run ();}} catch (RejectedExecutionException ree) {getLog () .warn (sm.getString ("endpoint.executor.fail", socketWrapper), ree); return false } catch (Throwable t) {ExceptionUtils.handleThrowable (t); / / This means we got an OOM or similar creating a thread, or that / / the pool and its queue are full getLog (). Error (sm.getString ("endpoint.process.fail"), t); return false;} return true;}

The above method is to get the thread that processes the business logic, the SocketProcessorBase,NioEndpoint inner class org.apache.tomcat.util.net.NioEndpoint.SocketProcessor inherits this abstract class, that is, the specific business processing logic, in the org.apache.tomcat.util.net.NioEndpoint.SocketProcessor#doRun method, and finally calls our Servlet.

Protected void doRun () {/ * * Do not cache and re-use the value of socketWrapper.getSocket () in * this method. If the socket closes the value will be updated to * CLOSED_NIO_CHANNEL and the previous value potentially re-used for * a new connection. That can result in a stale cached value which * in turn can result in unintentionally closing currently active * connections. * / Poller poller = NioEndpoint.this.poller; if (poller = = null) {socketWrapper.close (); return;} try {int handshake =-1 Try {/ / handshake related judgment logic.} catch (IOException x) {...} / / three-way handshake succeeded if (handshake = = 0) { SocketState state = SocketState.OPEN / / Process the request from this socket / / event is SocketEvent.OPEN_READ, and this variable is the org.apache.tomcat.util.net.NioEndpoint.Poller#processKey method assignment if (event = = null) {state = getHandler () .process (socketWrapper, SocketEvent.OPEN_READ) } else {/ / here begins to formally process the request state = getHandler (). Process (socketWrapper, event);} if (state = = SocketState.CLOSED) {poller.cancelledKey (getSelectionKey (), socketWrapper) }} else if (handshake = =-1) {getHandler () .process (socketWrapper, SocketEvent.CONNECT_FAIL); poller.cancelledKey (getSelectionKey (), socketWrapper);} else if (handshake = = SelectionKey.OP_READ) {socketWrapper.registerReadInterest () } else if (handshake = = SelectionKey.OP_WRITE) {socketWrapper.registerWriteInterest ();}} catch (CancelledKeyException cx) {poller.cancelledKey (getSelectionKey (), socketWrapper);} catch (VirtualMachineError vme) {ExceptionUtils.handleThrowable (vme) } catch (Throwable t) {log.error (sm.getString ("endpoint.processing.fail"), t); poller.cancelledKey (getSelectionKey (), socketWrapper);} finally {socketWrapper = null; event = null / / return to cache if (running & &! paused & & processorCache! = null) {processorCache.push (this);} Summary

How does Tomcat receive network requests?

Use java nio's synchronous non-blocking to monitor the network.

Initialize network snooping and SSL in org.apache.tomcat.util.net.AbstractEndpoint#bindWithCleanup

{.... ServerSock = ServerSocketChannel.open (); socketProperties.setProperties (serverSock.socket ()); InetSocketAddress addr = new InetSocketAddress (getAddress (), getPortWithOffset ()) / / when the number of connections at the application level reaches the maximum, the operating system can continue to receive connections, so the maximum number of connections that the operating system can continue to receive is this queue length, which can be configured through the acceptCount parameter. The default is 100 serverSock.bind (addr, getAcceptCount ());} serverSock.configureBlocking (true); / / mimic APR behavior

Thread pool, connection limiter, Poller thread, Acceptor thread initializing business processing in org.apache.tomcat.util.net.NioEndpoint#startInternal

How to achieve high-performance http protocol server?

Tomcat splits the receiving connection, detecting Iamp O events and processing requests, and uses threads of different sizes to do the corresponding things, which is why tomcat can process requests with high concurrency. Don't let threads block, try to keep CPU busy.

How is it designed?

Separate different processing logic through interfaces, abstract classes, etc., and perform their own duties.

Org.apache.tomcat.util.net.NioEndpoint.Poller: java.nio.channels.Selector is referenced, and there is an event queue inside. This is where listening for Icano events is done.

Org.apache.tomcat.util.net.NioEndpoint.NioSocketWrapper

Org.apache.tomcat.util.net.NioEndpoint.SocketProcessor: the thread class that specifically processes the request

The detection and processing logic of org.apache.tomcat.util.net.AbstractEndpoint:I/O events is in the implementation class of this class. Using the template method, different protocols have different implementation methods. NioEndpoint/Nio2Endpoint/AprEndpoint

Thank you for reading! This is the end of this article on "how to handle requests with high concurrency in Apache Tomcat". I hope the above content can be of some help to you, so that you can learn more knowledge. if you think the article is good, you can share it for more people to see!

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.