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的名字空間

#include <thrift/protocol/TBinaryProtocol.h>
#include <thrift/protocol/TCompactProtocol.h>
#include <thrift/transport/TSocket.h>
#include <thrift/transport/TTransportUtils.h>

using namespace std;
using namespace apache::thrift;
using namespace apache::thrift::protocol;
using namespace apache::thrift::transport;

2. 創建thrift的socket, transport, protocol, client.

class ThriftClient{
    private:
    /* Thrift protocol needings... */
    boost::shared_ptr<TTransport> socket;
    boost::shared_ptr<TTransport> transport;
    boost::shared_ptr<TProtocol> protocol;
    ThriftSourceProtocolClient* pClient;
}

    ThriftClient::ThriftClient(std::string inIpAddress, std::string inPort):
        socket(new TSocket(inIpAddress.c_str(), atoi(inPort.c_str()))),
        transport(new TFramedTransport(socket)),
        protocol(new TCompactProtocol(transport))
    {
        pClient = new ThriftSourceProtocolClient(protocol);
    }

 

3. 與遠端flume thrift source通信： 

      bool ThriftClient::sendEvent(const Event* event)
      {
          //build the head
          std::map<std::string, std::string>  headers;
          std::ostringstream timeBuffer;
          timeBuffer << event->timestamp <<"000";
          headers.insert(std::make_pair("timestamp", timeBuffer.str()));
          headers.insert(std::make_pair("appId", appId));
          //build the body
          std::ostringstream  osBody;
          osBody << *event;
          std::string sBody = osBody.str();
 
          ThriftFlumeEvent tfEvent;
          tfEvent.__set_body(sBody);
          tfEvent.__set_headers(headers);
          if(!transport->isOpen())
          {
              transport->open();
          }
          Status::type res=pClient->append(tfEvent);
          if(res == Status::OK)
          {
              return true;
          }
          else
          {
              printf("WARNING: send event via thrift failed, return code:%d\n",res);
              return false;
          }
      }

 

其他注意點：

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）。