## 10) 負載均衡獲取Route信息API(0.7) ### 10.1 proto通信協議定義 > base/proto/lars.proto ```protobuf /* Lars系統的消息ID */ enum MessageId { ID_UNKNOW = 0; //proto3 enum第一個屬性必須是0，用來占位 ID_GetRouteRequest = 1; //向DNS請求Route對應的關系的消息ID ID_GetRouteResponse = 2; //DNS回復的Route信息的消息ID ID_ReportStatusRequest = 3; //上報host調用狀態信息請求消息ID ID_GetHostRequest = 4; //API 發送請求host信息給 Lb Agent模塊 消息ID ID_GetHostResponse = 5; //agent 回執給 API host信息的 消息ID ID_ReportRequest = 6; //API report get_host的調用結果給agent的 消息ID // ======================================================= ID_API_GetRouteRequest = 7; //API 請求agent某個modid/cmdid的全部hosts信息的route 消息ID ID_API_GetRouteResponse = 8; //agent 回執給 API的全部hosts的route信息 消息ID // ======================================================= } ``` ? 增加兩個message ID， `ID_API_GetRouteRequest`和`ID_API_GetRouteResponse`，主要是針對API層獲取route全部的host節點信息通信使用。 ### 10.2 Lars-API:get_route()方法客戶端實現 > api/cpp/lars_api/lars_api.h ```c typedef std::pair<std::string, int> ip_port; typedef std::vector<ip_port> route_set; typedef route_set::iterator route_set_it; ``` > api/cpp/lars_api/lars_api.cpp ```c //lars 系統獲取某modid/cmdid全部的hosts(route)信息 int lars_client::get_route(int modid, int cmdid, route_set &route) { //1. 封裝請求消息 lars::GetRouteRequest req; req.set_modid(modid); req.set_cmdid(cmdid); //2. send char write_buf[4096], read_buf[80*1024]; //消息頭 msg_head head; head.msglen = req.ByteSizeLong(); head.msgid = lars::ID_API_GetRouteRequest; memcpy(write_buf, &head, MESSAGE_HEAD_LEN); //消息體 req.SerializeToArray(write_buf+MESSAGE_HEAD_LEN, head.msglen); //簡單的hash來發給對應的agent udp server int index = (modid + cmdid) %3; int ret = sendto(_sockfd[index], write_buf, head.msglen + MESSAGE_HEAD_LEN, 0, NULL, 0); if (ret == -1) { perror("sendto"); return lars::RET_SYSTEM_ERROR; } //3. recv lars::GetRouteResponse rsp; int message_len = recvfrom(_sockfd[index], read_buf, sizeof(read_buf), 0, NULL, NULL); if (message_len == -1) { perror("recvfrom"); return lars::RET_SYSTEM_ERROR; } //消息頭 memcpy(&head, read_buf, MESSAGE_HEAD_LEN); if (head.msgid != lars::ID_API_GetRouteResponse) { fprintf(stderr, "message ID error!\n"); return lars::RET_SYSTEM_ERROR; } //消息體 ret = rsp.ParseFromArray(read_buf + MESSAGE_HEAD_LEN, message_len - MESSAGE_HEAD_LEN); if (!ret) { fprintf(stderr, "message format error: head.msglen = %d, message_len = %d, message_len - MESSAGE_HEAD_LEN = %d, head msgid = %d, ID_GetHostResponse = %d\n", head.msglen, message_len, message_len-MESSAGE_HEAD_LEN, head.msgid, lars::ID_GetRouteResponse); return lars::RET_SYSTEM_ERROR; } if (rsp.modid() != modid || rsp.cmdid() != cmdid) { fprintf(stderr, "message format error\n"); return lars::RET_SYSTEM_ERROR; } //4 處理消息 for (int i = 0; i < rsp.host_size(); i++) { const lars::HostInfo &host = rsp.host(i); struct in_addr inaddr; inaddr.s_addr = host.ip(); std::string ip = inet_ntoa(inaddr); int port = host.port(); route.push_back(ip_port(ip,port)); } return lars::RET_SUCC; } ``` ### 10.3 Agent UDP Server處理API-get_route請求 > lars_loadbalance_agent/src/agent_udp_server.cpp ```c static void get_route_cb(const char *data, uint32_t len, int msgid, net_connection *net_conn, void *user_data) { //解析api發送的請求包 lars::GetRouteRequest req; req.ParseFromArray(data, len); int modid = req.modid(); int cmdid = req.cmdid(); //設置回執消息 lars::GetRouteResponse rsp; rsp.set_modid(modid); rsp.set_cmdid(cmdid); route_lb *ptr_route_lb = (route_lb*)user_data; //調用route_lb的獲取host方法，得到rsp返回結果 ptr_route_lb->get_route(modid, cmdid, rsp); //打包回執給api消息 std::string responseString; rsp.SerializeToString(&responseString); net_conn->send_message(responseString.c_str(), responseString.size(), lars::ID_API_GetRouteResponse); } void * agent_server_main(void * args) { // .... //給server注冊消息分發路由業務，針對ID_API_GetRouteRequest處理 server.add_msg_router(lars::ID_API_GetRouteRequest, get_route_cb, r_lb[port-8888]); // ... return NULL; } ``` 針對`ID_API_GetRouteRequest`添加一個消息處理方法，在回調業務中，通過`route_lb`的`get_route()`方法獲取信息.我們來實現這個方法。 > lars_loadbalance_agent/src/route_lb.cpp ```c //agent獲取某個modid/cmdid的全部主機，將返回的主機結果存放在rsp中 int route_lb::get_route(int modid, int cmdid, lars::GetRouteResponse &rsp) { int ret = lars::RET_SUCC; //1. 得到key uint64_t key = ((uint64_t)modid << 32) + cmdid; pthread_mutex_lock(&_mutex); //2. 當前key已經存在_route_lb_map中 if (_route_lb_map.find(key) != _route_lb_map.end()) { //2.1 取出對應的load_balance load_balance *lb = _route_lb_map[key]; std::vector<host_info*> vec; lb->get_all_hosts(vec); for (std::vector<host_info*>::iterator it = vec.begin(); it != vec.end(); it++) { lars::HostInfo host; host.set_ip((*it)->ip); host.set_port((*it)->port); rsp.add_host()->CopyFrom(host); } //超時重拉路由 //檢查是否要重新拉路由信息 //若路由并沒有處于PULLING狀態，且有效期已經超時，則重新拉取 if (lb->status == load_balance::NEW && time(NULL) - lb->last_update_time > lb_config.update_timeout) { lb->pull(); } } //3. 當前key不存在_route_lb_map中 else { //3.1 新建一個load_balance load_balance *lb = new load_balance(modid, cmdid); if (lb == NULL) { fprintf(stderr, "no more space to create loadbalance\n"); exit(1); } //3.2 新建的load_balance加入到map中 _route_lb_map[key] = lb; //3.3 從dns service服務拉取具體的host信息 lb->pull(); ret = lars::RET_NOEXIST; } pthread_mutex_unlock(&_mutex); return ret; } ``` 其中，`load_balance`的`get_all_hosts()`方法實現如下： > lars_loadbalance_agent/src/load_balance.cpp ```c //獲取當前掛載下的全部host信息 添加到vec中 void load_balance::get_all_hosts(std::vector<host_info*> &vec) { for (host_map_it it = _host_map.begin(); it != _host_map.end(); it++) { host_info *hi = it->second; vec.push_back(hi); } } ``` 接下來，我們可以簡單測試一些獲取route信息的api > api/cpp/example/example.cpp ```c #include "lars_api.h" #include <iostream> void usage() { printf("usage: ./example [modid] [cmdid]\n"); } int main(int argc, char **argv) { if (argc != 3) { usage(); return 1; } int modid = atoi(argv[1]); int cmdid = atoi(argv[2]); lars_client api; std::string ip; int port; route_set route; int ret = api.get_route(modid, cmdid, route); if (ret == 0) { std::cout << "get route succ!" << std::endl; for (route_set_it it = route.begin(); it != route.end(); it++) { std::cout << "ip = " << (*it).first << ", port = " << (*it).second << std::endl; } } ret = api.get_host(modid, cmdid, ip, port); if (ret == 0) { std::cout << "host is " << ip << ":" << port << std::endl; //上報調用結果 api.report(modid, cmdid, ip, port, 0); } return 0; } ``` --- ### 關于作者： 作者：`Aceld(劉丹冰)` mail: [danbing.at@gmail.com](mailto:danbing.at@gmail.com) github: [https://github.com/aceld](https://github.com/aceld) 原創書籍: [http://www.hmoore.net/@aceld](http://www.hmoore.net/@aceld)  >**原創聲明:未經作者允許請勿轉載, 如果轉載請注明出處**