Thrift从入门到TNonblockingServer学习

目录

入门

从0开始

设计做完了，小吴要开始编码了。如果按照“手工作坊”的思路，小吴至少需要完成如下几个方面：
（1）“客户端向服务器端发送数据”的代码
（2）“客户端接收服务器端查询结果”的代码
（3）“服务器端接收客户端数据”的代码
（4）“服务器端向客户端发送查询结果”的代码
（5）如果客户端会大批量发起查询，那可能还需要考虑改成多线程模型或异步模型

从1开始

第1步： 明确要交互的数据格式（如上例中的UserGradeInfo）和具体的方法（如上例中的Search），定义出thrift接口描述文件（英文叫做Inteface Description File）；
(即传输数据的 数据结构，与获取待传输数据的 业务逻辑)

第2步： 调用thrift工具，依据thrift接口文件，生成RPC代码；

第3步： 你的服务器端程序引用thrift生成的RPC代码，并实现其中的Search动作的逻辑，然后启动监听，等待客户端发来请求。

第4步： 客户端同样引入并调用RPC代码来与服务器端通信；

安装与初步使用

安装过程与生成代码

  • 安装过程

    • 首先保证安装好boost与libevent
    • yum install -y libevent libevent-devel libtool autoconf automake byacc flex bison pkgconfig crypto-utils
    • 下载对应版本的thrift (不同版本thrift并不兼容)，解压并进入
    • ./bootstrap.sh
    • ./configure -h
    • ./configure --prefix=/usr/local/thrift_0_12_0 --with-cpp --with-java --without-csharp --without-go --without-python --without-erlang --without-perl --without-php --without-php_extension --without-ruby --without-haskell --without-lua --without-nodejs --without-nodets --without-dart --without-rs --without-cl --without-haxe --without-dotnetcore --without-d --enable-tests=no
    • ./configure CPPFLAGS='-DNDEBUG'
    • make -j 20
    • sudo make install

  • 编译工具

    • thrift --gen ${开发语言} ${thrift接口描述文件}
    • 如果.thrift文件中有 include "other.thrift"，需要使用thrift -r ...

  • 生成的代码

    • processor
      • 从输入流读，向输出流写
    • handler
      • 需要实现的业务逻辑

  • 传输协议规范 TProtocol | 文本、二进制

    • TBinaryProtocol
    • TCompactProtocol
    • TJSONProtocol
    • TSimpleJSONProtocl
    • TDebugProtocol

  • 传输数据标准 TTransports | 传输层

    • TSocket
    • TFramedTransport
    • TFileTransport
    • TMemoryTransport
    • TZlibTransport

  • 服务端类型

    • TSimpleServer
    • TThreadPoolServer
    • TNonblockingServer

  • 内部序列化机制

    • 对传输数据进行 简化 和 压缩
    • 提高 高并发 和 大型系统中 数据交互的成本

代码实例

服务端相关代码
客户端相关代码

issue address

进阶 - NonblockingServer 学习

以下内容基于0.12.0版本，代码地址：https://github.com/apache/thrift/tree/0.12.0

TNonblockingIOThread

TNonblockingIOThread 是 IO 线程的封装。其提供两种事件回调函数：listenHandler方法是(0号 IO 线程中)发生连接事件的回调，它将创建新的或者复用已有的 TConnection 实例，并通过调用 TConnection 实例中的 transition 方法注册当前 socket 上的读事件；notifyHandler 方法是任务完成事件的回调，它调用 transition 方法切换状态(进而继续监听当前 socket 上的读事件)。TConnection 中 registerEvents() 方法将注册对当前 socket(如果是 listenSocket) 的可读(新连接)事件 和 对当前 TNonblocking 上 pipe 的可读(任务完成)事件，当它对应的 IO 线程启动后就会调用这个方法。

TConnection

