Golang&Python测试thrift

namespace go my.test.demo
namespace py my.test.demostruct Student{
1: i32 sid,
2: string sname,
3: bool ssex=0,
4: i16 sage,
}const map<string,string> MAPCONSTANT = {'hello':'world', 'goodnight':'moon'}service ClassMember {
list<Student> List(1:i64 callTime),
void Add(1: Student s),
bool IsNameExist(1:i64 callTime, 2:string name),
}

thrift -r --gen go my.test.thrift

thrift -r --gen py my.test.thrift

├── client
├── client.go
├── gen-py
│   ├── __init__.py
│   ├── client.py
│   └── my
│   ├── __init__.py
│   ├── __init__.pyc
│   └── test
│   ├── __init__.py
│   ├── __init__.pyc
│   └── demo
│   ├── ClassMember-remote
│   ├── ClassMember.py
│   ├── ClassMember.pyc
│   ├── __init__.py
│   ├── __init__.pyc
│   ├── constants.py
│   ├── ttypes.py
│   └── ttypes.pyc
├── my
│   └── test
│   └── demo
│   ├── class_member-remote
│   │   └── class_member-remote.go
│   ├── classmember.go
│   ├── constants.go
│   └── ttypes.go
├── my.test.thrift
├── server
└── server.go

类似Thrift的工具，还有Avro、protocol buffer,但相对于Thrift来讲，都没有Thrift支持全面和使用广泛。1) thrift内部框架一瞥
按照官方文档给出的整体框架，Thrift自下到上可以分为4层：
+-------------------------------------------+
| Server | -- 服务器进程调度
| (single-threaded, event-driven etc) |
+-------------------------------------------+
| Processor | -- RPC接口处理函数分发，IDL定义接口的实现将挂接到这里面
| (compiler generated) |
+-------------------------------------------+
| Protocol | -- 协议
| (JSON, compact etc) |
+-------------------------------------------+
| Transport | -- 网络传输
| (raw TCP, HTTP etc) |
+-------------------------------------------+Thrift实际上是实现了C/S模式，通过代码生成工具将接口定义文件生成服务器端和客户端代码（可以为不同语言），从而实现服务端和客户端跨语言的支持。用户在Thirft描述文件中声明自己的服务，这些服务经过编译后会生成相应语言的代码文件，然后用户实现服务（客户端调用服务，服务器端提服务）便可以了。其中protocol（协议层, 定义数据传输格式，可以为二进制或者XML等）和transport（传输层，定义数据传输方式，可以为TCP/IP传输，内存共享或者文件共享等）被用作运行时库。2)支持的数据传输格式、数据传输方式和服务模型
（a）支持的传输格式
TBinaryProtocol – 二进制格式.
TCompactProtocol – 压缩格式
TJSONProtocol – JSON格式
TSimpleJSONProtocol –提供JSON只写协议, 生成的文件很容易通过脚本语言解析。
TDebugProtocol – 使用易懂的可读的文本格式，以便于debug
（b） 支持的数据传输方式
TSocket -阻塞式socker
TFramedTransport – 以frame为单位进行传输，非阻塞式服务中使用。
TFileTransport – 以文件形式进行传输。
TMemoryTransport – 将内存用于I/O. java实现时内部实际使用了简单的ByteArrayOutputStream。
TZlibTransport – 使用zlib进行压缩， 与其他传输方式联合使用。当前无java实现。
（c）支持的服务模型
TSimpleServer – 简单的单线程服务模型，常用于测试
TThreadPoolServer – 多线程服务模型，使用标准的阻塞式IO。
TNonblockingServer – 多线程服务模型，使用非阻塞式IO（需使用TFramedTransport数据传输方式）3) Thrift IDL
Thrift定义一套IDL（Interface Definition Language）用于描述接口，通常后缀名为.thrift，通过thrift程序把.thrift文件导出成各种不一样的代码的协议定义。IDL支持的类型可以参考这里：http://thrift.apache.org/docs/types

