zoukankan      html  css  js  c++  java
  • Mina、Netty、Twisted一起学(十):线程模型

    要想开发一个高性能的TCPserver,熟悉所使用框架的线程模型非常重要。MINA、Netty、Twisted本身都是高性能的网络框架,假设再搭配上高效率的代码。才干实现一个高大上的server。

    可是假设不了解它们的线程模型。就非常难写出高性能的代码。框架本身效率再高。程序写的太差,那么server总体的性能也不会太高。就像一个电脑,CPU再好。内存小硬盘慢散热差,总体的性能也不会太高。

    玩过Android开发的同学会知道,在Android应用中有一个非常重要线程:UI线程(即主线程)。

    UI线程是负责一个Android的界面显示以及和用户交互。Activity的一些方法。比如onCreate、onStop、onDestroy都是运行在UI线程中的。

    可是在编写Activity代码的时候有一点须要非常注意。就是绝对不能把堵塞的或者耗时的任务写在这些方法中,假设写在这些方法中。则会堵塞UI线程,导致用户操作的界面反应迟钝。体验非常差。所以在Android开发中,耗时或者堵塞的任务会另外开线程去做。

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

    传统的BIO实现的TCPserver,特别对于TCP长连接,通常都要为每一个连接开启一个线程,线程也是操作系统的一种资源,所以非常难实现高性能高并发。

    而异步IO实现的TCPserver,因为IO操作都是异步的,能够用一个线程或者少量线程来处理大量连接的IO操作。所以仅仅须要少量的IO线程就能够实现高并发的server。


    在网络编程过程中,通常有一些业务逻辑是比較耗时、堵塞的,比如数据库操作,假设网络不好,加上数据库性能差。SQL不够优化,数据量大,一条SQL可能会运行非常久。

    因为IO线程本身数量就不多,通常仅仅有一个或几个,而假设这样的耗时堵塞的代码在IO线程中运行的话。IO线程的其它事情,比如网络read和write。就无法进行了,会影响IO性能以及整个server的性能。

    所以,不管是使用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:

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

    Connector thread:

    用于处理TCPclient连接到server,并将连接分配到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。


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

    MINA的TCPserver包括一个Acceptor thread和多个I/O processor thread,当有新的client连接到server,首先会由Acceptor thread获取到这个连接。同一时候将这个连接分配给多个I/O processor thread中的一个线程,当client发送数据给server,相应的I/O processor thread负责读取这个数据。并运行IoFilterChain中的IoFilter以及IoHandle。

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

    针对这个问题,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);
    	}
    }

    Netty:

    Netty的TCPserver启动时。会创建两个NioEventLoopGroup。一个boss,一个worker

    EventLoopGroup bossGroup = new NioEventLoopGroup();  
    EventLoopGroup workerGroup = new NioEventLoopGroup();

    NioEventLoopGroup实际上是一个线程组。能够通过构造方法设置线程数量。默觉得CPU核心数*2。

    boss用于server接收新的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);

    当有新的TCPclient连接到server,将由boss线程来接收连接。然后将连接注冊到worker线程。当client发送数据到server。worker线程负责接收数据,并运行ChannelPipeline中的ChannelHandler。

    和MINA的I/O processor thread 相似。Netty的worker线程本身数量不多。并且要实时处理IO事件,假设有耗时的业务逻辑堵塞住worker线程,比如在channelRead中运行一个耗时的数据库查询,会导致IO操作无法进行。server总体性能就会下降。


    在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); // server监听一个端口号,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);
    
    	}
    }

    Twisted:

    Twisted的线程模型是最简单粗暴的:单线程,即reactor线程。也就是,全部的IO操作、编码解码、业务逻辑等都是在一个线程中运行。

    实际上,即使是单线程。其性能也是非常高的。能够同一时候处理大量的连接。在单线程的环境下编程。不须要考虑线程安全的问题。

    只是。单线程带来一个问题,就是耗时的业务逻辑,假设运行在reactor线程中,那么其它事情,比如网络IO,就要等到reactor线程空暇时才干继续做,会影响到server的性能。

    以下的代码,通过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()


    作者:叉叉哥   转载请注明出处:http://blog.csdn.net/xiao__gui/article/details/40146351


    MINA、Netty、Twisted一起学系列

    MINA、Netty、Twisted一起学(一):实现简单的TCPserver

    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一起学(八):HTTPserver

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

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

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

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

    源代码

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




  • 相关阅读:
    python生成CSV文件并发送邮件
    Live2d Test Env
    Live2d Test Env
    Live2d Test Env
    Live2d Test Env
    Live2d Test Env
    Live2d Test Env
    Live2d Test Env
    Live2d Test Env
    扔鸡蛋
  • 原文地址:https://www.cnblogs.com/mfmdaoyou/p/6710441.html
Copyright © 2011-2022 走看看