Thrift简单原理及使用(JAVA)

    thrift是一个软件框架，用来进行可扩展且跨语言的服务的开发。它结合了功能强大的软件堆栈和代码生成引擎，以构建在 C++, Java, Python, PHP, Ruby, Erlang, Perl, Haskell, C#, Cocoa, JavaScript, Node.js, Smalltalk, and OCaml 这些编程语言间无缝结合的、高效的服务。    thrift最初由facebook开发，07年四月开放源码，08年5月进入apache孵化器。
    目前thrift在国内很多互联网公司已经开始用，thrift开发比较方便，并且支持语言多，你定义一个简单的定义文件中的数据类型和服务接口，以作为输入文件，编译器生成代码用来方便地生成RPC客户端和服务器通信的无缝跨编程语言。

下面是thrift传输过程图：


一. 定义service.thrift数据类型和服务接口

struct User{
  1:i64 id,
  2:string name,
  3:i64 timestamp,
  4:bool vip        
}

service UserService{
  User getById(1:i64 id)
}
复制代码

二. 生成API文件
        首先下载和安装thrift客户端,比如在windows平台下,下载thrift.exe,不过此处需要提醒,不同的thrift客户端版本生成的API可能不兼容.本例使用thrift-0.9.0.exe;通过"--gen"指定生成API所适配的语言.本实例为生成java客户端API.

> thrift.exe --gen java -o service service.thrift
复制代码

三. UserService实现类

public class UserServiceImpl implements UserService.Iface {
  @Override
  public User getById(long id){
    System.out.println("invoke...id:" + id);
    return new User();//for test
  }
}
复制代码

四.原理分析

1. User.java : thrift生成API的能力还是非常的有限,比如在struct中只能使用简单的数据类型(不支持Date,Collection<?>等),不过我们能从User中看出,它生成的类实现了"Serializable"接口和"TBase"接口.

    其中Serializable接口表明这个类的实例是需要序列化之后在网络中传输的,为了不干扰JAVA本身的序列化和反序列化机制,它还重写了readObject和writeObject方法.不过这对thrift本身并没有帮助.

    TBase接口是thrift序列化和反序列化时使用的,它的两个核心方法:read和write.在上述的thrift文件中,struct定义的每个属性都有一个序号,比如:1:id,那么thrift在序列化时,将会根据序号的顺序依次将属性的"名称 + 值"写入inputStream中,反序列化也是如此.(具体参见read和write的实现).

    因为thrift的序列化和反序列化实例数据时,是根据"属性序号"进行,这可以保证数据在inputstream和outputstream中顺序是严格的.这一点也要求API开发者,如果更改了thrift文件中的struct定义,需要重新生成客户端API,否则服务将无法继续使用(可能报错,也可能数据错误).thrift序列化/反序列化的过程和JAVA自带的序列化机制不同,它将不会携带额外的class结构,此外thrift这种序列化机制更加适合网络传输,而且性能更加高效.

    2. UserService.Client:   在生成的UserService中,有个Client静态类,这个类就是一个典型的代理类,此类已经实现了UserService的所有方法.开发者需要使用Client类中的API方法与Thrift server端交互,它将负责与Thrift server的Socket链接中,发送请求和接收响应.

    需要注意的时,每次Client方法调用,都会在一个Socket链接中进行,这就意味着,在使用Client消费服务之前,需要和Thrift server建立有效的TCP链接.(稍后代码示例)

1) 发送请求:

//参见:TServiceClient
//API方法调用时,发送请求数据流
protected void sendBase(String methodName, TBase args) throws TException {
  oprot_.writeMessageBegin(new TMessage(methodName, TMessageType.CALL, ++seqid_));//首先写入"方法名称"和"seqid_"
  args.write(oprot_);//序列化参数
  oprot_.writeMessageEnd();
  oprot_.getTransport().flush();
}

protected void receiveBase(TBase result, String methodName) throws TException {
  TMessage msg = iprot_.readMessageBegin();//如果执行有异常
  if (msg.type == TMessageType.EXCEPTION) {
    TApplicationException x = TApplicationException.read(iprot_);
    iprot_.readMessageEnd();
    throw x;
  }//检测seqid是否一致
  if (msg.seqid != seqid_) {
    throw new TApplicationException(TApplicationException.BAD_SEQUENCE_ID, methodName + " failed: out of sequence response");
  }
  result.read(iprot_);//反序列化
  iprot_.readMessageEnd();
}
复制代码