#client.py#!/usr/bin/env python
# -*- coding: utf-8 -*-
import sys, glob, time,datetime
sys.path.append('gen-py')
#print glob.glob('libpy/lib.*')
#sys.path.insert(0, glob.glob('libpy/lib.*')[0])from my.test.demo import ClassMember
from my.test.demo.ttypes import *from thrift import Thrift
from thrift.transport import TSocket
from thrift.transport import TTransport
from thrift.protocol import TBinaryProtocoltry:
startTime = time.time()*1000
# Make socket
transport = TSocket.TSocket('127.0.0.1', 10086) # Framed is critical. Raw sockets are very slow
transport = TTransport.TFramedTransport(transport) # Wrap in a protocol
protocol = TBinaryProtocol.TBinaryProtocol(transport) # Create a client to use the protocol encoder
client = ClassMember.Client(protocol) # Connect!
transport.open() for i in range(1,6):
i+=1
s = Student()
s.sid = i
s.sname = "name_{}".format(i)
r = client.Add(s)
print(s) endTime = time.time()*1000 print client.List(time.time()*1000) print "use:%d-%d=%dms" %(endTime,startTime, (endTime - startTime))
# Close!
transport.close()except Thrift.TException, tx:
print 'ERROR:%s' % (tx.message)

#python的thrift package安装
sudo easy_install thrift==0.9.0#golang的package安装
go get git.apache.org/thrift.git/lib/go/thrift

qpzhang@qpzhangdeMac-mini:~/project/thrift-0.9.2/mytest $./client
add 0 student <nil>
add 1 student <nil>
add 2 student <nil>
add 3 student <nil>
add 4 student <nil>
<nil>
Student({Sid:0 Sname:name_0 Ssex:false Sage:0})
Student({Sid:1 Sname:name_1 Ssex:false Sage:0})
Student({Sid:2 Sname:name_2 Ssex:false Sage:0})
Student({Sid:3 Sname:name_3 Ssex:false Sage:0})
Student({Sid:4 Sname:name_4 Ssex:false Sage:0})
calltime:1429856944847-1429856929841=15006ms

qpzhang@qpzhangdeMac-mini:~/project/thrift-0.9.2/mytest/gen-py $python client.py
Student(sname='name_2', sage=None, ssex=False, sid=2)
Student(sname='name_3', sage=None, ssex=False, sid=3)
Student(sname='name_4', sage=None, ssex=False, sid=4)
Student(sname='name_5', sage=None, ssex=False, sid=5)
Student(sname='name_6', sage=None, ssex=False, sid=6)
[Student(sname='name_2', sage=0, ssex=False, sid=2), Student(sname='name_3', sage=0, ssex=False, sid=3), Student(sname='name_4', sage=0, ssex=False, sid=4), Student(sname='name_5', sage=0, ssex=False, sid=5), Student(sname='name_6', sage=0, ssex=False, sid=6)]
use:1429858724602-1429858724600=1ms

qpzhang@qpzhangdeMac-mini:~/project/thrift-0.9.2/mytest $./client
add 0 student <nil>
add 1 student <nil>
add 2 student <nil>
add 3 student EOF
add 4 student write tcp 127.0.0.1:10086: broken pipe
write tcp 127.0.0.1:10086: broken pipe
calltime:1429856896694-1429856881690=15004ms

TThreadPoolServer – 多线程服务模型，使用标准的阻塞式IO。
TNonblockingServer – 多线程服务模型，使用非阻塞式IO（需使用TFramedTransport数据传输方式）

func (p *TSimpleServer) AcceptLoop() error {
for {
client, err := p.serverTransport.Accept()
if err != nil {
select {
case <-p.quit:
return nil
default:
}
return err
}
if client != nil {
go func() {
if err := p.processRequests(client); err != nil {
log.Println("error processing request:", err)
}
}()
}
}
}func (p *TSimpleServer) Serve() error {
err := p.Listen()
if err != nil {
return err
}
p.AcceptLoop()
return nil
}

