异步请求池的实现-阿里云开发者社区

前言

本文详细介绍异步请求池的实现过程，并使用DNS服务来测试异步请求池的性能。

两个必须牢记心中的概念：

同步：检测IO 与读写IO 在同一个流程里
异步：检测IO 与读写IO 不在同一个流程里

本专栏知识点是通过零声教育的线上课学习，进行梳理总结写下文章，对c/c++linux课程感兴趣的读者，可以点击链接 C/C++后台高级服务器课程介绍详细查看课程的服务。

同步请求与异步请求的处理流程

同步请求的处理流程

我们知道，同步：检测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、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异步请求池实现

http和redis的协议太复杂，不适合做demo，下面以DNS为例。关于DNS协议，直接看我儿子的这篇 LinuxC/C++ UDP编程实现DNS客户端，本文只实现异步请求池，对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条域名，可以看到差距还是非常明显的。

异步请求池的实现

前言

同步请求与异步请求的处理流程

同步请求的处理流程

异步请求的处理流程

同步请求与异步请求的差异

设计异步请求池

初步构思

King式四元组 init、commit、callback、destroy

应用协议DNS异步请求池实现

1. 初始化请求池init

2. 建立连接提交请求commit

3. epoll线程的回调函数callback

4. 销毁请求池destroy

Demo完整代码

DNS同步请求代码

DNS异步请求代码

DNS同步与异步的性能测试对比

热门文章

最新文章

相关电子书

探索云世界

热门

云计算

大数据

云原生

人工智能

数据库

开发与运维

活动广场

任务中心

训练营

直播

乘风者计划

下载

镜像站

技术资料

异步请求池的实现

前言

同步请求 与 异步请求 的处理流程

同步请求的处理流程

异步请求的处理流程

同步请求 与 异步请求的 差异

设计异步请求池

初步构思

King式四元组 init、commit、callback、destroy

应用协议DNS异步请求池实现

1. 初始化请求池init

2. 建立连接提交请求commit

3. epoll线程的回调函数callback

4. 销毁请求池destroy

Demo完整代码

DNS同步请求代码

DNS异步请求代码

DNS同步与异步的性能测试对比

热门文章

最新文章

相关电子书

同步请求与异步请求的处理流程

同步请求与异步请求的差异