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

This article mainly introduces the example analysis of the thread model in Netty, MINA and Twisted, which has a certain reference value, and interested friends can refer to it. I hope you will gain a lot after reading this article.

To develop a high-performance TCP server, it is important to be familiar with the threading model of the framework you are using. MINA, Netty, Twisted themselves are high-performance network frameworks, if coupled with efficient code, we can achieve a high-end server. But without understanding their threading model, it is difficult to write high-performance code. No matter how efficient the framework itself is and how poorly the program is written, the overall performance of the server will not be too high. Just like a computer, no matter how good the CPU is, the slow heat dissipation of the hard disk with small memory will not be too high.

Students who have played with Android development will know that there is a very important thread in Android applications: the UI thread (the main thread). The UI thread is responsible for displaying an Android interface and interacting with the user. Some of the methods of Activity, such as onCreate, onStop, and onDestroy, are run in UI threads. But one thing to be careful about when writing Activity code is that blocking or time-consuming tasks should never be written in these methods. If written in these methods, the UI thread will be blocked, resulting in an unresponsive user interface and a poor experience. So in Android development, time-consuming or blocked tasks will be done by separate threads.

There is also a very important thread in MINA, Netty, and Twisted: I thread.

The traditional TCP server implemented by BIO, especially for TCP persistent connections, usually starts a thread for each connection, and the thread is also a resource of the operating system, so it is difficult to achieve high performance and high concurrency. In the TCP server implemented by asynchronous IO, because the IO operations are asynchronous, one thread or a small number of threads can be used to handle a large number of connected IO operations, so only a small number of IO threads can achieve high concurrency of the server.

In the process of network programming, there are usually some business logic that is time-consuming and blocking, such as database operation. If the network is not good, and the database performance is poor, the SQL is not optimized enough, and the amount of data is large, a SQL may be executed for a long time. Because the number of IO threads themselves is small, there is usually only one or more, and if this time-consuming blocking code runs in IO threads, other things for IO threads, such as network read and write, cannot be done, affecting IO performance as well as the performance of the server as a whole.

Therefore, whether you are using MINA, Netty, or Twisted, if you have a time-consuming task, you should never run it in an IO thread, but open another thread to deal with it.

1. MINA

In MINA, there are three very important threads: Acceptor thread, Connector thread, and I _ processor thread.

Acceptor thread: this thread is used by the TCP server to receive a new connection and assign the connection to the IO O processor thread, which handles the IO operation. One Acceptor thread is created per NioSocketAcceptor, and the number of threads is not configurable.

Connector thread: used to process the TCP client connecting to the server and assigning the connection to the IO O processor thread, which handles the IO operation. One Connector thread is created per NioSocketConnector, and the number of threads is not configurable.

I processor thread O processor thread: the I processor thread O operations used to handle TCP connections, such as read, write. The number of threads in the NioSocketAcceptor O processor thread can be configured through the NioSocketAcceptor or NioSocketConnector constructor, which defaults to + 1 CPU cores.

Since this article focuses on the threading model of the TCP server, there is nothing wrong with Connector thread. The following is the flow of Acceptor thread and Imax O processor thread to handle TCP connections:

The TCP server of MINA contains an Acceptor thread and a number of Acceptor thread O processor thread. When a new client connects to the server, the Acceptor thread first acquires the connection and assigns the connection to one of the multiple Acceptor thread O servers. When the client sends data to the server, the corresponding processor thread reads the data and executes the IoFilter and IoHandle in the server.

Due to the limited number of IO O processor thread itself, there are usually only a few, but it has to handle thousands of connected IO operations, including read, write, protocol encoding and decoding, various Filter and business logic in IoHandle, especially business logic, such as IoHandle's messageReceived. If there are time-consuming and blocking tasks, such as querying the database, it will block I IO O processor thread and prevent other IO events from being processed in a timely manner. Server performance is degraded.

To solve this problem, an ExecutorFilter is provided in MINA to put in another thread the business logic that will block the I processor thread for a long time, so that it does not block the O processor thread and does not affect the IO operation. ExecutorFilter contains a thread pool, the default is OrderedThreadPoolExecutor, which ensures that multiple events for the same connection are executed sequentially, and you can also use UnorderedThreadPoolExecutor, which does not guarantee the order of events for the same connection and may be executed concurrently. You can choose between the two according to your needs.

