## 6) 負載均衡獲取Host主機信息API ? 接下來，我們來就來實現Lars系統第一個暴露給業務開發者的API，`get_host`獲取主機信息。 ### 6.1 Lars-API:GetHost()方法客戶端實現 ? 我們首先先提供一個`C++`語言的API接口層，以后根據不同的業務的實現語言，可以多實現一些其他語言的API接口層。 在`/Lars`下創建`api/`文件夾. ```bash /Lars/api/ └── cpp/ ├── example/ │?? ├── example.cpp │?? └── Makefile ├── lars_api/ │?? ├── lars_api.cpp │?? ├── lars_api.h │?? └── Makefile └── lib/ ``` ? > lars_api/lars_api.h ```c #pragma once #include "lars_reactor.h" #include <string> class lars_client { public: lars_client(); ~lars_client(); //lars 系統獲取host信息 得到可用host的ip和port int get_host(int modid, int cmdid, std::string& ip, int &port); private: int _sockfd[3]; //3個udp socket fd 對應agent 3個udp server uint32_t _seqid; //消息的序列號 }; ``` > lars_api/lars_api.cpp ```c #include "lars_api.h" #include "lars.pb.h" #include <string.h> #include <sys/socket.h> #include <netinet/in.h> #include <arpa/inet.h> lars_client::lars_client():_seqid(0) { printf("lars_client()\n"); //1 初始化服務器地址 struct sockaddr_in servaddr; bzero(&servaddr, sizeof(servaddr)); servaddr.sin_family = AF_INET; //默認的ip地址是本地，因為是API和agent應該部署于同一臺主機上 inet_aton("127.0.0.1", &servaddr.sin_addr); //2. 創建3個UDPsocket for (int i = 0; i < 3; i ++) { _sockfd[i] = socket(AF_INET, SOCK_DGRAM | SOCK_CLOEXEC, IPPROTO_UDP); if (_sockfd[i] == -1) { perror("socket()"); exit(1); } //agent的3個udp端口默認為8888，8889, 8890 servaddr.sin_port = htons(8888 + i); int ret = connect(_sockfd[i], (const struct sockaddr *)&servaddr, sizeof(servaddr)); if (ret == -1) { perror("connect()"); exit(1); } printf("connection agent udp server succ!\n"); } } lars_client::~lars_client() { printf("~lars_client()\n"); for (int i = 0; i < 3; ++i) { close(_sockfd[i]); } } //lars 系統獲取host信息 得到可用host的ip和port int lars_client::get_host(int modid, int cmdid, std::string &ip, int &port) { //從本地agent service獲取 host信息 uint32_t seq = _seqid++; //1. 封裝請求信息 lars::GetHostRequest req; req.set_seq(seq);//序列編號 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_GetHostRequest; 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 int message_len; lars::GetHostResponse rsp; do { 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_GetHostResponse) { 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_GetHostResponse); return lars::RET_SYSTEM_ERROR; } } while (rsp.seq() < seq); if (rsp.seq() != seq || rsp.modid() != modid || rsp.cmdid() != cmdid) { fprintf(stderr, "message format error\n"); return lars::RET_SYSTEM_ERROR; } //4 處理消息 if (rsp.retcode() == 0) { lars::HostInfo host = rsp.host(); struct in_addr inaddr; inaddr.s_addr = host.ip(); ip = inet_ntoa(inaddr); port = host.port(); } return rsp.retcode();//lars::RET_SUCC } ``` ? 模擬Lars的支持的傳輸協議，發送udp請求，其中message的ID是`ID_GetHostRequest`，返回的消息結構類型是`GetHostResponse`。每個消息有一個seq序列號，防止udp的丟包和消息不對稱情況。 > lars_api/Makefile ```makefile TARGET= ../lib/liblarsclient.a CXX=g++ CFLAGS=-g -O2 -Wall -Wno-deprecated BASE=../../../base BASE_H=$(BASE)/include PROTO = $(BASE)/proto PROTO_H = $(BASE)/proto LARS_REACTOR=../../../lars_reactor LARS_REACTOR_H =$(LARS_REACTOR)/include LARS_REACTOR_LIB=$(LARS_REACTOR)/lib -llreactor OTHER_LIB = -lpthread -ldl -lprotobuf SRC= ./ INC= -I./include -I$(BASE_H) -I$(LARS_REACTOR_H) -I$(PROTO_H) LIB= -L$(LARS_REACTOR_LIB) $(OTHER_LIB) OBJS = $(addsuffix .o, $(basename $(wildcard $(SRC)/*.cpp))) OBJS += $(PROTO)/lars.pb.o $(TARGET): $(OBJS) mkdir -p ../lib ar cqs $@ $^ %.o: %.cpp $(CXX) $(CFLAGS) -c -o $@ $< $(INC) .PHONY: clean clean: -rm -f ./*.o $(PROTO)/lars.pb.o $(TARGET) ``` ? 最終生成一個liblarsclient.a靜態庫，供開發者使用。 接下來我們來實現一個模擬的業務端，example/ > 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; int ret = api.get_host(modid, cmdid, ip, port); if (ret == 0) { std::cout << "host is " << ip << ":" << port << std::endl; //上報調用結果 } return 0; } ``` > example/Makefile ```makefile CXX = g++ CFLAGS = -g -O2 -Wall -Wno-deprecated LARS_REACTOR=../../../lars_reactor LARS_REACTOR_H =$(LARS_REACTOR)/include LARS_REACTOR_LIB=$(LARS_REACTOR)/lib -llreactor all: $(CXX) $(CFLAGS) -o example example.cpp -I$(LARS_REACTOR_H) -I../lars_api ../lib/liblarsclient.a -lprotobuf .PHONY: clean clean: -rm ./example ``` ### 6.2 Agent UDP Server處理API-GetHost請求 ```bash API:get_host()—>Agent UDP Server—>route_lb—>load_balance ``` 如上所示，get_host的請求消息會依次經過agent udp server處理`ID_GetHostRequest`,然后交給某個`route_lb` 那么，API一旦發送get_host, Agent UDP Server需要添加一個對應的處理該消息的路由處理業務。 > lars_loadbalance_agent/src/agent_udp_server.cpp ```c void * agent_server_main(void * args) { long index = (long)args; short port = index + 8888; event_loop loop; udp_server server(&loop, "0.0.0.0", port); //給server注冊消息分發路由業務, 每個udp擁有一個對應的route_lb server.add_msg_router(lars::ID_GetHostRequest, get_host_cb, r_lb[port-8888]); printf("agent UDP server :port %d is started...\n", port); loop.event_process(); return NULL; } ``` 其中`get_host_cb()`實現如下： > lars_loadbalance_agent/src/agent_udp_server.cpp ```c static void get_host_cb(const char *data, uint32_t len, int msgid, net_connection *net_conn, void *user_data) { printf("get_host_cb is called ....\n"); //解析api發送的請求包 lars::GetHostRequest req; req.ParseFromArray(data, len); int modid = req.modid(); int cmdid = req.cmdid(); //設置回執消息 lars::GetHostResponse rsp; rsp.set_seq(req.seq()); 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_host(modid, cmdid, rsp); //打包回執給api消息 std::string responseString; rsp.SerializeToString(&responseString); net_conn->send_message(responseString.c_str(), responseString.size(), lars::ID_GetHostResponse); printf("rspstring size = %d\n", responseString.size()); } ``` ? 這里面實際上的最終業務，交給了route_lb的`get_host()`方法，而且其中`lars::GetHostResponse rsp;`做為函數的返回參數類型。 ? `r_lb`是全局定義的3個route_lb對象.定義實現如下. > lars_loadbalance_agent/src/main_server.cpp ```c //每個Agent UDP server的 負載均衡器路由 route_lb route_lb * r_lb[3]; static void create_route_lb() { for (int i = 0; i < 3; i++) { int id = i + 1; //route_lb的id從1 開始計數 r_lb[i] = new route_lb(id); if (r_lb[i] == NULL) { fprintf(stderr, "no more space to new route_lb\n"); exit(1); } } } static void init_lb_agent() { //1. 加載配置文件 config_file::setPath("./conf/lars_lb_agent.conf"); lb_config.probe_num = config_file::instance()->GetNumber("loadbalance", "probe_num", 10); lb_config.init_succ_cnt = config_file::instance()->GetNumber("loadbalance", "init_succ_cnt", 180); //2. 初始化3個route_lb模塊 create_route_lb(); } int main(int argc, char **argv) { //1 初始化環境 init_lb_agent(); //1.5 初始化LoadBalance的負載均衡器 //2 啟動udp server服務,用來接收業務層(調用者/使用者)的消息 start_UDP_servers(); // ... return 0; } ``` ? 接下來我們在看看`route_lb->get_host(modid, cmdid, rsp);`的實現。 > lars_loadbalance_agent/src/route_lb.cpp ```c #include "route_lb.h" #include "lars.pb.h" //構造初始化 route_lb::route_lb(int id):_id(id) { pthread_mutex_init(&_mutex, NULL); } //agent獲取一個host主機，將返回的主機結果存放在rsp中 int route_lb::get_host(int modid, int cmdid, lars::GetHostResponse &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]; if (lb->empty() == true) { //存在lb 里面的host為空，說明正在pull()中，還沒有從dns_service返回來,所以直接回復不存在 assert(lb->status == load_balance::PULLING); rsp.set_retcode(lars::RET_NOEXIST); } else { ret = lb->choice_one_host(rsp); rsp.set_retcode(ret); //TODO 超時重拉路由 } } //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(); //3.4 設置rsp的回執retcode rsp.set_retcode(lars::RET_NOEXIST); ret = lars::RET_NOEXIST; } pthread_mutex_unlock(&_mutex); return ret; } ``` get_host在獲取host的時候是一個動態的獲取模式，如果根據當前的modid/cmdid請求的load_balance模塊來獲取，如果load_balance存在，則直接調用`load_balance`的`choice_one_host()`方法獲取。 如果load_balance不存在，需要新建load_balance，并且當前的load_balance所攜帶的host信息，需要從遠程dns service下載拉取下來，調用`pull()`方法來實現。 #### **本端選擇host信息(load_balance存在情況)** > lars_loadbalance_agent/src/load_balance.cpp ```c //從一個host_list中得到一個節點放到GetHostResponse 的HostInfo中 static void get_host_from_list(lars::GetHostResponse &rsp, host_list &l) { //選擇list中第一個節點 host_info *host = l.front(); //HostInfo自定義類型,proto3并沒提供set方法,而是通過mutable_接口返回HostInfo的指針,可根據此指針進行賦值操作 lars::HostInfo* hip = rsp.mutable_host(); hip->set_ip(host->ip); hip->set_port(host->port); //將上面處理過的第一個節點放在隊列的末尾 l.pop_front(); l.push_back(host); } //從兩個隊列中獲取一個host給到上層 int load_balance::choice_one_host(lars::GetHostResponse &rsp) { //1 判斷_dile_list隊列是否已經空,如果空表示沒有空閑節點 if (_idle_list.empty() == true) { // 1.1 判斷是否已經超過了probe_num if (_access_cnt >= lb_config.probe_num) { _access_cnt = 0; //從 overload_list中選擇一個已經過載的節點 get_host_from_list(rsp, _overload_list); } else { //明確返回給API層，已經過載了 ++_access_cnt; return lars::RET_OVERLOAD; } } else { // 2 判斷過載列表是否為空 if (_overload_list.empty() == true) { //2.1 當前modid/cmdid所有節點均為正常 //選擇一個idle節點 get_host_from_list(rsp, _idle_list); } else { //2.2 有部分節點過載 //判斷訪問次數是否超過probe_num閾值,超過則從overload_list取出一個 if (_access_cnt >= lb_config.probe_num) { _access_cnt = 0; get_host_from_list(rsp, _overload_list); } else { //正常從idle_list中選出一個節點 ++_access_cnt; get_host_from_list(rsp, _idle_list); } //選擇一個idle節點 get_host_from_list(rsp, _idle_list); } } return lars::RET_SUCC; } ``` 從`idle_list`和`over_list`中的去取出適當的host階段返回給上層。 #### **遠程拉取host信息(load_balance不存在情況)** load_balance首先向dns_client的thread_queue發送`GetRouteRequest`請求。load_balance更新為`PULLING`狀態。 ##### pull發送請求過程 ```bash load_balance->pull() ----> dns client ----> dns server ``` > lars_loadbalance_agent/src/load_balance.cpp ```c //如果list中沒有host信息，需要從遠程的DNS Service發送GetRouteHost請求申請 int load_balance::pull() { //請求dns service請求 lars::GetRouteRequest route_req; route_req.set_modid(_modid); route_req.set_cmdid(_cmdid); //通過dns client的thread queue發送請求 dns_queue->send(route_req); //由于遠程會有一定延遲，所以應該給當前的load_balance模塊標記一個正在拉取的狀態 status = PULLING; return 0; } ``` > lars_loadbalance_agent/src/dns_client.cpp ``` #include "lars_reactor.h" #include "main_server.h" #include <pthread.h> //typedef void io_callback(event_loop *loop, int fd, void *args); //只要thread_queue有數據，loop就會觸發此回調函數來處理業務 void new_dns_request(event_loop *loop, int fd, void *args) { tcp_client *client = (tcp_client*)args; //1. 將請求數據從thread_queue中取出， std::queue<lars::GetRouteRequest> msgs; //2. 將數據放在queue隊列中 dns_queue->recv(msgs); //3. 遍歷隊列，通過client依次將每個msg發送給reporter service while (!msgs.empty()) { lars::GetRouteRequest req = msgs.front(); msgs.pop(); std::string requestString; req.SerializeToString(&requestString); //client 發送數據 client->send_message(requestString.c_str(), requestString.size(), lars::ID_GetRouteRequest); } } //... void *dns_client_thread(void* args) { printf("dns client thread start\n"); event_loop loop; //1 加載配置文件得到dns service ip + port std::string ip = config_file::instance()->GetString("dnsserver", "ip", ""); short port = config_file::instance()->GetNumber("dnsserver", "port", 0); //2 創建客戶端 tcp_client client(&loop, ip.c_str(), port, "dns client"); //3 將thread_queue消息回調事件，綁定到loop中 dns_queue->set_loop(&loop); dns_queue->set_callback(new_dns_request, &client); //4 設置當收到dns service回執的消息ID_GetRouteResponse處理函數 client.add_msg_router(lars::ID_GetRouteResponse, deal_recv_route); //啟動事件監聽 loop.event_process(); return NULL; } void start_dns_client() { //開辟一個線程 pthread_t tid; //啟動線程業務函數 int ret = pthread_create(&tid, NULL, dns_client_thread, NULL); if (ret == -1) { perror("pthread_create"); exit(1); } //設置分離模式 pthread_detach(tid); } ``` ##### 接收遠程host信息回執過程 ```bash dns service ----> dns client ----> route_lb::update_host() ----> load_balance::update() ``` > lars_loadbalance_agent/src/dns_client.cpp ```c /* * 處理遠程dns service回復的modid/cmdid對應的路由信息 * */ void deal_recv_route(const char *data, uint32_t len, int msgid, net_connection *net_conn, void *user_data) { lars::GetRouteResponse rsp; //解析遠程消息到proto結構體中 rsp.ParseFromArray(data, len); int modid = rsp.modid(); int cmdid = rsp.cmdid(); int index = (modid+cmdid)%3; // 將該modid/cmdid交給一個route_lb處理 將rsp中的hostinfo集合加入到對應的route_lb中 r_lb[index]->update_host(modid, cmdid, rsp); } ``` > lars_loadbalance_agent/src/route_lb.cpp ```c //根據Dns Service返回的結果更新自己的route_lb_map int route_lb::update_host(int modid, int cmdid, lars::GetRouteResponse &rsp) { //1. 得到key uint64_t key = ((uint64_t)modid << 32) + cmdid; pthread_mutex_lock(&_mutex); //2. 在_route_map中找到對應的key if (_route_lb_map.find(key) != _route_lb_map.end()) { load_balance *lb = _route_lb_map[key]; if (rsp.host_size() == 0) { //2.1 如果返回的結果 lb下已經沒有任何host信息，則刪除該key delete lb; _route_lb_map.erase(key); } else { //2.2 更新新host信息 lb->update(rsp); } } pthread_mutex_unlock(&_mutex); return 0; } ``` > lars_loadbalance_agent/src/load_balance.cpp ```c //根據dns service遠程返回的結果，更新_host_map void load_balance::update(lars::GetRouteResponse &rsp) { //確保dns service返回的結果有host信息 assert(rsp.host_size() != 0); std::set<uint64_t> remote_hosts; std::set<uint64_t> need_delete; //1. 插入新增的host信息 到_host_map中 for (int i = 0; i < rsp.host_size(); i++) { //1.1 得到rsp中的一個host const lars::HostInfo & h = rsp.host(i); //1.2 得到ip+port的key值 uint64_t key = ((uint64_t)h.ip() << 32) + h.port(); remote_hosts.insert(key); //1.3 如果自身的_host_map找不到當下的key，說明是新增 if (_host_map.find(key) == _host_map.end()) { //新增 host_info *hi = new host_info(h.ip(), h.port(), lb_config.init_succ_cnt); if (hi == NULL) { fprintf(stderr, "new host_info error!\n"); exit(1); } _host_map[key] = hi; //新增的host信息加入到 空閑列表中 _idle_list.push_back(hi); } } //2. 刪除減少的host信息 從_host_map中 //2.1 得到哪些節點需要刪除 for (host_map_it it = _host_map.begin(); it != _host_map.end(); it++) { if (remote_hosts.find(it->first) == remote_hosts.end()) { //該key在host_map中存在，而在遠端返回的結果集不存在，需要鎖定被刪除 need_delete.insert(it->first); } } //2.2 刪除 for (std::set<uint64_t>::iterator it = need_delete.begin(); it != need_delete.end(); it++) { uint64_t key = *it; host_info *hi = _host_map[key]; if (hi->overload == true) { //從過載列表中刪除 _overload_list.remove(hi); } else { //從空閑列表刪除 _idle_list.remove(hi); } delete hi; } } ``` #### load balance agent V0.3--API get_host請求功能單元測試 為了方便我們編譯Lars的全部模塊，我們在/Lars/下提供一個Makefile來編譯子目錄模塊 > /Lars/Makefile ```makefile SUBDIRS = lars_reactor lars_dns lars_reporter lars_loadbalance_agent .PHONY: all all: @list='$(SUBDIRS)'; for subdir in $$list; do \ echo "Clean in $$subdir";\ $(MAKE) -C $$subdir;\ done .PHONY: clean clean: @echo Making clean @list='$(SUBDIRS)'; for subdir in $$list; do \ echo "Clean in $$subdir";\ $(MAKE) -C $$subdir clean;\ done ``` 現在我們編譯，然后分別啟動`lars_dns`,`lars_reporter`,`lars_loadbalance_agent` > lars_dns ```bash ~/Lars/lars_dns$ ./bin/lars_dns msg_router init... create 0 thread create 1 thread create 2 thread create 3 thread create 4 thread add msg cb msgid = 1 lars dns service .... now route version is 1573034612 modID = 1, cmdID = 1, ip = 3232235953, port = 7777 modID = 1, cmdID = 2, ip = 3232235954, port = 7776 modID = 2, cmdID = 1, ip = 3232235955, port = 7778 modID = 2, cmdID = 2, ip = 3232235956, port = 7779 load data to tmep succ! size is 4 load data to tmep succ! size is 4 ... ``` > lars_reporter ```c ~/Lars/lars_reporter$ ./bin/lars_reporter msg_router init... create 0 thread create 1 thread create 2 thread create 3 thread create 4 thread add msg cb msgid = 3 ``` > lars_lb_agent ```bash ~/Lars/lars_loadbalance_agent$ ./bin/lars_lb_agent msg_router init... UDP server on 0.0.0.0:8888 is running... add msg cb msgid = 4 agent UDP server :port 8888 is started... msg_router init... msg_router init... UDP server on 0.0.0.0:8890 is running... add msg cb msgid = 4 agent UDP server :port 8890 is started... [report] client thread start... done! msg_router init... UDP server on 0.0.0.0:8889 is running... add msg cb msgid = 4 agent UDP server :port 8889 is started... dns client thread start do_connect EINPROGRESS connect 127.0.0.1:7779 succ! msg_router init... do_connect EINPROGRESS add msg cb msgid = 2 connect 127.0.0.1:7778 succ! ``` 啟動API example/的單元測試程序 > API:get_host example ```bash ~/Lars/api/cpp/example$ ./example 1 1 lars_client() connection agent udp server succ! connection agent udp server succ! connection agent udp server succ! host is 177.1.168.192:7777 ~lars_client() ``` --- ### 關于作者： 作者：`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)  >**原創聲明:未經作者允許請勿轉載, 如果轉載請注明出處**