• 异步请求池的实现

    异步请求池的实现

    前言

      

      本文详细介绍异步请求池的实现过程，并使用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);

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • init：初始化函数只做三件事
    1.calloc()创建上下文结构体
2.epoll_create()创建一个epoll fd 
3.pthread_create()创建一个新线程。

    • 1
    • 2
    • 3
    • 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中

    • 1
    • 2
    • 3
    • 4
    • 5
    • pthread callback：创建线程时需要给他传递一个回调函数，它做下面伪代码的事情
    while(1){
	nready=epoll_wait()
	for(){
		recv();
		parser();//解析协议
		epoll_ctl(epfd, EPOLL_CTL_DEL, sockfd, NULL);
	}
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • destroy：init创建了什么就销毁什么
    1.close(epfd);
2.pthread_cancel(thid);
3.free(ctx);

    • 1
    • 2
    • 3

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

      http和redis的协议太复杂，不适合做demo，下面以DNS为例。关于DNS协议，直接看我儿子的这篇 LinuxC/C++ UDP编程实现DNS客户端 ，本文只实现异步请求池，对DNS协议不做解释。

    1. 初始化请求池init

    • init：初始化函数只做两件事
    1.calloc创建上下文结构体
2.epoll_create()创建一个epoll fd 
3.pthread_create()创建一个新线程。

    • 1
    • 2
    • 3
    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;
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25

    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中

    • 1
    • 2
    • 3
    • 4
    • 5

      注意这里有一个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;
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51

    3. epoll线程的回调函数callback

    • pthread callback：创建线程时需要给他传递一个回调函数，它做下面伪代码的事情
    while(1){
	nready=epoll_wait()
	for(){
		recv();
		parser();//解析协议
		epoll_ctl(epfd, EPOLL_CTL_DEL, sockfd, NULL);
	}
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    //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);
        }
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44

    4. 销毁请求池destroy

    • destroy：init创建了什么就销毁什么
    1.close(epfd);
2.pthread_cancel(thid);
3.free(ctx);

    • 1
    • 2
    • 3
    //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;
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10

    Demo完整代码

    DNS同步请求代码

    #include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 

#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);
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
    • 80
    • 81
    • 82
    • 83
    • 84
    • 85
    • 86
    • 87
    • 88
    • 89
    • 90
    • 91
    • 92
    • 93
    • 94
    • 95
    • 96
    • 97
    • 98
    • 99
    • 100
    • 101
    • 102
    • 103
    • 104
    • 105
    • 106
    • 107
    • 108
    • 109
    • 110
    • 111
    • 112
    • 113
    • 114
    • 115
    • 116
    • 117
    • 118
    • 119
    • 120
    • 121
    • 122
    • 123
    • 124
    • 125
    • 126
    • 127
    • 128
    • 129
    • 130
    • 131
    • 132
    • 133
    • 134
    • 135
    • 136
    • 137
    • 138
    • 139
    • 140
    • 141
    • 142
    • 143
    • 144
    • 145
    • 146
    • 147
    • 148
    • 149
    • 150
    • 151
    • 152
    • 153
    • 154
    • 155
    • 156
    • 157
    • 158
    • 159
    • 160
    • 161
    • 162
    • 163
    • 164
    • 165
    • 166
    • 167
    • 168
    • 169
    • 170
    • 171
    • 172
    • 173
    • 174
    • 175
    • 176
    • 177
    • 178
    • 179
    • 180
    • 181
    • 182
    • 183
    • 184
    • 185
    • 186
    • 187
    • 188
    • 189
    • 190
    • 191
    • 192
    • 193
    • 194
    • 195
    • 196
    • 197
    • 198
    • 199
    • 200
    • 201
    • 202
    • 203
    • 204
    • 205
    • 206
    • 207
    • 208
    • 209
    • 210
    • 211
    • 212
    • 213
    • 214
    • 215
    • 216
    • 217
    • 218
    • 219
    • 220
    • 221
    • 222
    • 223
    • 224
    • 225
    • 226
    • 227
    • 228
    • 229
    • 230
    • 231
    • 232
    • 233
    • 234
    • 235
    • 236
    • 237
    • 238
    • 239
    • 240
    • 241
    • 242
    • 243
    • 244
    • 245
    • 246
    • 247
    • 248
    • 249
    • 250
    • 251
    • 252
    • 253
    • 254
    • 255
    • 256
    • 257
    • 258
    • 259
    • 260
    • 261
    • 262
    • 263
    • 264
    • 265
    • 266
    • 267
    • 268
    • 269
    • 270
    • 271
    • 272
    • 273
    • 274
    • 275
    • 276
    • 277
    • 278
    • 279
    • 280
    • 281
    • 282
    • 283
    • 284
    • 285
    • 286
    • 287
    • 288
    • 289
    • 290
    • 291
    • 292
    • 293
    • 294
    • 295
    • 296
    • 297
    • 298
    • 299
    • 300
    • 301
    • 302
    • 303
    • 304
    • 305
    • 306
    • 307
    • 308
    • 309
    • 310
    • 311
    • 312
    • 313
    • 314
    • 315
    • 316
    • 317
    • 318
    • 319
    • 320
    • 321
    • 322
    • 323

    DNS异步请求代码

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

      这里就测试了44条域名，可以看到差距还是非常明显的。

    在这里插入图片描述

  • 相关阅读:
    06.多态
    fastAdmin表格列表的功能
    综述----知识蒸馏
    VS Code调试使用标准输入功能的go程序的问题
    nginx加权轮询,upstream,Keepalive,负载均衡实现案例
    2022年下半年（软考高级）信息系统项目管理师报名条件
    Python列表排序
    导航【mysql高级】【java提高】
    DTU配置工具-F2x16工具
    探讨javascript的程序性能
  • 原文地址：https://blog.csdn.net/qq_42956653/article/details/126046913