Mina、Netty、Twisted一起学(十):线程模型

释放双眼,带上耳机,听听看~!

要想开发一个高性能的TCP服务器,熟悉所使用框架的线程模型非常重要。MINA、Netty、Twisted本身都是高性能的网络框架,如果再搭配上高效率的代码,才能实现一个高大上的服务器。但是如果不了解它们的线程模型,就很难写出高性能的代码。框架本身效率再高,程序写的太差,那么服务器整体的性能也不会太高。就像一个电脑,CPU再好,内存小硬盘慢散热差,整体的性能也不会太高。

玩过Android开发的同学会知道,在Android应用中有一个非常重要线程:UI线程(即主线程)。UI线程是负责一个Android的界面显示以及和用户交互。Activity的一些方法,例如onCreate、onStop、onDestroy都是运行在UI线程中的。但是在编写Activity代码的时候有一点需要非常注意,就是绝对不能把阻塞的或者耗时的任务写在这些方法中,如果写在这些方法中,则会阻塞UI线程,导致用户操作的界面反应迟钝,体验很差。所以在Android开发中,耗时或者阻塞的任务会另外开线程去做。

同样在MINA、Netty、Twisted中,也有一个非常重要的线程:IO线程

传统的BIO实现的TCP服务器,特别对于TCP长连接,通常都要为每个连接开启一个线程,线程也是操作系统的一种资源,所以很难实现高性能高并发。而异步IO实现的TCP服务器,由于IO操作都是异步的,可以用一个线程或者少量线程来处理大量连接的IO操作,所以只需要少量的IO线程就可以实现高并发的服务器。

在网络编程过程中,通常有一些业务逻辑是比较耗时、阻塞的,例如数据库操作,如果网络不好,加上数据库性能差,SQL不够优化,数据量大,一条SQL可能会执行很久。由于IO线程本身数量就不多,通常只有一个或几个,而如果这种耗时阻塞的代码在IO线程中运行的话,IO线程的其他事情,例如网络read和write,就无法进行了,会影响IO性能以及整个服务器的性能。

所以,无论是使用MINA、Netty、Twisted,如果有耗时的任务,就绝对不能在IO线程中运行,而是要另外开启线程来处理。

MINA:

在MINA中,有三种非常重要的线程:Acceptor thread、Connector thread、I/O processor thread。

下面是官方文档的介绍:

In MINA, there are three kinds of I/O worker threads in the NIO socket implementation.
Acceptor thread accepts incoming connections, and forwards the connection to the I/O processor thread for read and write operations.
Each SocketAcceptor creates one acceptor thread. You can't configure the number of the acceptor threads.
Connector thread attempts connections to a remote peer, and forwards the succeeded connection to the I/O processor thread for read and write operations.
Each SocketConnector creates one connector thread. You can't configure the number of the connector threads, either.
I/O processor thread performs the actual read and write operation until the connection is closed.
Each SocketAcceptor or SocketConnector creates its own I/O processor thread(s). You can configure the number of the I/O processor threads. The default maximum number of the I/O processor threads is the number of CPU cores + 1.

Acceptor thread:

这个线程用于TCP服务器接收新的连接,并将连接分配到I/O processor thread,由I/O processor thread来处理IO操作。每个NioSocketAcceptor创建一个Acceptor thread,线程数量不可配置。

Connector thread:

用于处理TCP客户端连接到服务器,并将连接分配到I/O processor thread,由I/O processor thread来处理IO操作。每个NioSocketConnector创建一个Connector thread,线程数量不可配置。

I/O processor thread:

用于处理TCP连接的I/O操作,如read、write。I/O processor thread的线程数量可通过NioSocketAcceptor或NioSocketConnector构造方法来配置,默认是CPU核心数+1。

由于本文主要介绍TCP服务器的线程模型,所以就没有Connector thread什么事了。下面说下Acceptor thread和I/O processor thread处理TCP连接的流程:

MINA的TCP服务器包含一个Acceptor thread和多个I/O processor thread,当有新的客户端连接到服务器,首先会由Acceptor thread获取到这个连接,同时将这个连接分配给多个I/O processor thread中的一个线程,当客户端发送数据给服务器,对应的I/O processor thread负责读取这个数据,并执行IoFilterChain中的IoFilter以及IoHandle。

由于I/O processor thread本身数量有限,通常就那么几个,但是又要处理成千上万个连接的IO操作,包括read、write、协议的编码解码、各种Filter以及IoHandle中的业务逻辑,特别是业务逻辑,比如IoHandle的messageReceived,如果有耗时、阻塞的任务,例如查询数据库,那么就会阻塞I/O processor thread,导致无法及时处理其他IO事件,服务器性能下降。