func (p *TSimpleServer) processRequests(client TTransport) error {
processor := p.processorFactory.GetProcessor(client)
inputTransport := p.inputTransportFactory.GetTransport(client)
outputTransport := p.outputTransportFactory.GetTransport(client)
inputProtocol := p.inputProtocolFactory.GetProtocol(inputTransport)
outputProtocol := p.outputProtocolFactory.GetProtocol(outputTransport)
defer func() {
if e := recover(); e != nil {
log.Printf("panic in processor: %s: %s", e, debug.Stack())
}
}()
if inputTransport != nil {
defer inputTransport.Close()
}
if outputTransport != nil {
defer outputTransport.Close()
}
for {
ok, err := processor.Process(inputProtocol, outputProtocol)
if err, ok := err.(TTransportException); ok && err.TypeId() == END_OF_FILE {
return nil
} else if err != nil {
log.Printf("error processing request: %s", err)
return err
}
if !ok {
break
}
}
return nil
}

func NewClassMemberProcessor(handler ClassMember) *ClassMemberProcessor { self6 := &ClassMemberProcessor{handler: handler, processorMap: make(map[string]thrift.TProcessorFunction)}
self6.processorMap["List"] = &classMemberProcessorList{handler: handler}
self6.processorMap["Add"] = &classMemberProcessorAdd{handler: handler}
self6.processorMap["IsNameExist"] = &classMemberProcessorIsNameExist{handler: handler}
return self6
}func (p *ClassMemberProcessor) Process(iprot, oprot thrift.TProtocol) (success bool, err thrift.TException) {
name, _, seqId, err := iprot.ReadMessageBegin()
if err != nil {
return false, err
}
if processor, ok := p.GetProcessorFunction(name); ok {
return processor.Process(seqId, iprot, oprot)
}
iprot.Skip(thrift.STRUCT)
iprot.ReadMessageEnd()
x7 := thrift.NewTApplicationException(thrift.UNKNOWN_METHOD, "Unknown function "+name)
oprot.WriteMessageBegin(name, thrift.EXCEPTION, seqId)
x7.Write(oprot)
oprot.WriteMessageEnd()
oprot.Flush()
return false, x7}type classMemberProcessorList struct {
handler ClassMember
}func (p *classMemberProcessorList) Process(seqId int32, iprot, oprot thrift.TProtocol) (success bool, err thrift.TException) {
args := ListArgs{}
if err = args.Read(iprot); err != nil {
iprot.ReadMessageEnd()
x := thrift.NewTApplicationException(thrift.PROTOCOL_ERROR, err.Error())
oprot.WriteMessageBegin("List", thrift.EXCEPTION, seqId)
x.Write(oprot)
oprot.WriteMessageEnd()
oprot.Flush()
return false, err
} iprot.ReadMessageEnd()
result := ListResult{}
var retval []*Student
var err2 error
if retval, err2 = p.handler.List(args.CallTime); err2 != nil {
x := thrift.NewTApplicationException(thrift.INTERNAL_ERROR, "Internal error processing List: "+err2.Error())
oprot.WriteMessageBegin("List", thrift.EXCEPTION, seqId)
x.Write(oprot)
oprot.WriteMessageEnd()
oprot.Flush()
return true, err2
} else {
result.Success = retval
}
if err2 = oprot.WriteMessageBegin("List", thrift.REPLY, seqId); err2 != nil {
err = err2
}
if err2 = result.Write(oprot); err == nil && err2 != nil {
err = err2
}
if err2 = oprot.WriteMessageEnd(); err == nil && err2 != nil {
err = err2
}
if err2 = oprot.Flush(); err == nil && err2 != nil {
err = err2
}
if err != nil {
return
}
return true, err
}