前言
本文详细介绍异步请求池的实现过程，并使用DNS服务来测试异步请求池的性能。
两个必须牢记心中的概念：
- 同步：
检测IO 与 读写IO 在同一个流程里 - 异步：
检测IO 与 读写IO 不在同一个流程
同步请求 与 异步请求 的处理流程
同步请求的处理流程
我们知道，同步：检测IO 与 读写IO 在同一个流程里,那么就意味着，发送消息之后，需要等待返回结果，在结果没有返回之前都在阻塞等待，图中我们发了3次请求，很明显的看出，这三个请求是串行的。都串行了，怎么搞并发呀！下面来看看异步请求。
同步请求 与 异步请求的 差异
很明显，同步需要阻塞等待一个请求的完成，异步不需要。同步是一个线程处理所有请求，异步是两个线程。那么如果请1000个请求需要完成呢？必然不可能采用同步阻塞等待的方案，第1000个请求不得等睡着啊。
设计异步请求池
初步构思
在第三方服务中，连接sockfd都是同步的，也就是说，在同步的流程上，一个连接，可以发多个请求，只不过需要阻塞等待上一条请求返回结果而已。
而异步呢，异步的其中一个线程的任务必然是一直发送请求，那么必然是非阻塞的，那么我们设计一个请求对应一个fd。“池的概念就出来了”，在send之后，我们将fd加入到epoll中，而epoll_wait在哪呢？在另一个线程中，epoll所在的线程就一直检测epoll中是否有fd可读。
说的通俗一点，一个请求send之后，将对应的fd加入到epoll里面，另一个线程在一直epoll_wait()读数据。
四元组 init、commit、callback、destroy
//init struct async_context *dns_async_client_init(void); //commit int dns_async_client_commit(struct async_context *ctx, const char *domain, async_result_cb cb); //pthread callback static void *dns_async_client_proc(void *arg); //destroy int dns_async_client_destroy(struct async_context *ctx);
- init：初始化函数只做三件事
1.calloc()创建上下文结构体 2.epoll_create()创建一个epoll fd 3.pthread_create()创建一个新线程。
- commit：commit函数就是发送请求，它做五件事
1.socket 创建socket 2.connect连接到第三方服务 3.encode--->mysql/redis/dns 根据对应的协议将发送的数据封装好 4.send将数据发送出去 5.epoll_ctl(ctx->epfd, EPOLL_CTL_ADD, sockfd, &ev);把fd加入到epoll中
- pthread callback：创建线程时需要给他传递一个回调函数，它做下面伪代码的事情
while(1){ nready=epoll_wait() for(){ recv(); parser();//解析协议 epoll_ctl(epfd, EPOLL_CTL_DEL, sockfd, NULL); } }
- destroy：init创建了什么就销毁什么
1.close(epfd); 2.pthread_cancel(thid); 3.free(ctx);
- 应用协议DNS异步请求池实现
1. 初始化请求池init
- init：初始化函数只做两件事
1.calloc创建上下文结构体 2.epoll_create()创建一个epoll fd 3.pthread_create()创建一个新线程。
struct async_context { int ep_fd; pthread_t thread_id; }; //TODO init //1.malloc ctx; //2.epoll_create //3.pthread_create struct async_context *dns_async_client_init(void) { int epfd = epoll_create(1); // if (epfd < 0) return NULL; struct async_context *ctx = calloc(1, sizeof(struct async_context)); if (ctx == NULL) { close(epfd); return NULL; } ctx->ep_fd = epfd; int ret = pthread_create(&ctx->thread_id, NULL, dns_async_client_proc, ctx); if (ret) { perror("pthread_create"); return NULL; } usleep(1); //child go first return ctx; }
2. 建立连接提交请求commit
- commit：commit函数就是发送请求，它做五件事
1.socket 创建socket 2.connect连接到第三方服务 3.encode--->mysql/redis/dns 根据对应的协议将发送的数据封装好 4.send将数据发送出去 5.epoll_ctl(ctx->epfd, EPOLL_CTL_ADD, sockfd, &ev);把fd加入到epoll中
注意这里有一个async_result_cb回调函数，它是负责对fd接收到第三方服务返回的数据之后的回调函数。
//TODO commit //1.socket //2.connect //3.encode ---> redis/mysql/dns //4.send //5.epoll_ctl(ctx->ep_fd, EPOLL_CTL_ADD, sockfd, &ev); int dns_async_client_commit(struct async_context *ctx, const char *domain, async_result_cb cb) { //socket int sockfd = socket(AF_INET, SOCK_DGRAM, 0); if (sockfd < 0) { perror("create socket failed\n"); exit(-1); } printf("url:%s\n", domain); set_block(sockfd, 0); //nonblock struct sockaddr_in dest; bzero(&dest, sizeof(dest)); dest.sin_family = AF_INET; dest.sin_port = htons(53); dest.sin_addr.s_addr = inet_addr(DNS_SVR); //connect connect(sockfd, (struct sockaddr *) &dest, sizeof(dest)); //encode struct dns_header header = {0}; dns_create_header(&header); struct dns_question question = {0}; dns_create_question(&question, domain); char request[1024] = {0}; int req_len = dns_build_request(&header, &question, request); //send sendto(sockfd, request, req_len, 0, (struct sockaddr *) &dest, sizeof(struct sockaddr)); struct ep_arg *eparg = (struct ep_arg *) calloc(1, sizeof(struct ep_arg)); if (eparg == NULL) return -1; eparg->sockfd = sockfd; eparg->cb = cb; struct epoll_event ev; ev.data.ptr = eparg; ev.events = EPOLLIN; //epoll_ctl int ret = epoll_ctl(ctx->ep_fd, EPOLL_CTL_ADD, sockfd, &ev); return ret; }
3. epoll线程的回调函数callback
- pthread callback：创建线程时需要给他传递一个回调函数，它做下面伪代码的事情
while(1){ nready=epoll_wait() for(){ recv(); parser();//解析协议 epoll_ctl(epfd, EPOLL_CTL_DEL, sockfd, NULL); } }
//TODO pthread callback /* while(1){ epoll_wait(); recv; parser(); data callback(); epoll_ctl(ep_fd, EPOLL_CTL_DEL, sockfd, NULL); free(date); } */ static void *dns_async_client_proc(void *arg) { struct async_context *ctx = (struct async_context *) arg; int epfd = ctx->ep_fd; while (1) { struct epoll_event events[ASYNC_CLIENT_NUM] = {0}; int nready = epoll_wait(epfd, events, ASYNC_CLIENT_NUM, -1); if (nready <= 0) { continue; } printf("nready:%d\n", nready); int i = 0; for (i = 0; i < nready; i++) { struct ep_arg *data = (struct ep_arg *) events[i].data.ptr; int sockfd = data->sockfd; char buffer[1024] = {0}; struct sockaddr_in addr; size_t addr_len = sizeof(struct sockaddr_in); //recv recvfrom(sockfd, buffer, sizeof(buffer), 0, (struct sockaddr *) &addr, (socklen_t *) &addr_len); //parse struct dns_item *domain_list = NULL; int count = dns_parse_response(buffer, &domain_list); //call cb data->cb(domain_list, count); //del epoll_ctl(epfd, EPOLL_CTL_DEL, sockfd, NULL); close(sockfd); //free dns_async_client_free_domains(domain_list, count); free(data); } } }
4. 销毁请求池destroy
- destroy：init创建了什么就销毁什么
1.close(epfd); 2.pthread_cancel(thid); 3.free(ctx);
//TODO destroy //1.close(ep_fd) //2.pthread_cancel(ctx->thread_id); //3.free(ctx); int dns_async_client_destroy(struct async_context *ctx) { close(ctx->ep_fd); pthread_cancel(ctx->thread_id); free(ctx); return 0; }
Demo完整代码
DNS同步请求代码
#include <stdio.h> #include <stdlib.h> #include <string.h> #include <unistd.h> #include <sys/types.h> #include <sys/socket.h> #include <netinet/in.h> #include <sys/epoll.h> #include <arpa/inet.h> #include <pthread.h> #define DNS_SVR "114.114.114.114" #define DNS_HOST 0x01 #define DNS_CNAME 0x05 // DNS报文头部 struct dns_header { unsigned short id; unsigned short flags; unsigned short qdcount; unsigned short ancount; unsigned short nscount; unsigned short arcount; }; // DNS报文正文 struct dns_question { int length; unsigned short qtype; unsigned short qclass; char *qname; }; // DNS服务器返回的ip信息 struct dns_item { char *domain; char *ip; }; // header填充与函数实现 int dns_create_header(struct dns_header *header) { if (header == NULL) return -1; memset(header, 0, sizeof(struct dns_header)); srandom(time(NULL)); header->id = random(); header->flags |= htons(0x0100); header->qdcount = htons(1); return 0; } // question填充与函数实现 int dns_create_question(struct dns_question *question, const char *hostname) { if (question == NULL) return -1; memset(question, 0, sizeof(struct dns_question)); question->qname = (char *) malloc(strlen(hostname) + 2); if (question->qname == NULL) return -2; question->length = strlen(hostname) + 2; question->qtype = htons(1); question->qclass = htons(1); const char delim[2] = "."; char *hostname_dup = strdup(hostname); char *token = strtok(hostname_dup, delim); char *qname_p = question->qname; while (token != NULL) { size_t len = strlen(token); *qname_p = len; qname_p++; strncpy(qname_p, token, len + 1); qname_p += len; token = strtok(NULL, delim); } free(hostname_dup); return 0; } // 对头部和问题区做一个打包 int dns_build_request(struct dns_header *header, struct dns_question *question, char *request) { int header_s = sizeof(struct dns_header); int question_s = question->length + sizeof(question->qtype) + sizeof(question->qclass); int length = question_s + header_s; int offset = 0; memcpy(request + offset, header, sizeof(struct dns_header)); offset += sizeof(struct dns_header); memcpy(request + offset, question->qname, question->length); offset += question->length; memcpy(request + offset, &question->qtype, sizeof(question->qtype)); offset += sizeof(question->qtype); memcpy(request + offset, &question->qclass, sizeof(question->qclass)); return length; } // 解析服务器发过来的数据 static int is_pointer(int in) { return ((in & 0xC0) == 0xC0); } static void dns_parse_name(unsigned char *chunk, unsigned char *ptr, char *out, int *len) { int flag = 0, n = 0, alen = 0; char *pos = out + (*len); while (1) { flag = (int) ptr[0]; if (flag == 0) break; if (is_pointer(flag)) { n = (int) ptr[1]; ptr = chunk + n; dns_parse_name(chunk, ptr, out, len); break; } else { ptr++; memcpy(pos, ptr, flag); pos += flag; ptr += flag; *len += flag; if ((int) ptr[0] != 0) { memcpy(pos, ".", 1); pos += 1; (*len) += 1; } } } } //解析响应信息 buffer为response返回的信息 static int dns_parse_response(char *buffer, struct dns_item **domains) { int i = 0; unsigned char *ptr = buffer; ptr += 4; int querys = ntohs(*(unsigned short *) ptr); ptr += 2; int answers = ntohs(*(unsigned short *) ptr); ptr += 6; for (i = 0; i < querys; i++) { while (1) { int flag = (int) ptr[0]; ptr += (flag + 1); if (flag == 0) break; } ptr += 4; } char cname[128], aname[128], ip[20], netip[4]; int len, type, ttl, datalen; int cnt = 0; struct dns_item *list = (struct dns_item *) calloc(answers, sizeof(struct dns_item)); if (list == NULL) { return -1; } for (i = 0; i < answers; i++) { bzero(aname, sizeof(aname)); len = 0; dns_parse_name(buffer, ptr, aname, &len); ptr += 2; type = htons(*(unsigned short *) ptr); ptr += 4; ttl = htons(*(unsigned short *) ptr); ptr += 4; datalen = ntohs(*(unsigned short *) ptr); ptr += 2; if (type == DNS_CNAME) { bzero(cname, sizeof(cname)); len = 0; dns_parse_name(buffer, ptr, cname, &len); ptr += datalen; } else if (type == DNS_HOST) { bzero(ip, sizeof(ip)); if (datalen == 4) { memcpy(netip, ptr, datalen); inet_ntop(AF_INET, netip, ip, sizeof(struct sockaddr)); printf("%s has address %s\n", aname, ip); printf("\tTime to live: %d minutes , %d seconds\n", ttl / 60, ttl % 60); list[cnt].domain = (char *) calloc(strlen(aname) + 1, 1); memcpy(list[cnt].domain, aname, strlen(aname)); list[cnt].ip = (char *) calloc(strlen(ip) + 1, 1); memcpy(list[cnt].ip, ip, strlen(ip)); cnt++; } ptr += datalen; } } *domains = list; ptr += 2; return cnt; } int dns_client_commit(const char *domain) { int sockfd = socket(AF_INET, SOCK_DGRAM, 0); if (sockfd < 0) { perror("create socket failed\n"); exit(-1); } printf("url:%s\n", domain); struct sockaddr_in dest; bzero(&dest, sizeof(dest)); dest.sin_family = AF_INET; dest.sin_port = htons(53); dest.sin_addr.s_addr = inet_addr(DNS_SVR); int ret = connect(sockfd, (struct sockaddr *) &dest, sizeof(dest)); printf("connect :%d\n", ret); struct dns_header header = {0}; dns_create_header(&header); struct dns_question question = {0}; dns_create_question(&question, domain); char request[1024] = {0}; int req_len = dns_build_request(&header, &question, request); int slen = sendto(sockfd, request, req_len, 0, (struct sockaddr *) &dest, sizeof(struct sockaddr)); char buffer[1024] = {0}; struct sockaddr_in addr; size_t addr_len = sizeof(struct sockaddr_in); int n = recvfrom(sockfd, buffer, sizeof(buffer), 0, (struct sockaddr *) &addr, (socklen_t *) &addr_len); printf("recvfrom n : %d\n", n); struct dns_item *domains = NULL; dns_parse_response(buffer, &domains); return 0; } char *domain[] = { // "www.ntytcp.com", "bojing.wang", "www.baidu.com", "tieba.baidu.com", "news.baidu.com", "zhidao.baidu.com", "music.baidu.com", "image.baidu.com", "v.baidu.com", "map.baidu.com", "baijiahao.baidu.com", "xueshu.baidu.com", "cloud.baidu.com", "www.163.com", "open.163.com", "auto.163.com", "gov.163.com", "money.163.com", "sports.163.com", "tech.163.com", "edu.163.com", "www.taobao.com", "q.taobao.com", "sf.taobao.com", "yun.taobao.com", "baoxian.taobao.com", "www.tmall.com", "suning.tmall.com", "www.tencent.com", "www.qq.com", "www.aliyun.com", "www.ctrip.com", "hotels.ctrip.com", "hotels.ctrip.com", "vacations.ctrip.com", "flights.ctrip.com", "trains.ctrip.com", "bus.ctrip.com", "car.ctrip.com", "piao.ctrip.com", "tuan.ctrip.com", "you.ctrip.com", "g.ctrip.com", "lipin.ctrip.com", "ct.ctrip.com" }; int main(int argc, char *argv[]) { int begin, end; begin = clock(); //计时开始 int i; for (i = 0; i < sizeof(domain) / sizeof(domain[0]); i++) { dns_client_commit(domain[i]); } end = clock(); //计时结束 getchar(); printf("\n\nRunning Time：%lfs\n", (double)(end-begin)/CLOCKS_PER_SEC); }
DNS异步请求代码
DNS同步与异步的性能测试对比
这里就测试了44条域名，可以看到差距还是非常明显的。
如果想用同步的编程方式去实现异步的性能，那么就需要用到协程的思想来进行改变这个异步请求池
也就是我们异步请求池里的操作是，一个线程处理结果，另一个线程检测IO事件是否就绪，
那么用协程的思想去操作的时候，那么就知道了，我们同步和异步的区别就是，发送请求
等待结果，并处理结果，这里会等待，等待的原因是因为IO事件没有就绪，那么就可以用
跳转，如果事件没有就绪，直接跳转，重新发送，并检测IO事件，往返多次，就达到了我
们的异步的性能，主要是优化了等待资源就绪的时间！！！！
