异步请求池——池式组件

前言

本文详细介绍异步请求池的实现过程，并使用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 firstreturn 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) {//socketint 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); //nonblockstruct 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);//connectconnect(sockfd, (struct sockaddr *) &dest, sizeof(dest));//encodestruct 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);//sendsendto(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_ctlint 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);//recvrecvfrom(sockfd, buffer, sizeof(buffer), 0, (struct sockaddr *) &addr, (socklen_t *) &addr_len);//parsestruct dns_item *domain_list = NULL;int count = dns_parse_response(buffer, &domain_list);//call cbdata->cb(domain_list, count);//delepoll_ctl(epfd, EPOLL_CTL_DEL, sockfd, NULL);close(sockfd);//freedns_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);
}