RPC框架之Thrift简单使用

前言

Thrift是一个可以跨平台，跨语言的RPC(远程过程调用)框架，通过IDL(接口描述语言)来定义数据类型和接口，相当于Protobuf和gRPC的结合体。Thrift最开始由Facebook开源，后来贡献给了Apache。

下载编译器

官网，这里我们下载windows版本的编译器thrift-0.16.0.exe。

定义thrift文件

namespace java com.imooc.sourcecode.java.google.thrift.test1
typedef i16 short
typedef i32 int
typedef i64 long
typedef bool boolean
typedef string String
struct Person {
1: optional String username
2: optional int age
3: optional boolean married
}
exception DataException {
1: optional String message
2: optional String callStack
3: optional String date
}
service PersonService {
Person getPersonByUsername(1: required String username) throws (1: DataException dataException),
void savePerson(1: required Person person) throws (1: DataException dataException),
}


namespace 定义包名，类似ProtoBuf的java_package。

typedef 定义类型的别名。

optional 表示当前字段可选，非必填。

1: optional String message 每个字段需要有一个唯一的号码，必须大于0。

exception 定义异常，可以看做Java中的异常。

service 定义服务，可以看做Java中的接口。

struct 可以看做一个Java类，类似ProtoBuf的message。

Java处理

添加maven依赖


org.apache.thrift
libthrift
0.16.0



根据thrift文件创建Java数据类型和接口代码

thrift -out $DST_DIR --gen java $SRC_DIR/data.proto


-out表示生成代码的路径，java表示生成Java语言的代码，实际命令为

.\thrift-0.16.0.exe -out D:\java\code_resp\github_resp\source_code\src\main\java --gen java D:\java\code_resp\github_resp\source_code\src\main\java\com\imooc\sourcecode\java\google\thrift\test1\data.thrift


-out值不要包含具体的包路径，thrift文件中已经配置了。

序列化和反序列化

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import org.apache.thrift.TDeserializer;
import org.apache.thrift.TSerializer;
import org.apache.thrift.protocol.TCompactProtocol.Factory;
public class TestSerialize {
public static void main(String[] args) throws Exception {
testSerialize();
testDeserialize();
}
private static void testSerialize() throws Exception {
Person person = new Person();
person.setUsername("李四");
person.setAge(23);
person.setMarried(true);
TSerializer serializer = new TSerializer(new Factory());
byte[] bytes = serializer.serialize(person);
new ByteArrayInputStream(bytes)
.transferTo(new FileOutputStream("D:/testjar/thrift_java_serialize_person"));
}
private static void testDeserialize() throws Exception {
TDeserializer deserializer = new TDeserializer(new Factory());
Person person = new Person();
ByteArrayOutputStream output = new ByteArrayOutputStream();
new FileInputStream("D:/testjar/thrift_java_serialize_person").transferTo(output);
deserializer.deserialize(person, output.toByteArray());
System.out.println(person);
}
}


序列化和反序列化要使用相同的TProtocolFactory类型。

服务器端代码

import com.imooc.sourcecode.java.google.thrift.test1.PersonService.Processor;
import org.apache.thrift.TException;
import org.apache.thrift.TProcessorFactory;
import org.apache.thrift.protocol.TCompactProtocol;
import org.apache.thrift.server.THsHaServer;
import org.apache.thrift.server.THsHaServer.Args;
import org.apache.thrift.transport.TNonblockingServerSocket;
import org.apache.thrift.transport.TTransportException;
import org.apache.thrift.transport.layered.TFramedTransport;
public class MyServer {
public static void main(String[] args) throws TTransportException {
TNonblockingServerSocket serverSocket = new TNonblockingServerSocket(8883);
THsHaServer.Args serverArgs = new Args(serverSocket).minWorkerThreads(2).maxWorkerThreads(4);
Processor

processor = new Processor<>(new PersonServiceImpl());
serverArgs.protocolFactory(new TCompactProtocol.Factory());
serverArgs.transportFactory(new TFramedTransport.Factory());
serverArgs.processorFactory(new TProcessorFactory(processor));
System.out.println("Server start");
THsHaServer server = new THsHaServer(serverArgs);
server.serve();
}
public static class PersonServiceImpl implements PersonService.Iface {
@Override
public Person getPersonByUsername(String username) throws DataException, TException {
System.out.println("param: " + username);
Person person = new Person();
person.setUsername(username);
person.setAge(23);
person.setMarried(true);
return person;
}
@Override
public void savePerson(Person person) throws DataException, TException {
System.out.println("param: " + person);
System.out.println(person);
}
}
}


