C++ Thrift Client 与 Flume Thrift Source 对接
项目需要C++代码与flume对接，进而将日志写入HDFS。flume原生为java代码，原先的解决方案是通过JNI调用flume java方法。
但是由于一来对jni的调用效率的担心，二来C++调用JNI需要照顾local reference和GC的问题，被搞得头痛了。
一怒之下，重写代码了，使用C++与远端的JAVA Flume对接。
在协议的选择上，AVRO C++虽然也有apache的开源项目，但是目前只支持读写文件，而不能使用RPC。
故使用了thrift与远端Flume thrift source通信。
以下是一些实现的具体方法：
Flume thrift 协议准备：
1. 安装thrift，安装flume。
2. 下载flume的source包，解压。
3. 在解压目录flume-ng-sdk/src/main/thrift下，存在文件flume.thrift。
这是flume的thrift协议规则文件，执行命令：thrift -r --gen cpp ./flume.thrift
在./gen-cpp/目录下会生成一些.h .cpp文件，之后C++代码中需要调用其中的方法进行RPC通信。
注意：如果是在windows/visual studio下开发，由于ERROR与关键词冲突，需要对flume.thrift修改。
enum Status {
OK,
FAILED,
FERROR, //原先是ERROR，修改为FERROR。
UNKNOWN
}
如果是linux下工作，则无需改动任何代码。
C++ 代码实现：
1. include一些头文件，以及使用thrift的名字空间
1 #include <thrift/protocol/TBinaryProtocol.h>
2 #include <thrift/protocol/TCompactProtocol.h>
3 #include <thrift/transport/TSocket.h>
4 #include <thrift/transport/TTransportUtils.h>
5
6 using namespace std;
7 using namespace apache::thrift;
8 using namespace apache::thrift::protocol;
9 using namespace apache::thrift::transport;
2. 创建thrift的socket, transport, protocol, client.
1 class ThriftClient{
2 private:
3 /* Thrift protocol needings... */
4 boost::shared_ptr<TTransport> socket;
5 boost::shared_ptr<TTransport> transport;
6 boost::shared_ptr<TProtocol> protocol;
7 ThriftSourceProtocolClient* pClient;
8 }
1 ThriftClient::ThriftClient(std::string inIpAddress, std::string inPort):
2 socket(new TSocket(inIpAddress.c_str(), atoi(inPort.c_str()))),
3 transport(new TFramedTransport(socket)),
4 protocol(new TCompactProtocol(transport))
5 {
6 pClient = new ThriftSourceProtocolClient(protocol);
7 }
3. 与远端flume thrift source通信：
1 bool ThriftClient::sendEvent(const Event* event)
2 {
3 //build the head
4 std::map<std::string, std::string>headers;
5 std::ostringstream timeBuffer;
6 timeBuffer << event->timestamp <<"000";
7 headers.insert(std::make_pair("timestamp", timeBuffer.str()));
8 headers.insert(std::make_pair("appId", appId));
9 //build the body
10 std::ostringstreamosBody;
11 osBody << *event;
12 std::string sBody = osBody.str();
13
14 ThriftFlumeEvent tfEvent;
15 tfEvent.__set_body(sBody);
16 tfEvent.__set_headers(headers);
17 if(!transport->isOpen())
18 {
19 transport->open();
20 }
21 Status::type res=pClient->append(tfEvent);
22 if(res == Status::OK)
23 {
24 return true;
25 }
26 else
27 {
28 printf("WARNING: send event via thrift failed, return code:%d\n",res);
29 return false;
30 }
31 }
其他注意点：
1. append方法用来发送一条event：
Status::type ThriftSourceProtocolClient::append(const ThriftFlumeEvent& event)
同时还有一个方法appendBatch用来一次发送多个event：
Status::type ThriftSourceProtocolClient::appendBatch(const std::vector<ThriftFlumeEvent> & events)
2. 无论是append还是appendBatch方法，都是阻塞方法。
3. 可能可以通过send_append，send_appendBatch来发送无需确认成功的event(未测试)。
void ThriftSourceProtocolClient::send_append(const ThriftFlumeEvent& event)
void ThriftSourceProtocolClient::send_appendBatch(const std::vector<ThriftFlumeEvent> & events)
4. Thrift只提供协议RPC功能，并没有提供flume的channel功能，以及多个source的情况下的load balance功能。这些都需要自己实现。
关于性能：
测试环境： vmware+ubuntu，i3-4150 CPU, 配置1G内存，双核CPU。
在本机配置两个flume thrift source（load balance），网络回环。
在每条event大约50字符的情况下，可以达到16000条每秒的吞吐量，此时CPU被耗尽（两个flume thrift source大约占用30%CPU）。
页:
[1]