TConnection 封装了 socket。其内部维护着 app state 与 socket state 两类状态，前者代表 server 处理任务的阶段，后者代表处理 socket 数据的阶段，它们都在 transition 方法中进行状态切换。TConnection 可以简单理解为 "处理 socket 读写" 与 "执行任务" 之间的桥梁。具体来讲，transition 中的 setRead/setWrite 方法会注册当前 socket 上的可读/可写事件，其回调函数为当前 TConnection 实例的 workSocket 方法的静态包装器 eventHandler。 而 workSocket方法在完成相应的 socket 读写操作后，调用 transition 方法进行状态切换。

ThreadManager

TNonblockingServe r中，任务是由 worker 线程池中的线程来执行的。ThreadManager 的功能包含创建、销毁 worker 线程以及统计不同状态的 worker 线程数量等。这个类还涉及到任务的调度，简单来说主要就是有生产者操作(添加任务)，消费者操作(取出并执行任务)，保存任务的队列，以及协调多个线程各种操作之间的同步关系；ThreadManager 是一个抽象类，ThreadManager::Impl 是其实现，而用户直接创建的是 ThreadManager::Impl 的子类 SimpleThreadManager 的实例；对于用户而言，workerCount_ 是线程池中有效的worker线程数量，workerMaxCount_ 则是用户设置的线程池大小，pendingTaskCountMax_ 是用户设置的任务队列(保存的是正在排队的任务)大小

  • 条件变量及含义

    • monitor_
      • 队列非空，即通知可以任务队列中取出任务并执行
    • maxMonitor_
      • 队列非满，即通知可以向任务队列中添加任务
    • workerMonitor_
      • 添加worker线程，或者清除失效worker线程，使用它进行同步

  • 执行任务

    • 每个启动的 worker 线程都会在其 run 方法中，循环的(或等待)从同一个任务队列中取任务并执行

  • 添加任务

    • 每个 IO 线程监听到可读事件发生并读取出 request frame，之后调用 transition 方法将(创建的)任务添加到上面提到的任务队列，当队列满时发生阻塞

(伪)时序图

TNonblockingServer 架构/IO模型

TNonblockingServer 采用的是 多个IO线程 + worker线程池 模型，两种线程的数量都可以设置；主 IO 线程负责建立连接并创建 TConnection 对象，然后会把TConnection 分配给某 IO 线程(round robin)监听，IO 线程使用 libevent 进行 socket 的异步读写(即数据的收发)

  • 连接管理

    • activeConnections_
      • 活跃连接队列
    • connectionStack_
      • 空闲连接队列，或者是已"closed"连接队列，当作 TConnection 对象池用

  • 管道

    • 用来通知任务开始、任务完成
    • 其实现为 socketpair

  • 事件管理

    • 新建连接事件
      • listenHandler -> handleEvent
    • 已建立的连接上发生的事件
      • eventHandler -> workSocket
    • 管道事件
      • notifyHandler

  • 状态切换

    • 状态类型
      • socketStat_
      • appStat_
    • APP状态切换过程中，对已建立的连接上监听的事件会做修改
      • 当进入APP_READ_REQUEST状态(读取请求体)时，设置不监听任何事件；进入APP_WAIT_TASK状态(处理完成)后，才进入APP_INIT状态，设置监听可读事件；即单个连接上的任务是串行执行的
      • 处理任务经过的几个状态的切换，都是来自事件管理中的回调函数调用 transition 方法

TNonblockingServer 实践及注意

过载

通过 setOverloadAction 方法设置，这个过载指的是活跃连接个数超过了限制(maxConnections_)或者 正在处理+等待处理 的连接个数超过了限制(maxActiveProcessors_)，具体用法以及使用场景待补充

任务排队(超出设置的队列最大长度)及死锁情况

上面提到过，当任务队列满时，IO 线程会阻塞在向队列添加任务的操作上；thrift 代码中直接将阻塞timeout设置为0，表示一直阻塞下去；但代码中没有提供接口设置阻塞时间。这会导致某种条件下的死锁，分析如下：

