RPC远程协议之Thrift非阻塞多线程处理

Posted 2022-12-03 云水之路

RPC远程协议之Thrift非阻塞多线程处理

 

在上一篇文章《RPC远程协议之Thrift入门》中，我已经介绍了Thrift的基本使用，并举例验证了如何使用，当然这个例子是阻塞单线程的模式，实际客户端请求增多，单线程处理时间变长时，无可厚非就会卡顿延迟，所以需要实现非阻塞多线程的通信模式，而Thrift已经想到，并且为我们提供好了几种Server的实现方式，以供在实际业务需求下做出合适的选择。本篇文章主要介绍Java版本的几种Server实现的异同，并对上篇文章的例子做出改进，使其支持非阻塞并且是多线程的。

 

 

l  模块化结构

l  几种Server版本

l  例子验证

 

 

一、模块化结构

引用Thrift官网的结构栈如下：

 

如上图所示，Thrift含有四个主要的组件：Server、Processor、Protocol，以及Transport。其中，protocol定义了消息是按照什么格式序列化的，transport则定义了消息是怎样在客户端和服务端传递的，processor则是由编译工具自动生成的业务处理器，而server则负责从transport接收序列化消息，并根据protocol反序列化它，调用用户自定义的消息处理器，并序列化消息处理器的响应，最后再将其写会transport返回给客户端。

因为Thrift的模块化结构，使得它可以提供多种server实现方式，来满足不同业务需求，比如：

1.  TSimpleServer

2.  TNonblockingServer

3.  THsHaServer

4.  TThreadedSelectorServer

5.  TThreadPoolServer

 

这几种不同的Server版本的异同会在下面介绍，请往下俯瞰。

 

 

二、几种Server版本

1、TSimpleServer

接收一个客户端的连接，并处理连接请求，直到客户端关闭连接，它才会接收新的请求连接。所以，它只在一个单独的线程中以阻塞方式完成工作，所以它只能同时服务于一个客户端连接，其它所有客户端在被服务端接受前都在排队等待。我们上一篇文章的服务端实现，就是采用这种方式来演示的。

 

PS：其主要用于测试使用，一般不在生产环境使用。

 

2、TNonblockingServer

顾名思义，TNonblockingServer为非阻塞的实现方式，解决了TSimpleServer同时只能提供一个连接而阻塞其它所有客户端连接的问题。实际上，其使用了java.nio.channels.Selector，通过调用select()，可以使连接阻塞在多个连接点上，而不是阻塞在单一连接上。当一个或多个连接准备好接收/读写时，select()便会返回。

TNonblockingServer在处理这些连接时，要么接受它，要么从他们那读取数据，要么写入数据到它们，然后再调用select()来等待下一个可用的连接。所以，此种方式的服务端可以同时服务多个客户端，而不会出现一个客户端把其它客户端阻塞的情况。

 

3、THsHaServer

TNonblockingServer解决了同时可以为多个客户端提供服务，但是也存在问题是：所有消息被调用了select()的同一个线程处理，如果有10个客户端，处理每条消息需要100毫秒，那么延迟和吞吐量会怎样？当一个消息被处理，其它9个客户端就等着被select，所以客户端需等待1秒才能从服务端得到响应，吞吐里就是10次/秒，那要是同时可以处理多条消息，是不是100次/秒，这可好？

 

THsHaServer解决了上面的问题：它使用单一线程处理网络IO，一个独立的worker线程池来处理消息，只要有空闲的worker线程，消息就会立即被处理，因此多条消息可被并行处理。

 

4、TThreadedSelectorServer

与THsHaServer主要不同的地方是：TThreadedSelectorServer允许用多个线程来处理网络IO，它维护了两个线程池子，一个用来处理网络IO，一个用来处理请求消息。所以，当网络IO出现瓶颈时，其要比THsHaServer表现更好。而本篇的例子也是采用此种方式来实现服务端，具体查看第三部分。

 

5、TThreadPoolServer

