Thrift学习笔记
一：thrift介绍
Thrift是facebook开发的用来处理各不同系统之间数据通讯的rpc服务框架，后来成为apche的开源项目。thrift支持多种程序语言，包括Java,Python,Ruby,JavaScript,Node.js,Go,C,C++,C#,Erlang,Delphi,Perl,Php,SmallTalk,OCaml,Haxe,Haskell,D语言。Thrift采用IDL(Interface Defination Language)描述性语言来定义数据结构和接口。Thrift模型如下所示：
图 thrift模型图
二 thrift数据传输协议
TBinaryProtocol 二进制传输协议
TCompactProtocol 使用VLQ编码进行压缩的数据传输协议
TJSONProtocol JSON格式的数据传输协议
TSimpleJSONProtocol 简单的JSON格式数据传输协议
TDebugProtocol 调试时使用的文本传输协议
三 thrift传输层
TFramedTransport 按块的大小进行传输
TFileTransport 按照文件的方式进行传输
TMemoryTransport 使用内存IO方式进行传输
TZlibTransport 执行zlib压缩方式传输
四 thrift服务器端
TSimpleServer 简单的单线程标准阻塞式服务器
TThreadPoolServer 多线程阻塞式服务器
TNonblockingServer 多线程非阻塞式服务器
THsHaServer 半同步半异步服务器
其实传输层的传输只有阻塞和非阻塞，再加上具体的工作方式 单线程 多线程
五 thrift客户端
TClient 简单单线程阻塞式客户端
TAsynClient 异步客户端(多线程)
六 thrift开发步骤
1服务器端
实现服务处理接口impl
创建TProcessor
创建TServerTransport(TServerSocket) 创建阻塞通信的还是非阻塞通信
创建TProtocol 数据传输协议
创建TServer 服务器类型 单工(单线程) 双工(多线程) 半单工半双工(多线程)
启动Server
2客户端
创建Transport(TSocket) 创建阻塞通信(客户端只有阻塞)
创建TProtocol 数据传输协议
基于TTransport和TProtocol创建Client
调用Client的相应方法
七 thrift数据类型
1基本类型
bool：布尔值，true 或 false
byte：8 位有符号整数
i16：16 位有符号整数
i32：32 位有符号整数
i64：64 位有符号整数
double：64 位浮点数
string：utf-8编码的字符串
2结构体类型
struct：定义公共的对象
enum: 枚举类型
3容器类型
list：对应 Java 的 ArrayList
set：对应 Java 的 HashSet
map：对应 Java 的 HashMap
4异常类型
exception：对应 Java 的 Exception
5服务类型
service：对应服务的类 提供接口
八 thrift例子
enum 类型
struct Student{
1: required i32 id
2: required string username
3: required string password
4: requried string number
5: optional double age
}
struct 类型
struct School{
1: required i32 id
2: required string name
3: required set<Student> students
4: required list<Student> rank
5: required map<string, string> number_name
}
service 类型
service ThriftMysqlService{
void addUser(1:Student user)
list<Student> queryAllUser()
Student queryOneUser(1:i32 id)
map<string, string> queryOneArticle(1:i32 id)
}
具体代码
thrift.thrift 定义数据类型和接口的文件
namespace java org.seava.thrift_example.thrift
struct User{
1: required i32 userId
2: required string username
3: required string password
}
service ThriftService{
void addUser(1:User user)
User queryUser(1:i32 id)
list<User> queryUserList()
map<string, string> queryUserNamePass()
map<i32, User> queryUserMap()
}
到apache的官网下载thrift.exe程序， 下载地址 http://thrift.apache.org/ ，下下来通过cmd命令窗口去运行如下命令
thrift -gen java xxx.thrift
接口实现 ThriftServiceImpl.java
package org.seava.thrift_example.thrift; import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map; public class ThriftServiceImpl implements ThriftService.Iface { public void addUser(User user) throws org.apache.thrift.TException{
System.out.println(user.userId + " " + user.username + " " + user.password);
} public User queryUser(int id) throws org.apache.thrift.TException{
System.out.println(id);
User user = new User();
user.userId = 100;
user.username = "FFF";
user.password = "NNN";
return user;
} public List<User> queryUserList() throws org.apache.thrift.TException{
User user = new User();
user.userId = 100;
user.username = "FFF";
user.password = "NNN";
User user2 = new User();
user2.userId = 102;
user2.username = "FFF2";
user2.password = "NNN2";
List<User> list = new ArrayList<User>();
list.add(user2);
list.add(user);
return list;
} public Map<String,String> queryUserNamePass() throws org.apache.thrift.TException{
User user = new User();
user.userId = 100;
user.username = "FFF";
user.password = "NNN";
Map<String, String> map = new HashMap<String, String>();
map.put("password", user.password);
map.put("useranme", user.username);
return map;
} public Map<Integer,User> queryUserMap() throws org.apache.thrift.TException{
User user = new User();
user.userId = 100;
user.username = "FFF";
user.password = "NNN";
User user2 = new User();
user2.userId = 102;
user2.username = "FFF2";
user2.password = "NNN2";
Map<Integer, User> map = new HashMap<Integer, User>();
map.put(user.userId, user);
map.put(user2.userId, user2);
return map;
} }
服务器 Server.java
package org.seava.thrift_example.thrift; import org.apache.thrift.TProcessor;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TCompactProtocol;
import org.apache.thrift.server.THsHaServer;
import org.apache.thrift.server.TNonblockingServer;
import org.apache.thrift.server.TServer;
import org.apache.thrift.server.TSimpleServer;
import org.apache.thrift.server.TThreadPoolServer;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TNonblockingServerSocket;
import org.apache.thrift.transport.TNonblockingServerTransport;
import org.apache.thrift.transport.TServerSocket;
import org.apache.thrift.transport.TTransportException; public class Server { public static int port = 8090; /**
* 简单服务器类型 阻塞单线程
* 步骤
* 创建TProcessor
* 创建TServerTransport
* 创建TProtocol
* 创建TServer
* 启动Server
*/
public static void startSimpleServer(){
//创建processor
TProcessor tprocessor = new ThriftService.Processor<ThriftService.Iface>(new ThriftServiceImpl());
try {
//创建transport 阻塞通信
TServerSocket serverTransport = new TServerSocket(port);
//创建protocol
TBinaryProtocol.Factory protocol = new TBinaryProtocol.Factory();
//将processor transport protocol设入到服务器server中
TServer.Args args = new TServer.Args(serverTransport);
args.processor(tprocessor);
args.protocolFactory(protocol);
//定义服务器类型 设定参数
TServer server = new TSimpleServer(args);
//开启服务
server.serve();
} catch (TTransportException e) {
e.printStackTrace();
}
} /**
* 多线程服务器 阻塞多线程
*/
public static void startThreadPoolServer(){
//创建processor
TProcessor tprocessor = new ThriftService.Processor<ThriftService.Iface>(new ThriftServiceImpl());
try{
//创建transport 阻塞通信
TServerSocket serverTransport = new TServerSocket(port);
//创建protocol 数据传输协议
TBinaryProtocol.Factory protocol = new TBinaryProtocol.Factory();
TThreadPoolServer.Args args = new TThreadPoolServer.Args(serverTransport);
args.processor(tprocessor);
args.protocolFactory(protocol);
//创建服务器类型 多线程
TServer server = new TThreadPoolServer(args);
//开启服务
server.serve();
}catch(Exception e){
e.printStackTrace();
}
} /**
* 非阻塞I/O
*/
public static void startTNonblockingServer(){
//创建processor
TProcessor tprocessor = new ThriftService.Processor<ThriftService.Iface>(new ThriftServiceImpl());
try{
//创建transport 非阻塞 nonblocking
TNonblockingServerTransport serverTransport = new TNonblockingServerSocket(port);
//创建protocol 数据传输协议
TCompactProtocol.Factory protocol = new TCompactProtocol.Factory();
//创建transport 数据传输方式 非阻塞需要用这种方式传输
TFramedTransport.Factory transport = new TFramedTransport.Factory();
TNonblockingServer.Args args = new TNonblockingServer.Args(serverTransport);
args.processor(tprocessor);
args.transportFactory(transport);
args.protocolFactory(protocol);
//创建服务器 类型是非阻塞
TServer server = new TNonblockingServer(args);
//开启服务
server.serve();
}catch(Exception e){
e.printStackTrace();
}
} /**
* 半同步半异步的非阻塞I/O
*/
public static void startTHsHaServer(){
//创建processor
TProcessor tprocessor = new ThriftService.Processor<ThriftService.Iface>(new ThriftServiceImpl());
try{
//创建transport 非阻塞
TNonblockingServerTransport serverTransport = new TNonblockingServerSocket(port);
//非阻塞需要的传输方式
TFramedTransport.Factory transport = new TFramedTransport.Factory();
//数据传输协议
TCompactProtocol.Factory protocol = new TCompactProtocol.Factory();
//创建半同步半异步服务
THsHaServer.Args args = new THsHaServer.Args(serverTransport);
args.processor(tprocessor);
args.transportFactory(transport);
args.protocolFactory(protocol);
//创建 服务类型
TServer server = new THsHaServer(args);
//开启服务
server.serve();
}catch(Exception e){
e.printStackTrace();
}
} public static void main(String args[]){
//开启简单服务器
// Server.startSimpleServer();
//开启多线程服务器
// Server.startThreadPoolServer();
// Server.startTNonblockingServer();
// Server.startTHsHaServer();
Server.startTNonblockingServer();
}
}
Server.java实现了简单服务器(阻塞单线程) 阻塞多线程 非阻塞 半同步半异步非阻塞
注意： 非阻塞时传输层需要选择TFramedTransport
客户端 Client.java
package org.seava.thrift_example.thrift; import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit; import org.apache.thrift.async.TAsyncClientManager;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TCompactProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.protocol.TProtocolFactory;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TNonblockingSocket;
import org.apache.thrift.transport.TNonblockingTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport; public class Client implements Runnable { public static String ip = "localhost";
public static int port = 8090;
public static int time_out = 30000; /**
* 客户端设置
* 创建Transport
* 创建TProtocol
* 基于TTransport和TProtocol创建Client
* 调用Client的相应方法
*/
public static void startSimpleClient(){
TTransport transport = null;
try{
//创建Transport
transport = new TSocket(ip, port, time_out);
//创建TProtocol
TProtocol protocol = new TBinaryProtocol(transport);
//基于TTransport和TProtocol创建Client
ThriftService.Client client = new ThriftService.Client(protocol);
transport.open();
//调用client方法
List<User> list = client.queryUserList();
for(User user : list){
System.out.println(user.userId + " " + user.username + " " + user.password);
}
Map<String, String> map = client.queryUserNamePass();
System.out.println(map);
User user = client.queryUser(10);
System.out.println(user.userId + " " + user.username + " " + user.password);
Map<Integer, User> map_u = client.queryUserMap();
System.out.println(map_u);
User uu = new User();
uu.userId = 1111;
uu.username = "mmbbmmbb";
uu.password = "ppbbppbb";
client.addUser(uu);
}catch(Exception e){
e.printStackTrace();
}
} /**
* 调用阻塞服务器的客户端
*/
public static void startNonblockingClient(){
TTransport transport = null;
try{
transport = new TFramedTransport(new TSocket(ip, port));
TCompactProtocol protocol = new TCompactProtocol(transport);
ThriftService.Client client = new ThriftService.Client(protocol);
transport.open();
//调用client方法
List<User> list = client.queryUserList();
for(User user : list){
System.out.println(user.userId + " " + user.username + " " + user.password);
}
Map<String, String> map = client.queryUserNamePass();
System.out.println(map);
User user = client.queryUser(10);
System.out.println(user.userId + " " + user.username + " " + user.password);
Map<Integer, User> map_u = client.queryUserMap();
System.out.println(map_u);
User uu = new User();
uu.userId = 1111;
uu.username = "mmbbmmbb";
uu.password = "ppbbppbb";
client.addUser(uu);
}catch(Exception e){
e.printStackTrace();
}
} public static void startAsynClient(){
try{
TAsyncClientManager clientManager = new TAsyncClientManager();
TNonblockingTransport transport = new TNonblockingSocket(ip, port, time_out);
TProtocolFactory tprotocol = new TCompactProtocol.Factory();
ThriftService.AsyncClient asyncClient = new ThriftService.AsyncClient(tprotocol, clientManager, transport);
System.out.println("Client start ...");
CountDownLatch latch = new CountDownLatch(1);
AsynCallback callBack = new AsynCallback(latch);
System.out.println("call method queryUser start ...");
asyncClient.queryUser(100, callBack);
System.out.println("call method queryUser end");
boolean wait = latch.await(30, TimeUnit.SECONDS);
System.out.println("latch.await =:" + wait);
}catch(Exception e){
e.printStackTrace();
}
} public void run(){
Client.startSimpleClient();
} public static void main(String args[]){
//调用简单服务器
// Client.startSimpleClient();
/*Client c1 = new Client();
Client c2 = new Client(); new Thread(c1).start();
new Thread(c2).start();*/ // Client.startNonblockingClient();
// Client.startNonblockingClient();
Client.startAsynClient();
}
}
客户端实现了 阻塞单线程 和 异步客户端
具体代码在github上：https://github.com/WaterHsu/thrift-example.git