定义PersonServiceImpl实现类，来处理客户端请求。

客户端代码

import com.imooc.sourcecode.java.google.thrift.test1.PersonService.Client;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TCompactProtocol;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.layered.TFramedTransport;
public class MyClient {
public static void main(String[] args) throws TException {
TFramedTransport transport = new TFramedTransport(new TSocket("localhost", 8883));
TCompactProtocol protocol = new TCompactProtocol(transport);
Client client = new Client(protocol);
transport.open();
Person queryPerson = client.getPersonByUsername("张三");
System.out.println(queryPerson);
Person savePerson = new Person();
savePerson.setUsername("李四");
savePerson.setAge(21);
savePerson.setMarried(false);
client.savePerson(savePerson);
transport.close();
}
}


可以看到，客户端调用服务器端接口就像调用本地方法一样简单。

Python处理

安装依赖

pip install thrift


PyCharm中可以这样安装

根据thrift文件创建Python数据类型和接口代码

thrift -out $DST_DIR --gen py $SRC_DIR/data.proto


-out表示生成代码的路径，py表示生成Python语言的代码，实际命令为

.\thrift-0.16.0.exe -out D:\java\code_resp\PycharmProjects\test_protobuf\testthrift --gen py D:\java\code_resp\PycharmProjects\test_protobuf\testthrift\data.thrift


代码结构如下，data目录为生成的目录，包含数据类型和服务接口。

序列化和反序列化

from data import ttypes
from thrift import TSerialization
from thrift.protocol import TCompactProtocol
def test_serialize():
p = ttypes.Person()
p.username = "小明"
p.age = 27
p.married = False
bytes = TSerialization.serialize(p,TCompactProtocol.TCompactProtocolFactory())
with open("D:/testjar/thrift_python_serialize_person", "wb") as f:
f.write(bytes)
return
def test_deserialize():
with open("D:/testjar/thrift_python_serialize_person", "rb") as rf:
bytes_read = rf.read()
p = ttypes.Person()
TSerialization.deserialize(p, bytes_read, TCompactProtocol.TCompactProtocolFactory())
print(p)
return
if __name__ == '__main__':
test_deserialize()
pass


服务器端代码

from PersonServiceImpl import PersonServiceImpl
from data import PersonService
from thrift.protocol import TCompactProtocol
from thrift.server import TServer
from thrift.transport import TSocket
from thrift.transport import TTransport
def test():
# Make socket
persOnServiceImpl= PersonServiceImpl()
processor = PersonService.Processor(personServiceImpl)
serverSocket = TSocket.TServerSocket(port=8883)
transportFactory = TTransport.TFramedTransportFactory()
protocolFactory = TCompactProtocol.TCompactProtocolFactory()
server = TServer.TThreadPoolServer(processor, serverSocket, transportFactory, protocolFactory)
server.serve()
pass
if __name__ == '__main__':
test()
pass


客户端代码

from thrift.transport import TSocket
from thrift.transport import TTransport
from thrift.protocol import TCompactProtocol
from data import PersonService
def test():
# Make socket
transport = TSocket.TSocket('localhost', 8883)
# Buffering is critical. Raw sockets are very slow
transport = TTransport.TFramedTransport(transport)
# Wrap in a protocol
protocol = TCompactProtocol.TCompactProtocol(transport)
# Create a client to use the protocol encoder
client = PersonService.Client(protocol)
# Connect!
transport.open()
p = client.getPersonByUsername("李四")
print(p)
pass
if __name__ == '__main__':
test()
pass


总结

更多Thrift相关用法，请查看官方文档。