Netty、MINA、Twisted中线程模型的示例分析
这篇文章主要介绍了Netty、MINA、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线程中运行，而是要另外开启线程来处理。
1. MINA
在MINA中，有三种非常重要的线程：Acceptor thread、Connector thread、I/O processor thread。
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，它不会保证同一连接的事件的执行顺序，并且可能会并发执行。二者之间可以根据需要来选择。
public class TcpServer {
public static void main(String[] args) throws IOException {
IoAcceptor acceptor = new NioSocketAcceptor(4); // 配置I/O processor thread线程数量
acceptor.getFilterChain().addLast("codec", new ProtocolCodecFilter(new TextLineCodecFactory()));
acceptor.getFilterChain().addLast("executor", new ExecutorFilter()); // 将TcpServerHandle中的业务逻辑拿到ExecutorFilter的线程池中执行
acceptor.setHandler(new TcpServerHandle());
acceptor.bind(new InetSocketAddress(8080));
}
}
class TcpServerHandle extends IoHandlerAdapter {
@Override
public void messageReceived(IoSession session, Object message)
throws Exception {
// 假设这里有个变态的SQL要执行3秒
Thread.sleep(3000);
}
}2. Netty
Netty的TCP服务器启动时，会创建两个NioEventLoopGroup，一个boss，一个worker：
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();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。
EventLoopGroup bossGroup = new NioEventLoopGroup(1);当有新的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来处理耗时的业务逻辑：
public class TcpServer {
public static void main(String[] args) throws InterruptedException {
EventLoopGroup bossGroup = new NioEventLoopGroup(1); // 服务器监听一个端口号，boss线程数建议设置成1
EventLoopGroup workerGroup = new NioEventLoopGroup(4); // worker线程数设置成4
try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
// 创建一个16个线程的线程组来处理耗时的业务逻辑
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));
// 将TcpServerHandler中的业务逻辑放到EventExecutorGroup线程组中执行
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 {
// 假设这里有个变态的SQL要执行3秒
Thread.sleep(3000);
}
}3.Twisted
Twisted的线程模型是最简单粗暴的：单线程，即reactor线程。也就是，所有的IO操作、编码解码、业务逻辑等都是在一个线程中执行。实际上，即使是单线程，其性能也是非常高的，可以同时处理大量的连接。在单线程的环境下编程，不需要考虑线程安全的问题。不过，单线程带来一个问题，就是耗时的业务逻辑，如果运行在reactor线程中，那么其他事情，例如网络IO，就要等到reactor线程空闲时才能继续做，会影响到服务器的性能。
下面的代码，通过reactor.callInThread将耗时的业务逻辑放到单独的线程池中执行，而不在reactor线程中运行。这样就不会影响到reactor线程的网络IO了。可以通过reactor.suggestThreadPoolSize设置这个线程池的线程数量。
# -*- coding:utf-8 –*-
import time
from twisted.internet.protocol import Protocol
from twisted.internet.protocol import Factory
from twisted.internet import reactor
# 耗时、阻塞的业务逻辑
def logic(data):
print data
time.sleep(3) # 假设这里有个变态的SQL要执行3秒
class TcpServerHandle(Protocol):
def dataReceived(self, data):
reactor.callInThread(logic, data) # 在线程池中运行logic(data)耗时任务，不在reactor线程中运行
reactor.suggestThreadPoolSize(8) # 设置线程池的线程数量为8
factory = Factory()
factory.protocol = TcpServerHandle
reactor.listenTCP(8080, factory)
reactor.run()由于Twisted的reactor的单线程设计，它的很多代码都不是线程安全的。所以在非reactor线程中执行的代码需要注意线程安全问题。例如transport.write就不是线程安全的。不过在非reactor线程中可以调用reactor.callFromThread方法，这个方法功能和callInThread相反，将一个函数从别的线程放到reactor线程中运行。不过还是要注意，reactor.callFromThread调用的函数由于运行在reactor线程中，如果运行耗时，同样会阻塞reactor线程，影响IO。
# -*- coding:utf-8 –*-
import time
from twisted.internet.protocol import Protocol
from twisted.internet.protocol import Factory
from twisted.internet import reactor
# 非线程安全的代码
def notThreadSafe():
print "notThreadSafe"
# 耗时、阻塞的业务逻辑
def logic(data):
print data
time.sleep(3) # 假设这里有个变态的SQL要执行3秒
reactor.callFromThread(notThreadSafe) # 在reactor线程中运行notThreadSafe()
class TcpServerHandle(Protocol):
def dataReceived(self, data):
reactor.callInThread(logic, data) # 在线程池中运行logic(data)耗时任务，不在reactor线程中运行
reactor.suggestThreadPoolSize(8) # 设置线程池的线程数量为8
factory = Factory()
factory.protocol = TcpServerHandle
reactor.listenTCP(8080, factory)
reactor.run()此外，twisted.internet.threads中提供了许多很方便的函数。例如threads.deferToThread用于将一个耗时任务放在线程池中执行，与reactor.callInThread不同的是，它的返回值是Deferred类型，可以通过添加回调函数，处理耗时任务完成后的结果（返回值）。
# -*- coding:utf-8 –*-
import time
from twisted.internet.protocol import Protocol
from twisted.internet.protocol import Factory
from twisted.internet import reactor, threads
# 耗时、阻塞的业务逻辑
def logic(data):
print data
time.sleep(3) # 假设这里有个变态的SQL要执行3秒
return "success"
# 回调函数
def logicSuccess(result):
# result即为logic函数的返回值，即"success"
print result
class TcpServerHandle(Protocol):
def dataReceived(self, data):
d = threads.deferToThread(logic, data) # 将耗时的业务逻辑logic(data)放到线程池中运行，deferToThread返回值类型是Deferred
d.addCallback(logicSuccess) # 添加回调函数
reactor.suggestThreadPoolSize(8) # 设置线程池的线程数量为8
factory = Factory()
factory.protocol = TcpServerHandle
reactor.listenTCP(8080, factory)
reactor.run()感谢你能够认真阅读完这篇文章，希望小编分享的“Netty、MINA、Twisted中线程模型的示例分析”这篇文章对大家有帮助，同时也希望大家多多支持编程网，关注编程网行业资讯频道，更多相关知识等着你来学习!
免责声明：
① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的，并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据，供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。
② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341