与其它版本Server不同的是：提供一个专门线程来接受连接，一旦接受到连接，就将其放入到ThreadPoolExecutor中的一个worker线程里处理。而这个worker线程被绑定到特定的客户端连接上，直到客户端关闭，这个worker线程才会被放回线程池中，供其他客户端使用。另外，我们可以灵活配置线程池的大小，当然，如果客户端数量超过了线程池中的最大线程数，在有一个worker线程可用之前，请求将被一直阻塞在那里。

 

如果你提前知道了将要连接到你服务器上的客户端数量，并且你不介意运行大量线程的话，TThreadPoolServer对你可能是个很好的选择。

 

我认为TThreadedSelectorServer对大多数案例来说都是个安全之选。如果你的系统资源允许运行大量并发线程的话，你可能会考虑使用TThreadPoolServer。

 

 

三、例子验证

本篇文章继续以上一篇文章中的例子介绍下，在Thrift框架中，如何实现非阻塞多线程的C/S通信方式。当然，这里我们只对UserServer和UserClient进行升级重构，具体代码的意思，请查看代码中的注释说明，不再赘述，具体如下：

 

1、服务端(UserServer.java)

/**
 * 将业务处理逻辑UserHandler作为具体的业务
 * 处理器，传递给Thrift服务器，执行处理逻辑.
 * 服务端实现方式采用：
 * TThreadedSelectorServer
 */
public class UserServer
    private static final int port = 9081;

    private void invoke()
        try
            // 非阻塞式的Transport，配合TFrameTransport使用
            TNonblockingServerTransport serverTransport = new TNonblockingServerSocket(port);
            // 关联处理器Processor与Service服务的实现
            TProcessor processor = new UserService.Processor<UserService.Iface>(new UserHandler());

            // 采用TThreadedSelectorServer.Args配置可并发处理客户端请求
            TThreadedSelectorServer.Args args = new TThreadedSelectorServer.Args(serverTransport);
            // 指定业务处理器processor
            args.processor(processor);
            // 设定协议工厂，指定高效的、密集的二进制编码格式传输
            args.protocolFactory(new TCompactProtocol.Factory());
            // 设定传输工厂，指定非阻塞、按块的大小传输，类似NIO
            args.transportFactory(new TFramedTransport.Factory());
            // 设定处理器工厂，指定只返回一个单例实例
            args.processorFactory(new TProcessorFactory(processor));

            // 指定多个线程，负责客户端的网络IO处理
            args.selectorThreads(2);
            // 指定worker线程池的大小
            ExecutorService pool = Executors.newFixedThreadPool(3);
            args.executorService(pool);

            // 采用TThreadedSelectorServer方式实现服务端
            TThreadedSelectorServer server = new TThreadedSelectorServer(args);
            System.out.println("Starting server on port:" + port + " ...");
            server.serve();
        catch(TTransportException e)
            e.printStackTrace();
       
   

    public static void main(String[]args)
        UserServer us = new UserServer();
        us.invoke();
   

 

 

 

 

2、客户端(UserClient.java)

/**
 * 远程调用服务接口，获取用户信息及订单列表
 * 这里采用非阻塞的传输方式
 */
public class UserClient
    private static final int port = 9081;
    private static final String addr = "localhost";
    private static final int timeout = 100 * 1000;

    public void start()
        // 非阻塞方式，按块的大小进行传输
        TTransport transport = new TFramedTransport(new TSocket(addr,port,timeout));
        // 采用高效率的、密集的二进制格式进行数据传输协议
        TProtocol protocol = new TCompactProtocol(transport);

        UserService.Client client = new UserService.Client(protocol);
        try
            open(transport);
            User user = client.getUserOrders(10000021);
            System.out.println(user);   // 打印返回结果
            close(transport);
        catch (TException e)
            e.printStackTrace();
       
   

    private void open(TTransport transport)
        if(null != transport && !transport.isOpen())
            try
                transport.open();
            catch(TTransportException e)
                e.printStackTrace();
           
       
   

    private void close(TTransport transport)
        if(null != transport && transport.isOpen())
            transport.close();
       
   

    public static void main(String[] args)
        UserClient client = new UserClient();
        client.start();
   

 

 

 

 

 

好了，就介绍到这里，至于运行的结果，与上篇的相同，这里就不列出来了。