worker 线程在执行完任务后，会调用 TNonblockingIOThread::notify 通知 IO 线程当前任务已完成，然后会阻塞等待管道(socket pair)的可写事件，如果管道已满，并且如果 IO 线程来不及立即处理这些任务完成事件，那么 (所有) worker 线程会阻塞在其 run 方法中(但在这之前已经释放manager_->mutex_)，无法从队列中获取新任务执行；同时有许多新请求过来，需要 IO 线程向任务队列中添加任务，由于没有worker线程取任务执行了，队列保持最大长度，(所有) IO 线程阻塞在添加任务的操作上(ThreadManager::maxMonitor_，即等待可用的worker线程，在这之前已经获取manager_->mutex_)，这样就会造成死锁

本文最后给出了避免这种死锁的修改，可供参考

任务过期

通过 setTaskExpireTime 可以设置一个超时时间(毫秒)，从队列中取出任务时判断在队列中停留超过这个时间就会被丢弃而不是被执行，这个时间为0表示不会过期，但为负数时，所有任务都会按过期处理

过期任务的处理及死锁情况

针对过期任务，在 ThreadManager::Worker::run 中 worker 线程是不会释放 threadManager->mutex_ 的，而且直接调用 manager_->expireCallback_，进而又调用 TNonblockingIOThread::notify，然后会阻塞等待管道(socket pair)的可写事件。此时可能会有的死锁情况：

如果管道已满，IO 线程又无法立即处理这些任务完成事件，并且又有许多新请求过来，需要 IO 线程向任务队列中添加任务(需要对 threadManager->mutex_加锁，也使用上面提到的阻塞timeout=0)，则会造成死锁

当获取到过期任务或者添加任务超时后的处理

发现任务过期后会调用 manager_->expireCallback_，进而调用 forceClose，server 主动关闭了连接；在 addTask 中获取锁或者等待条件变量超时后，抛出异常，之后也会通过 transition 函数主动关闭连接

修改代码

实际使用中还是需要对 thrift 代码稍作修改的，主要有以下几方面：

  • 添加一些获取 server 状态数据的方法
  • 发生添加超时后丢弃这个任务(给client返回一个empty frame)，而不是直接关闭连接
  • 解决对于上述分析中的添加任务阻塞可能导致死锁和任务过期可能导致死锁的问题
    • 设置添加任务时发生阻塞的超时时间
    • 处理过期任务(调用manager_->expireCallback_)前先释放threadManager->mutex_，执行完后再获取
    • (优化)当发生添加任务超时、任务过期返回一个空结果(而不是直接关闭连接)

相关代码

大量连接断开时发生瞬时阻塞

接收到客户端断开连接请求时，TNonblockingServer::TConnection::close 函数会关闭与客户端的连接，activeConnections_ 维护着当前活跃的 TConnection，其类型是std::vector，因此当删除其中被关闭的 TConnection 时，首先要遍历整个vector，然后由于 TConnection 没有实现移动赋值运算符，形如 V.erase( std::remove(V.begin, V.end, target), V.end) 的调用，其删除实际是由拷贝操作实现，最后erase方法执行实际上的删除并将vector size设置成正确的值

因此大量的比较和拷贝操作会比较耗时，从而导致 workSocket 函数中的 close 函数比较耗时，进一步导致 workSocket 处理后续请求变慢，直到处理完所有断开的 TConnection后才恢复，即表现为瞬时阻塞住了。那么解决这个问题的比较简单做法是使用std::set替换std::vector即可，或者将更新 activeConnections_ 操作放到独立的线程中并对相关操作加以限流

参考资料

Thrift 入门教程
Apache Thrift
Apache Thrift paper
thrift解读（三）——TNonblockingServer 内的 IO线程 和 work线程之间数据流处理逻辑
Thrift解读（四）——TNonblockingServer 作业调度
Thrift 的TNonblockingServer运行原理分析
Thrift线程和状态机分析
Thrift异步IO服务器源码分析
How to get client IP Address with C++ in thrift
Erasing elements from a vector

【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/wangzhiyi/p/9490544.html