针对这个问题,MINA中提供了一个ExecutorFilter,用于将需要执行很长时间的会阻塞I/O processor thread的业务逻辑放到另外的线程中,这样就不会阻塞I/O processor thread,不会影响IO操作。ExecutorFilter中包含一个线程池,默认是OrderedThreadPoolExecutor,这个线程池保证同一个连接的多个事件按顺序依次执行,另外还可以使用UnorderedThreadPoolExecutor,它不会保证同一连接的事件的执行顺序,并且可能会并发执行。二者之间可以根据需要来选择。


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
1public class TcpServer {
2
3    public static void main(String[] args) throws IOException {
4        IoAcceptor acceptor = new NioSocketAcceptor(4); // 配置I/O processor thread线程数量
5        acceptor.getFilterChain().addLast("codec", new ProtocolCodecFilter(new TextLineCodecFactory()));
6        acceptor.getFilterChain().addLast("executor", new ExecutorFilter()); // 将TcpServerHandle中的业务逻辑拿到ExecutorFilter的线程池中执行
7        acceptor.setHandler(new TcpServerHandle());
8        acceptor.bind(new InetSocketAddress(8080));
9    }
10
11}
12
13class TcpServerHandle extends IoHandlerAdapter {
14
15    @Override
16    public void messageReceived(IoSession session, Object message)
17            throws Exception {
18        
19        // 假设这里有个变态的SQL要执行3秒
20        Thread.sleep(3000);
21    }
22}
23

Netty:

Netty的TCP服务器启动时,会创建两个NioEventLoopGroup,一个boss,一个worker


1
2
3
1EventLoopGroup bossGroup = new NioEventLoopGroup();  
2EventLoopGroup workerGroup = new NioEventLoopGroup();
3

NioEventLoopGroup实际上是一个线程组,可以通过构造方法设置线程数量,默认为CPU核心数*2。boss用于服务器接收新的TCP连接,boss线程接收到新的连接后将连接注册到worker线程。worker线程用于处理IO操作,例如read、write。

Netty中的boss线程类似于MINA的Acceptor thread,work线程和MINA的I/O processor thread类似。不同的一点是MINA的Acceptor thread是单个线程,而Netty的boss是一个线程组。实际上Netty的ServerBootstrap可以监听多个端口号,如果只监听一个端口号,那么只需要一个boss线程即可,推荐将bossGroup的线程数量设置成1。


1
2
1EventLoopGroup bossGroup = new NioEventLoopGroup(1);
2

当有新的TCP客户端连接到服务器,将由boss线程来接收连接,然后将连接注册到worker线程,当客户端发送数据到服务器,worker线程负责接收数据,并执行ChannelPipeline中的ChannelHandler。

和MINA的I/O processor thread 类似,Netty的worker线程本身数量不多,而且要实时处理IO事件,如果有耗时的业务逻辑阻塞住worker线程,例如在channelRead中执行一个耗时的数据库查询,会导致IO操作无法进行,服务器整体性能就会下降。

在Netty 3中,存在一个ExecutionHandler,它是ChannelHandler的一个实现类,用于处理耗时的业务逻辑,类似于MINA的ExecutorFilter,但是在Netty 4中被删除了。所以这里不再介绍ExecutionHandler。

Netty 4中可以使用EventExecutorGroup来处理耗时的业务逻辑:


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
1public class TcpServer {
2
3    public static void main(String[] args) throws InterruptedException {
4        EventLoopGroup bossGroup = new NioEventLoopGroup(1); // 服务器监听一个端口号,boss线程数建议设置成1
5        EventLoopGroup workerGroup = new NioEventLoopGroup(4); // worker线程数设置成4
6        try {
7            ServerBootstrap b = new ServerBootstrap();
8            b.group(bossGroup, workerGroup)
9                    .channel(NioServerSocketChannel.class)
10                    .childHandler(new ChannelInitializer<SocketChannel>() {
11                        
12                        // 创建一个16个线程的线程组来处理耗时的业务逻辑
13                        private EventExecutorGroup group = new DefaultEventExecutorGroup(16);
14                        
15                        @Override
16                        public void initChannel(SocketChannel ch) throws Exception {
17                            ChannelPipeline pipeline = ch.pipeline();
18                            pipeline.addLast(new LineBasedFrameDecoder(80));
19                            pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
20                            
21                            // 将TcpServerHandler中的业务逻辑放到EventExecutorGroup线程组中执行
22                            pipeline.addLast(group, new TcpServerHandler());
23                        }
24                    });
25            ChannelFuture f = b.bind(8080).sync();
26            f.channel().closeFuture().sync();
27        } finally {
28            workerGroup.shutdownGracefully();
29            bossGroup.shutdownGracefully();
30        }
31    }
32
33}
34
35class TcpServerHandler extends ChannelInboundHandlerAdapter {
36
37    @Override
38    public void channelRead(ChannelHandlerContext ctx, Object msg) throws InterruptedException {
39        
40        // 假设这里有个变态的SQL要执行3秒
41        Thread.sleep(3000);
42
43    }
44}
45

Twisted:

Twisted的线程模型是最简单粗暴的:单线程,即reactor线程。也就是,所有的IO操作、编码解码、业务逻辑等都是在一个线程中执行。实际上,即使是单线程,其性能也是非常高的,可以同时处理大量的连接。在单线程的环境下编程,不需要考虑线程安全的问题。不过,单线程带来一个问题,就是耗时的业务逻辑,如果运行在reactor线程中,那么其他事情,例如网络IO,就要等到reactor线程空闲时才能继续做,会影响到服务器的性能。