Public class TcpServer {

Public static void main (String [] args) throws IOException {

IoAcceptor acceptor = new NioSocketAcceptor (4); / / configure the number of O processor thread threads

Acceptor.getFilterChain () .addLast ("codec", new ProtocolCodecFilter (new TextLineCodecFactory ())

Acceptor.getFilterChain () .addLast ("executor", new ExecutorFilter ()); / / get the business logic in TcpServerHandle to the thread pool of ExecutorFilter for execution

Acceptor.setHandler (new TcpServerHandle ())

Acceptor.bind (new InetSocketAddress (8080))

}

}

Class TcpServerHandle extends IoHandlerAdapter {

@ Override

Public void messageReceived (IoSession session, Object message)

Throws Exception {

/ / suppose there is a perverted SQL to execute for 3 seconds

Thread.sleep (3000)

}

}

2. Netty

When Netty's TCP server starts, two NioEventLoopGroup are created, one boss and one worker:

EventLoopGroup bossGroup = new NioEventLoopGroup ()

EventLoopGroup workerGroup = new NioEventLoopGroup ()

NioEventLoopGroup is actually a thread group, and you can set the number of threads by constructor. The default is the number of CPU cores * 2. Boss is used by the server to receive a new TCP connection, and the boss thread receives the new connection and registers the connection with the worker thread. Worker threads are used to handle IO operations, such as read, write.

The boss thread in Netty is similar to the Acceptor thread,work thread in MINA and the I _ processor thread in MINA. The difference is that MINA's Acceptor thread is a single thread, while Netty's boss is a thread group. In fact, the ServerBootstrap of Netty can listen to multiple port numbers. if you listen to only one port number, only one boss thread is needed. It is recommended to set the number of threads in bossGroup to 1.

EventLoopGroup bossGroup = new NioEventLoopGroup (1)

When a new TCP client connects to the server, the boss thread receives the connection, and then registers the connection with the worker thread. When the client sends data to the server, the worker thread receives the data and executes the ChannelHandler in the ChannelPipeline.

Similar to MINA's processor thread O processor thread, Netty has a small number of worker threads and has to deal with IO events in real time. If time-consuming business logic blocks the worker thread, such as executing a time-consuming database query in channelRead, it will cause the IO operation to fail, and the overall performance of the server will be degraded.

In Netty 3, there is an ExecutionHandler, which is an implementation class of ChannelHandler that handles time-consuming business logic, similar to MINA's ExecutorFilter, but is removed in Netty 4. So ExecutionHandler is no longer introduced here.

EventExecutorGroup can be used in Netty 4 to handle time-consuming business logic:

Public class TcpServer {

Public static void main (String [] args) throws InterruptedException {

EventLoopGroup bossGroup = new NioEventLoopGroup (1); / / the server listens for a port number. It is recommended to set the number of Boss threads to 1.

EventLoopGroup workerGroup = new NioEventLoopGroup (4); / / set the number of worker threads to 4

Try {

ServerBootstrap b = new ServerBootstrap ()

B.group (bossGroup, workerGroup)

.channel (NioServerSocketChannel.class)

.childHandler (new ChannelInitializer () {

/ / create a thread group of 16 threads to handle time-consuming business logic

Private EventExecutorGroup group = new DefaultEventExecutorGroup (16)

@ Override

Public void initChannel (SocketChannel ch) throws Exception {

ChannelPipeline pipeline = ch.pipeline ()

Pipeline.addLast (new LineBasedFrameDecoder (80))

Pipeline.addLast (new StringDecoder (CharsetUtil.UTF_8))

/ / put the business logic in TcpServerHandler into the EventExecutorGroup thread group to execute

Pipeline.addLast (group, new TcpServerHandler ())

}

});

ChannelFuture f = b.bind (8080) .sync ()

F.channel (). CloseFuture (). Sync ()

} finally {

WorkerGroup.shutdownGracefully ()

BossGroup.shutdownGracefully ()

}

}

}

Class TcpServerHandler extends ChannelInboundHandlerAdapter {

@ Override

Public void channelRead (ChannelHandlerContext ctx, Object msg) throws InterruptedException {

/ / suppose there is a perverted SQL to execute for 3 seconds

Thread.sleep (3000)

}

}

3.Twisted

Twisted's threading model is the simplest and roughest: single thread, or reactor thread. That is, all IO operations, encoding and decoding, business logic, and so on are performed in a single thread. In fact, even if it is a single thread, its performance is very high and can handle a large number of connections at the same time. When programming in a single-threaded environment, there is no need to consider thread safety. However, single threading brings a problem, that is, time-consuming business logic, if running in a reactor thread, then other things, such as network IO, can not continue to do until the reactor thread is idle, which will affect the performance of the server.

The following code executes the time-consuming business logic in a separate thread pool via reactor.callInThread instead of running in the reactor thread. This will not affect the network IO of the reactor thread. You can set the number of threads in this thread pool through reactor.suggestThreadPoolSize.

#-*-coding:utf-8-*-

Import time

From twisted.internet.protocol import Protocol

From twisted.internet.protocol import Factory

From twisted.internet import reactor

# time-consuming and blocking business logic

Def logic (data):

Print data

Time.sleep (3) # suppose there is a perverted SQL to execute for 3 seconds

Class TcpServerHandle (Protocol):

Def dataReceived (self, data):

Reactor.callInThread (logic, data) # runs logic (data) time-consuming tasks in the thread pool, not in reactor threads

Reactor.suggestThreadPoolSize (8) # sets the number of threads in the thread pool to 8

Factory = Factory ()

Factory.protocol = TcpServerHandle

Reactor.listenTCP (8080, factory)

Reactor.run ()

Because of the single-threaded design of Twisted's reactor, much of its code is not thread-safe. So code executing in non-reactor threads needs to pay attention to thread safety issues. Transport.write, for example, is not thread safe. However, the reactor.callFromThread method can be called from a non-reactor thread, which, in contrast to callInThread, runs a function from another thread into the reactor thread. It is important to note, however, that functions called by reactor.callFromThread will also block the reactor thread and affect IO if it is running in a reactor thread.

#-*-coding:utf-8-*-

Import time

From twisted.internet.protocol import Protocol

From twisted.internet.protocol import Factory

From twisted.internet import reactor

# non-thread-safe code

Def notThreadSafe ():

Print "notThreadSafe"

# time-consuming and blocking business logic

Def logic (data):

Print data

Time.sleep (3) # suppose there is a perverted SQL to execute for 3 seconds

Reactor.callFromThread (notThreadSafe) # runs notThreadSafe () in the reactor thread

Class TcpServerHandle (Protocol):

Def dataReceived (self, data):

Reactor.callInThread (logic, data) # runs logic (data) time-consuming tasks in the thread pool, not in reactor threads

Reactor.suggestThreadPoolSize (8) # sets the number of threads in the thread pool to 8

Factory = Factory ()

Factory.protocol = TcpServerHandle

Reactor.listenTCP (8080, factory)

Reactor.run ()

In addition, many convenient functions are provided in twisted.internet.threads. For example, threads.deferToThread is used to execute a time-consuming task in a thread pool. Unlike reactor.callInThread, its return value is of type Deferred, and you can add a callback function to deal with the result (return value) of the time-consuming task.

#-*-coding:utf-8-*-

Import time

From twisted.internet.protocol import Protocol

From twisted.internet.protocol import Factory

From twisted.internet import reactor, threads

# time-consuming and blocking business logic

Def logic (data):

Print data

Time.sleep (3) # suppose there is a perverted SQL to execute for 3 seconds

Return "success"

# callback function

Def logicSuccess (result):

# result is the return value of logic function, that is, "success"

Print result

Class TcpServerHandle (Protocol):

Def dataReceived (self, data):

D = threads.deferToThread (logic, data) # put the time-consuming business logic logic (data) into the thread pool to run, and the deferToThread return type is Deferred

D.addCallback (logicSuccess) # add callback function

Reactor.suggestThreadPoolSize (8) # sets the number of threads in the thread pool to 8

Factory = Factory ()

Factory.protocol = TcpServerHandle

Reactor.listenTCP (8080, factory)

Reactor.run () Thank you for reading this article carefully. I hope the article "sample Analysis of Thread Model in Netty, MINA and Twisted" shared by the editor will be helpful to you. At the same time, I also hope you will support us and pay attention to the industry information channel. More related knowledge is waiting for you to learn!

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.