概述
异步请求的方式其实适用于http、mysql、redis、dns等请求
同步请求与异步请求的处理流程
同步请求的处理流程
我们知道，同步：检测IO 与 读写IO 在同一个流程里,那么就意味着，发送消息之后，需要等待返回结果，在结果没有返回之前都在阻塞等待，图中我们发了3次请求，很明显的看出，这三个请求是串行的。都串行了，怎么搞并发呀！下面来看看异步请求。
异步请求的处理流程
异步：检测IO 与 读写IO 不在同一个流程里，不在一个流程里，那就好办了。发请求的时候我就一直发，不等待结果返回；读IO在另一个线程由epoll进行检测。
同步请求与异步请求的差异
很明显，同步需要阻塞等待一个请求的完成，异步不需要。同步是一个线程处理所有请求，异步是两个线程。那么如果请1000个请求需要完成呢？必然不可能采用同步阻塞等待的方案，第1000个请求不得等睡着啊。
设计异步请求池
初步构思
在第三方服务中，连接sockfd都是同步的，也就是说，在同步的流程上，一个连接，可以发多个请求，只不过需要阻塞等待上一条请求返回结果而已。
而异步呢，异步的其中一个线程的任务必然是一直发送请求，那么必然是非阻塞的，那么我们设计一个请求对应一个fd。“池的概念就出来了”，在send之后，我们将fd加入到epoll中，而epoll_wait在哪呢？在另一个线程中，epoll所在的线程就一直检测epoll中是否有fd可读。
说的通俗一点，一个请求send之后，将对应的fd加入到epoll里面，另一个线程在一直epoll_wait()读数据。
King式四元组
//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异步请求池实现
http和redis的协议太复杂，不适合做demo，下面以DNS为例。关于DNS协议，本文只实现异步请求池，对DNS协议不做解释。
1、初始化请求池init
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
//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
//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
//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; }
DNS同步与异步的性能测试对比
这里就测试了44条域名，可以看到差距还是非常明显的。
mysql的断线重连问题解决：
断线重连
在cli环境下，PHP程序需要长时间运行，客户端与MySQL服务器之间的TCP连接是不稳定的。
MySQL-Server会在一定时间内自动切断连接
PHP程序遇到空闲期时长时间没有MySQL查询，MySQL-Server也会切断连接回收资源
其他情况，在MySQL服务器中执行kill process杀掉某个连接，MySQL服务器重启
这时PHP程序中的MySQL连接就失效了。如果仍然执行mysql_query，就会报一个“MySQL server has gone away”的错误。程序处理不到就直接遇到致命错误并退出了。所以PHP程序中需要断线重连。
有很多人提出了mysql_ping的方案，每次mysql_query进行连接检测或者定时连接检测。这个方案不是最好的。原因是
mysql_ping需要主动侦测连接，带来了额外的消耗
定时执行mysql_ping不能解决问题，如刚刚执行过mysql_ping检测之后，连接就关闭了
最佳的方案是进行断线重连 。它的原理是：
(Swoole\Coroutine\Mysql和PDO等其他客户端同理, 此处简单举例)
mysql_query执行后检测返回值
如果mysql_query返回失败，检测错误码发现为2006/2013（这2个错误表示连接失败），再执行一次mysql_connect
执行mysql_connect后，重新执行mysql_query，这时必然会成功，因为已经重新建立了连接
如果mysql_query返回成功，那么连接是有效的，这是一次正常的调用
需要注意：
asyc_database_operator在真正的实战项目中需要加上连接池，这样才能更好的提升性能,因为mysql的连接数量是有限制的，需要有个连接池来控制着连接的数量，后面有空实现一下
