Netty、MINA、Twisted中线程模型的示例分析

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);
  }
}

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

EventLoopGroup bossGroup = new NioEventLoopGroup(1);

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);

  }
}

# -*- 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()

# -*- 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()

# -*- 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()