假设用户编写了以下Thrift文件:
struct LogInfo {
1: required string name,
2: optional string content,
}
service LogSender {
void SendLog(1:list<LogInfo> loglist);
}
用户使用命令“thrift –gen cpp example.thrift”可生成C++代码,该代码包含以下文件:
example_constants.h
example_constants.cpp
example_types.h //struct定义
example_types.cpp //struct实现
LogSender.h //service定义
LogSender.cpp //service实现和LogSenderClient实现
LogSender_server.skeleton.cpp //一个实例RPC Server
用户可以这样编写Client:
shared_ptr socket(new TSocket(“8.8.8.8″, 9090));
shared_ptr transport(new TBufferedTransport(socket));
shared_ptr PRotocol(new TBinaryProtocol(transport));
LogSenderClient client(protocol);
try {
transport->open();
vector<LogInfo> logInfos;
LogInfo logInfo(“image”, “10:9:0 visit:xxxxxx”);
logInfos.push_back(logInfo);
…..
client.SendLog(logInfos);
transport->close();
} catch (TException &tx) {
printf(“ERROR: %s/n”, tx.what());
}
为了深入分析这段代码,我们看一下client.SendLog()函数的内部实现(在LogSender.cpp中):
void LogSenderClient::SendLog(const std::vector<LogInfo> & loglist)
{
send_SendLog(loglist);
recv_SendLog();
}
void LogSenderClient::send_SendLog(const std::vector<LogInfo> & loglist)
{
int32_t cseqid = 0;
oprot_->writeMessageBegin(“SendLog”, ::apache::thrift::protocol::T_CALL, cseqid);
LogSender_SendLog_pargs args;
args.loglist = &loglist;
args.write(oprot_);
oprot_->writeMessageEnd();
oprot_->getTransport()->flush();
oprot_->getTransport()->writeEnd();
}
void LogSenderClient::recv_SendLog()
{
int32_t rseqid = 0;
std::string fname;
::apache::thrift::protocol::TMessageType mtype;
iprot_->readMessageBegin(fname, mtype, rseqid);
if (mtype == ::apache::thrift::protocol::T_EXCEPTION) {
…..
}
if (mtype != ::apache::thrift::protocol::T_REPLY) {
……
}
if (fname.compare(“SendLog”) != 0) {
……
}
LogSender_SendLog_presult result;
result.read(iprot_);
iprot_->readMessageEnd();
iprot_->getTransport()->readEnd();
return;
}
阅读上面的代码,可以看出,RPC函数SendLog()实际上被转化成了两个函数:send_SendLog和recv_SendLog,分别用于发送数据和接收结果。数据是以消息的形式表示的,消息头部是RPC函数名,消息内容是RPC函数的参数。
我们再进一步分析RPC Server端,一个server的编写方法(在LogSender.cpp中)如下:
shared_ptr protocolFactory(new TBinaryProtocolFactory());
shared_ptr handler(new LogSenderHandler());
shared_ptr processor(new LogSenderProcessor(handler));
shared_ptr serverTransport(new TServerSocket(9090));
shared_ptr transportFactory(new TBufferedTransportFactory());
TSimpleServer server(processor,
serverTransport,
transportFactory,
protocolFactory);
printf(“Starting the server…/n”);
server.serve();
Server端最重要的类是LogSenderProcessor,它内部有一个映射关系processMap_,保存了所有RPC函数名到函数实现句柄的映射,对于LogSender而言,它只保存了一个RPC映射关系:
processMap_[" SendLog"] = &LogSenderProcessor::process_SendLog;
其中,process_SendLog是一个函数指针,它的实现如下:
void LogSenderProcessor::process_SendLog(int32_t seqid, ::apache::thrift::protocol::TProtocol* iprot, ::apache::thrift::protocol::TProtocol* oprot)
{
LogSender_SendLog_args args;
args.read(iprot);
iprot->readMessageEnd();
iprot->getTransport()->readEnd();
LogSender_SendLog_result result;
try {
iface_->SendLog(args.loglist);//调用用户编写的函数
} catch (const std::exception& e) {
……
}
oprot->writeMessageBegin(“SendLog”, ::apache::thrift::protocol::T_REPLY, seqid);
result.write(oprot);
oprot->writeMessageEnd();
oprot->getTransport()->flush();
oprot->getTransport()->writeEnd();
}
LogSenderProcessor中一个最重要的函数是process(),它是服务器的主体函数,服务器端(socket server)监听到客户端有请求到达后,会检查消息类型,并检查processMap_映射,找到对应的消息处理函数,并调用之(注意,这个地方可以采用各种并发模型,比如one-request-one-thread,thread pool等)。
通过上面的分析可以看出,Thrift最重要的组件是编译器(采用C++编写),它为用户生成了网络通信相关的代码,从而大大减少了用户的编码工作。
新闻热点
疑难解答