Thrift提供了简单的容错方式:每次方法调用,都会在Client端标记一个seqid,这是一个自增的本地ID,在TCP请求时将此seqid追加到流中,同时Server端响应时,也将此seqid原样返回过来;这样客户端就可以根据此值用来判断"请求--响应"是对应的,如果出现乱序,将会导致此请求以异常的方式结束.

  2) 响应

//参考: TBaseProcessor.java
@Override
public boolean process(TProtocol in, TProtocol out) throws TException {
  TMessage msg = in.readMessageBegin();
  ProcessFunction fn = processMap.get(msg.name);
  if (fn == null) {
    TProtocolUtil.skip(in, TType.STRUCT);
    in.readMessageEnd();
    TApplicationException x = new TApplicationException(TApplicationException.UNKNOWN_METHOD, "Invalid method name: '"+msg.name+"'");
    out.writeMessageBegin(new TMessage(msg.name, TMessageType.EXCEPTION, msg.seqid));
    x.write(out);//序列化响应结果,直接输出
    out.writeMessageEnd();
    out.getTransport().flush();
    return true;
  }
  fn.process(msg.seqid, in, out, iface);
  return true;
}
复制代码

3) Server端Socket管理和执行策略

TThreadPoolServer
public void serve() {
  try {
    //启动服务
    serverTransport_.listen();
  } catch (TTransportException ttx) {
    LOGGER.error("Error occurred during listening.", ttx);
    return;
  }

  // Run the preServe event
  if (eventHandler_ != null) {
    eventHandler_.preServe();
  }

  stopped_ = false;
  setServing(true);
  //循环,直到被关闭
  while (!stopped_) {
    int failureCount = 0;
    try {
    //accept客户端Socket链接,
    //对于每个新链接,将会封装成runnable,并提交给线程或者线程池中运行.
    TTransport client = serverTransport_.accept();
    WorkerProcess wp = new WorkerProcess(client);
    executorService_.execute(wp);
    } catch (TTransportException ttx) {
    if (!stopped_) {
      ++failureCount;
      LOGGER.warn("Transport error occurred during acceptance of message.", ttx);
    }
    }
  }
  //....
}
复制代码

Thrift Server端,设计思路也非常的直接...当前Service server启动之后,将会以阻塞的方式侦听Socket链接(代码参考TThreadPoolServer),每建立一个Socket链接,都会将此Socket经过封装之后,放入线程池中,本质上也是一个Socket链接对应一个Worker Thread.这个Thread只会处理此Socket中的所有数据请求,直到Socket关闭.

//参考:WorkerProcess
while (true) {

  if (eventHandler != null) {
    eventHandler.processContext(connectionContext, inputTransport, outputTransport);
  }

  if(stopped_ || !processor.process(inputProtocol, outputProtocol)) {
    break;
  }
}
复制代码

当有Socket链接不是很多的时候,TThreadPoolServer并不会有太大的性能问题,可以通过指定ThreadPool中线程的个数进行简单的调优..如果Socket链接很多,我们只能使用TThreadedSelectorServer来做支撑,TThreadedSelectorServer内部基于NIO + Socket模式,具有异步的特性;不过在绝大多数情况下,在thrift中使用"异步"似乎不太容易让人接受,毕竟这意味着Client端需要阻塞,并且这个阻塞时间是不可控的.单SelecorServer确实可以有效的提升Server的并发能力.
3. Client端代码示例

public class UserServiceClient {

    public void startClient() {
        TTransport transport;
        try {
            transport = new TSocket("localhost", 1234);
            TProtocol protocol = new TBinaryProtocol(transport);
            UserService.Client client = new UserService.Client(protocol);
            transport.open();
            User user = client.getById(1000);
      ////
            transport.close();
        } catch (TTransportException e) {
            e.printStackTrace();
        } catch (TException e) {
            e.printStackTrace();
        }
    }

}
复制代码

    4. Server端代码示例

public class Server {
    public void startServer() {
        try {
            TServerSocket serverTransport = new TServerSocket(1234);
            UserService.Processor process = new Processor(new UserServiceImpl());
            Factory portFactory = new TBinaryProtocol.Factory(true, true);
            Args args = new Args(serverTransport);
            args.processor(process);
            args.protocolFactory(portFactory);
            TServer server = new TThreadPoolServer(args);
            server.serve();
        } catch (TTransportException e) {
            e.printStackTrace();
        }
    }
}
复制代码

由于thrift对java socket封装一层为TSOCKET，大家在使用thrift客户端连接的时候，一定要用连接池方式，我后面也会将我以前写过的连接池分享给大家。thrift将传输数据封装成thrift对象传输所以在传输过程发送数据字节数要比socket传输大。一般thrift性能体现在使用多线程发送。