下面的代码,通过reactor.callInThread将耗时的业务逻辑放到单独的线程池中执行,而不在reactor线程中运行。这样就不会影响到reactor线程的网络IO了。可以通过reactor.suggestThreadPoolSize设置这个线程池的线程数量。


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
1# -*- coding:utf-8 –*-
2
3import time
4from twisted.internet.protocol import Protocol
5from twisted.internet.protocol import Factory
6from twisted.internet import reactor
7
8# 耗时、阻塞的业务逻辑
9def logic(data):
10    print data
11    time.sleep(3) # 假设这里有个变态的SQL要执行3秒    
12
13class TcpServerHandle(Protocol):
14    
15    def dataReceived(self, data):
16        reactor.callInThread(logic, data) # 在线程池中运行logic(data)耗时任务,不在reactor线程中运行
17
18reactor.suggestThreadPoolSize(8) # 设置线程池的线程数量为8
19
20factory = Factory()
21factory.protocol = TcpServerHandle
22reactor.listenTCP(8080, factory)
23reactor.run()
24

由于Twisted的reactor的单线程设计,它的很多代码都不是线程安全的。所以在非reactor线程中执行的代码需要注意线程安全问题。例如transport.write就不是线程安全的。不过在非reactor线程中可以调用reactor.callFromThread方法,这个方法功能和callInThread相反,将一个函数从别的线程放到reactor线程中运行。不过还是要注意,reactor.callFromThread调用的函数由于运行在reactor线程中,如果运行耗时,同样会阻塞reactor线程,影响IO。


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
1# -*- coding:utf-8 –*-
2
3import time
4from twisted.internet.protocol import Protocol
5from twisted.internet.protocol import Factory
6from twisted.internet import reactor
7
8# 非线程安全的代码
9def notThreadSafe():
10    print "notThreadSafe"
11
12# 耗时、阻塞的业务逻辑
13def logic(data):
14    print data
15    time.sleep(3) # 假设这里有个变态的SQL要执行3秒
16    reactor.callFromThread(notThreadSafe) # 在reactor线程中运行notThreadSafe()
17    
18
19class TcpServerHandle(Protocol):
20    
21    def dataReceived(self, data):
22        reactor.callInThread(logic, data) # 在线程池中运行logic(data)耗时任务,不在reactor线程中运行
23
24reactor.suggestThreadPoolSize(8) # 设置线程池的线程数量为8
25
26factory = Factory()
27factory.protocol = TcpServerHandle
28reactor.listenTCP(8080, factory)
29reactor.run()
30

此外,twisted.internet.threads中提供了许多很方便的函数。例如threads.deferToThread用于将一个耗时任务放在线程池中执行,与reactor.callInThread不同的是,它的返回值是Deferred类型,可以通过添加回调函数,处理耗时任务完成后的结果(返回值)。


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
1# -*- coding:utf-8 –*-
2
3import time
4from twisted.internet.protocol import Protocol
5from twisted.internet.protocol import Factory
6from twisted.internet import reactor, threads
7
8# 耗时、阻塞的业务逻辑
9def logic(data):
10    print data
11    time.sleep(3) # 假设这里有个变态的SQL要执行3秒
12    return "success"
13
14# 回调函数
15def logicSuccess(result):
16    # result即为logic函数的返回值,即"success"
17    print result
18
19class TcpServerHandle(Protocol):
20    
21    def dataReceived(self, data):
22        d = threads.deferToThread(logic, data) # 将耗时的业务逻辑logic(data)放到线程池中运行,deferToThread返回值类型是Deferred
23        d.addCallback(logicSuccess) # 添加回调函数
24
25reactor.suggestThreadPoolSize(8) # 设置线程池的线程数量为8
26
27factory = Factory()
28factory.protocol = TcpServerHandle
29reactor.listenTCP(8080, factory)
30reactor.run()
31

 

MINA、Netty、Twisted一起学系列

MINA、Netty、Twisted一起学(一):实现简单的TCP服务器

MINA、Netty、Twisted一起学(二):TCP消息边界问题及按行分割消息

MINA、Netty、Twisted一起学(三):TCP消息固定大小的前缀(Header)

MINA、Netty、Twisted一起学(四):定制自己的协议

MINA、Netty、Twisted一起学(五):整合protobuf

MINA、Netty、Twisted一起学(六):session

MINA、Netty、Twisted一起学(七):发布/订阅(Publish/Subscribe)

MINA、Netty、Twisted一起学(八):HTTP服务器

MINA、Netty、Twisted一起学(九):异步IO和回调函数

MINA、Netty、Twisted一起学(十):线程模型

MINA、Netty、Twisted一起学(十一):SSL/TLS

MINA、Netty、Twisted一起学(十二):HTTPS

源码

https://github.com/wucao/mina-netty-twisted

给TA打赏
共{{data.count}}人
人已打赏
安全网络

CDN安全市场到2022年价值76.3亿美元

2018-2-1 18:02:50

安全经验

分类与监督学习,朴素贝叶斯分类算法

2021-11-28 16:36:11

个人中心
购物车
优惠劵
今日签到
有新私信 私信列表
搜索