前提
成功创建:集群,表格组,tb_online表
syntax = "proto2";package myTcaplusTable;import "tcaplusservice.optionv1.proto";message tb_online {option(tcaplusservice.tcaplus_primary_key) = "openid,tconndid,timekey";required int32 openid = 1; //QQ Uinrequired int32 tconndid = 2;required string timekey = 3;required string gamesvrid = 4;optional int32 logintime = 5 [default = 1];repeated int64 lockid = 6 [packed = true]; //repeated 类型字段使用 packed 关键字修饰optional pay_info pay = 7;message pay_info {optional uint64 total_money = 1;optional uint64 pay_times = 2;}}
步骤1:定义配置参数
// 目标集群的访问地址static const char DIR_URL_ARRAY[][TCAPLUS_MAX_STRING_LENGTH] ={"tcp://10.191.***.99:9999","tcp://10.191.***.88:9999"};// 目标集群的地址个数static const int32_t DIR_URL_COUNT = 2;// 目标业务的集群 IDstatic const int32_t APP_ID = 3;// 目标业务的表格组 IDstatic const int32_t ZONE_ID = 1;// 目标业务的业务密码static const char * SIGNATURE = "*******";// 目标业务的表名 tb_onlinestatic const char * TABLE_NAME = "tb_online";
步骤2:初始化 TcaplusPB 客户端
//Tcaplus PB API 客户端TcaplusAsyncPbApi g_stAsyncApi;int32_t InitAsyncPbApi(){//PB API 配置ClientOptions cfg;cfg.app_id = APP_ID;cfg.zones.push_back(ZONE_ID);strcpy(cfg.signature, SIGNATURE);for (int32_t i = 0; i < DIR_URL_COUNT; i++){cfg.dirs.push_back(DIR_URL_ARRAY[i]);}//访问的 PB 表cfg.tables.push_back(TABLE_NAME);//日志配置strncpy(cfg.log_cfg, "tlogconf.xml", sizeof(cfg.log_cfg));//初始化连接超时时间5scfg.timeout = 5000;//初始化连接int32_t iRet = g_stAsyncApi.Init(cfg);if (0 != iRet){cout << "ERROR: g_stAsyncApi.Init failed, log cfg: " << cfg.log_cfg << ", iRet: " << iRet << "." << endl;return iRet;}return iRet;}
步骤3:定义异步回调
//收到响应消息计数uint32_t g_dwTotalRevNum = 0;class CommonCallback : public TcaplusPbCallback{public:CommonCallback(){cout << "Init CommonCallback." << endl;}~CommonCallback(){cout << "Fini ~CommonCallback." << endl;}//收到响应消息的回调,msgs 为获取的记录的数组int OnRecv(const std::vector< ::google::protobuf::Message *> &msgs){cout << "OnRecv[" << msgs.size() << endl;g_dwTotalRevNum++;for (size_t idx = 0; idx < msgs.size(); idx++){tb_online* t = dynamic_cast<tb_online *>(msgs[idx]);if (NULL == t){cout << "ERROR: msgs[" << idx << "] not tb_online type." << endl;return -1;}cout << "---------- receive a response----------:" << endl;cout << "openid=" << t->openid() << endl;cout << "tconndid=" << t->tconndid() << endl;cout << "timekey=" << t->timekey() << endl;cout << "gamesvrid=" << t->gamesvrid() << endl;cout << "logintime=" << t->logintime() << endl;for (int32_t i = 0; i < t->lockid_size();i++){cout << "lockid[" << i << "]=" << t->lockid(i) << endl;}tb_online_pay_info pay = t->pay();cout << "pay total_money=" << pay.total_money() << "; pay_times=" << pay.pay_times() << endl;//获取记录的版本号std::string version;int iRet = g_stAsyncApi.GetMessageOption(*t, NS_TCAPLUS_PROTOBUF_API::MESSAGE_OPTION_DATA_VERSION, &version);cout << "after GetMessageOption iRet= [" << iRet << "] version:" << version.c_str() << " msg:" << t << endl;}return 0;}//收到错误响应消息的回调int OnError(const std::vector< ::google::protobuf::Message *> &msgs, int errorcode){cout << "OnError[" << msgs.size() << endl;g_dwTotalRevNum++;for (size_t idx = 0; idx < msgs.size(); idx++){tb_online* t = dynamic_cast<tb_online *>(msgs[idx]);if (NULL == t){cout << "ERROR: msgs[" << idx << "] not tb_online type." << endl;return -1;}if (TcapErrCode::TXHDB_ERR_RECORD_NOT_EXIST == errorcode){cout << "ERROR: openid= " << t->openid() << ", tconndid= " << t->tconndid() << ", timekey= " << t->timekey() << ", record not exists" << endl;}cout << "ERROR: openid = [" << t->openid() << "], tconndid = [" << t->tconndid() << "], timekey =[" << t->timekey() << "] failed:%d" << errorcode << endl;}return 0;}//消息超时的回调int OnTimeout(const std::vector< ::google::protobuf::Message *> &msgs){cout << "OnTimeout[" << msgs.size() << endl;for (size_t idx = 0; idx < msgs.size(); idx++){tb_online* t = dynamic_cast<tb_online *>(msgs[idx]);if (NULL == t){cout << "TIMEOUT: msgs[" << idx << "] not tb_online type!" << endl;return -1;}cout << "TIMEOUT: openid = [" << t->openid() << "], tconndid = [" << t->tconndid() << "], timekey =[" << t->timekey() << "] timeout" << endl;}return 0;}int OnFinish(const NS_TCAPLUS_PROTOBUF_API::MsgParam ¶m){cout << "OnFinish: " << param.m_nOperation << " req: " << param.m_vecMsgs.size()<< endl;return 0;}};
步骤4:发送 delete 请求
int SendDelRequest(){static tb_online t;t.set_openid(1);t.set_tconndid(1);t.set_timekey("test_tcaplus");static CommonCallback cb;std::string version = "-1";g_stApi.SetMessageOption(t, NS_TCAPLUS_PROTOBUF_API::MESSAGE_OPTION_DATA_VERSION, version);int32_t iRet = g_stAsyncApi.Del(&t, &cb);if (iRet != TcapErrCode::GEN_ERR_SUC){cout << "ERROR: openid= " << t.openid() << ", tconndid= " << t.tconndid() << ", timekey= " << t.timekey() << ", Delete Error iRet = " << iRet << endl;return -1;}return 0;}
示例
int main(void) {//初始化 API 客户端int ret = InitAsyncPbApi();if ( ret != 0){printf("InitAsyncPbApi failed\\n");return -1;}//发送请求ret = SendDelRequest();if (0 != ret){printf("SendDelRequest failed\\n");return -1;}//接收响应do{// 更新,接收回包g_stAsyncApi.UpdateNetwork();usleep(1000 * 10);} while (g_dwTotalRevNum != 1);return 0;}