• 异步请求池——池式组件

    前言

      

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

    1. //init
    2. struct async_context *dns_async_client_init(void);
    3.  
    4. //commit
    5. int dns_async_client_commit(struct async_context *ctx, const char *domain, async_result_cb cb);
    6.  
    7. //pthread callback
    8. static void *dns_async_client_proc(void *arg);
    9.  
    10. //destroy
    11. int dns_async_client_destroy(struct async_context *ctx);
    • init：初始化函数只做三件事
      1. 1.calloc()创建上下文结构体
      2. 2.epoll_create()创建一个epoll fd 
      3. 3.pthread_create()创建一个新线程。

    • commit：commit函数就是发送请求，它做五件事
      1. 1.socket 创建socket 
      2. 2.connect连接到第三方服务 
      3. 3.encode--->mysql/redis/dns 根据对应的协议将发送的数据封装好 
      4. 4.send将数据发送出去 
      5. 5.epoll_ctl(ctx->epfd, EPOLL_CTL_ADD, sockfd, &ev);把fd加入到epoll中

    • pthread callback：创建线程时需要给他传递一个回调函数，它做下面伪代码的事情
      1. while(1){
      2. 	nready=epoll_wait()
      3. 	for(){
      4. 		recv();
      5. 		parser();//解析协议
      6. 		epoll_ctl(epfd, EPOLL_CTL_DEL, sockfd, NULL);
      7. 	}
      8. }
    • destroy：init创建了什么就销毁什么
      1. 1.close(epfd);
      2. 2.pthread_cancel(thid);
      3. 3.free(ctx);

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

    1. 初始化请求池init

    • init：初始化函数只做两件事
    1. 1.calloc创建上下文结构体
    2. 2.epoll_create()创建一个epoll fd 
    3. 3.pthread_create()创建一个新线程。
    1. struct async_context {
    2.     int ep_fd;
    3.     pthread_t thread_id;
    4. };
    5. //TODO init
    6. //1.malloc ctx;
    7. //2.epoll_create
    8. //3.pthread_create
    9. struct async_context *dns_async_client_init(void) {
    10.     int epfd = epoll_create(1); //
    11.     if (epfd < 0) return NULL;
    12.     struct async_context *ctx = calloc(1, sizeof(struct async_context));
    13.     if (ctx == NULL) {
    14.         close(epfd);
    15.         return NULL;
    16.     }
    17.     ctx->ep_fd = epfd;
    18.     int ret = pthread_create(&ctx->thread_id, NULL, dns_async_client_proc, ctx);
    19.     if (ret) {
    20.         perror("pthread_create");
    21.         return NULL;
    22.     }
    23.     usleep(1); //child go first
    24.     return ctx;
    25. }

    2. 建立连接提交请求commit

    • commit：commit函数就是发送请求，它做五件事
    1. 1.socket 创建socket 
    2. 2.connect连接到第三方服务 
    3. 3.encode--->mysql/redis/dns 根据对应的协议将发送的数据封装好 
    4. 4.send将数据发送出去 
    5. 5.epoll_ctl(ctx->epfd, EPOLL_CTL_ADD, sockfd, &ev);把fd加入到epoll中

     注意这里有一个async_result_cb回调函数，它是负责对fd接收到第三方服务返回的数据之后的回调函数。

    1. //TODO commit
    2. //1.socket
    3. //2.connect
    4. //3.encode ---> redis/mysql/dns
    5. //4.send
    6. //5.epoll_ctl(ctx->ep_fd, EPOLL_CTL_ADD, sockfd, &ev);
    7. int dns_async_client_commit(struct async_context *ctx, const char *domain, async_result_cb cb) {
    8.     //socket
    9.     int sockfd = socket(AF_INET, SOCK_DGRAM, 0);
    10.     if (sockfd < 0) {
    11.         perror("create socket failed\n");
    12.         exit(-1);
    13.     }
    14.  
    15.     printf("url:%s\n", domain);
    16.  
    17.     set_block(sockfd, 0); //nonblock
    18.  
    19.     struct sockaddr_in dest;
    20.     bzero(&dest, sizeof(dest));
    21.     dest.sin_family = AF_INET;
    22.     dest.sin_port = htons(53);
    23.     dest.sin_addr.s_addr = inet_addr(DNS_SVR);
    24. 	//connect
    25.     connect(sockfd, (struct sockaddr *) &dest, sizeof(dest));
    26.  
    27. 	//encode
    28.     struct dns_header header = {0};
    29.     dns_create_header(&header);
    30.  
    31.     struct dns_question question = {0};
    32.     dns_create_question(&question, domain);
    33.     
    34.     char request[1024] = {0};
    35.     int req_len = dns_build_request(&header, &question, request);
    36.  
    37.     //send
    38.     sendto(sockfd, request, req_len, 0, (struct sockaddr *) &dest, sizeof(struct sockaddr));
    39.  
    40.     struct ep_arg *eparg = (struct ep_arg *) calloc(1, sizeof(struct ep_arg));
    41.     if (eparg == NULL) return -1;
    42.     eparg->sockfd = sockfd;
    43.     eparg->cb = cb;
    44.  
    45.     struct epoll_event ev;
    46.     ev.data.ptr = eparg;
    47.     ev.events = EPOLLIN;
    48.     //epoll_ctl
    49.     int ret = epoll_ctl(ctx->ep_fd, EPOLL_CTL_ADD, sockfd, &ev);
    50.     return ret;
    51. }

    3. epoll线程的回调函数callback

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

    1. //TODO pthread callback
    2. /*
    3.  while(1){
    4.     epoll_wait();
    5.     recv;
    6.     parser();
    7.     data callback();
    8.     epoll_ctl(ep_fd, EPOLL_CTL_DEL, sockfd, NULL);
    9.     free(date);
    10.  }
    11.  */
    12. static void *dns_async_client_proc(void *arg) {
    13.     struct async_context *ctx = (struct async_context *) arg;
    14.     int epfd = ctx->ep_fd;
    15.     while (1) {
    16.         struct epoll_event events[ASYNC_CLIENT_NUM] = {0};
    17.         int nready = epoll_wait(epfd, events, ASYNC_CLIENT_NUM, -1);
    18.         if (nready <= 0) {
    19.             continue;
    20.         }
    21.         printf("nready:%d\n", nready);
    22.         int i = 0;
    23.         for (i = 0; i < nready; i++) {
    24.             struct ep_arg *data = (struct ep_arg *) events[i].data.ptr;
    25.             int sockfd = data->sockfd;
    26.             char buffer[1024] = {0};
    27.             struct sockaddr_in addr;
    28.             size_t addr_len = sizeof(struct sockaddr_in);
    29.             //recv
    30.             recvfrom(sockfd, buffer, sizeof(buffer), 0, (struct sockaddr *) &addr, (socklen_t *) &addr_len);
    31. 			//parse
    32.             struct dns_item *domain_list = NULL;
    33.             int count = dns_parse_response(buffer, &domain_list);
    34.             //call cb
    35.             data->cb(domain_list, count);
    36.             //del
    37.             epoll_ctl(epfd, EPOLL_CTL_DEL, sockfd, NULL);
    38.             close(sockfd);
    39.             //free
    40.             dns_async_client_free_domains(domain_list, count);
    41.             free(data);
    42.         }
    43.     }
    44. }

    4. 销毁请求池destroy

    • destroy：init创建了什么就销毁什么
    1. 1.close(epfd);
    2. 2.pthread_cancel(thid);
    3. 3.free(ctx);
    1. //TODO destroy
    2. //1.close(ep_fd)
    3. //2.pthread_cancel(ctx->thread_id);
    4. //3.free(ctx);
    5. int dns_async_client_destroy(struct async_context *ctx) {
    6.     close(ctx->ep_fd);
    7.     pthread_cancel(ctx->thread_id);
    8.     free(ctx);
    9.     return 0;
    10. }

    Demo完整代码

    DNS同步请求代码

    1. #include 
    2. #include 
    3. #include 
    4. #include 
    5. #include 
    6. #include 
    7. #include 
    8. #include 
    9. #include 
    10. #include 
    11.  
    12. #define DNS_SVR                "114.114.114.114"
    13. #define DNS_HOST            0x01
    14. #define DNS_CNAME            0x05
    15.  
    16. // DNS报文头部
    17. struct dns_header {
    18.     unsigned short id;
    19.     unsigned short flags;
    20.     unsigned short qdcount;
    21.     unsigned short ancount;
    22.     unsigned short nscount;
    23.     unsigned short arcount;
    24. };
    25. // DNS报文正文
    26. struct dns_question {
    27.     int length;
    28.     unsigned short qtype;
    29.     unsigned short qclass;
    30.     char *qname;
    31. };
    32. // DNS服务器返回的ip信息
    33. struct dns_item {
    34.     char *domain;
    35.     char *ip;
    36. };
    37.  
    38. // header填充与函数实现
    39. int dns_create_header(struct dns_header *header) {
    40.     if (header == NULL) return -1;
    41.     memset(header, 0, sizeof(struct dns_header));
    42.     srandom(time(NULL));
    43.     header->id = random();
    44.     header->flags |= htons(0x0100);
    45.     header->qdcount = htons(1);
    46.     return 0;
    47. }
    48.  
    49. // question填充与函数实现
    50. int dns_create_question(struct dns_question *question, const char *hostname) {
    51.     if (question == NULL) return -1;
    52.     memset(question, 0, sizeof(struct dns_question));
    53.  
    54.     question->qname = (char *) malloc(strlen(hostname) + 2);
    55.     if (question->qname == NULL) return -2;
    56.  
    57.     question->length = strlen(hostname) + 2;
    58.     question->qtype = htons(1);
    59.     question->qclass = htons(1);
    60.  
    61.     const char delim[2] = ".";
    62.  
    63.     char *hostname_dup = strdup(hostname);
    64.     char *token = strtok(hostname_dup, delim);
    65.     char *qname_p = question->qname;
    66.  
    67.     while (token != NULL) {
    68.         size_t len = strlen(token);
    69.         *qname_p = len;
    70.         qname_p++;
    71.         strncpy(qname_p, token, len + 1);
    72.         qname_p += len;
    73.         token = strtok(NULL, delim);
    74.     }
    75.  
    76.     free(hostname_dup);
    77.     return 0;
    78. }
    79.  
    80. // 对头部和问题区做一个打包
    81. int dns_build_request(struct dns_header *header, struct dns_question *question, char *request) {
    82.     int header_s = sizeof(struct dns_header);
    83.     int question_s = question->length + sizeof(question->qtype) + sizeof(question->qclass);
    84.     int length = question_s + header_s;
    85.  
    86.     int offset = 0;
    87.     memcpy(request + offset, header, sizeof(struct dns_header));
    88.     offset += sizeof(struct dns_header);
    89.  
    90.     memcpy(request + offset, question->qname, question->length);
    91.     offset += question->length;
    92.  
    93.     memcpy(request + offset, &question->qtype, sizeof(question->qtype));
    94.     offset += sizeof(question->qtype);
    95.  
    96.     memcpy(request + offset, &question->qclass, sizeof(question->qclass));
    97.     return length;
    98. }
    99.  
    100. // 解析服务器发过来的数据
    101. static int is_pointer(int in) {
    102.     return ((in & 0xC0) == 0xC0);
    103. }
    104.  
    105.  
    106. static void dns_parse_name(unsigned char *chunk, unsigned char *ptr, char *out, int *len) {
    107.  
    108.     int flag = 0, n = 0, alen = 0;
    109.     char *pos = out + (*len);
    110.     while (1) {
    111.         flag = (int) ptr[0];
    112.         if (flag == 0) break;
    113.         if (is_pointer(flag)) {
    114.             n = (int) ptr[1];
    115.             ptr = chunk + n;
    116.             dns_parse_name(chunk, ptr, out, len);
    117.             break;
    118.         }
    119.         else {
    120.             ptr++;
    121.             memcpy(pos, ptr, flag);
    122.             pos += flag;
    123.             ptr += flag;
    124.             *len += flag;
    125.             if ((int) ptr[0] != 0) {
    126.                 memcpy(pos, ".", 1);
    127.                 pos += 1;
    128.                 (*len) += 1;
    129.             }
    130.         }
    131.     }
    132. }
    133.  
    134. //解析响应信息	buffer为response返回的信息
    135. static int dns_parse_response(char *buffer, struct dns_item **domains) {
    136.     int i = 0;
    137.     unsigned char *ptr = buffer;
    138.  
    139.     ptr += 4;
    140.     int querys = ntohs(*(unsigned short *) ptr);
    141.  
    142.     ptr += 2;
    143.     int answers = ntohs(*(unsigned short *) ptr);
    144.  
    145.     ptr += 6;
    146.     for (i = 0; i < querys; i++) {
    147.         while (1) {
    148.             int flag = (int) ptr[0];
    149.             ptr += (flag + 1);
    150.  
    151.             if (flag == 0) break;
    152.         }
    153.         ptr += 4;
    154.     }
    155.  
    156.     char cname[128], aname[128], ip[20], netip[4];
    157.     int len, type, ttl, datalen;
    158.  
    159.     int cnt = 0;
    160.     struct dns_item *list = (struct dns_item *) calloc(answers, sizeof(struct dns_item));
    161.     if (list == NULL) {
    162.         return -1;
    163.     }
    164.  
    165.     for (i = 0; i < answers; i++) {
    166.  
    167.         bzero(aname, sizeof(aname));
    168.         len = 0;
    169.  
    170.         dns_parse_name(buffer, ptr, aname, &len);
    171.         ptr += 2;
    172.  
    173.         type = htons(*(unsigned short *) ptr);
    174.         ptr += 4;
    175.  
    176.         ttl = htons(*(unsigned short *) ptr);
    177.         ptr += 4;
    178.  
    179.         datalen = ntohs(*(unsigned short *) ptr);
    180.         ptr += 2;
    181.  
    182.         if (type == DNS_CNAME) {
    183.  
    184.             bzero(cname, sizeof(cname));
    185.             len = 0;
    186.             dns_parse_name(buffer, ptr, cname, &len);
    187.             ptr += datalen;
    188.  
    189.         }
    190.         else if (type == DNS_HOST) {
    191.  
    192.             bzero(ip, sizeof(ip));
    193.  
    194.             if (datalen == 4) {
    195.                 memcpy(netip, ptr, datalen);
    196.                 inet_ntop(AF_INET, netip, ip, sizeof(struct sockaddr));
    197.  
    198.                 printf("%s has address %s\n", aname, ip);
    199.                 printf("\tTime to live: %d minutes , %d seconds\n", ttl / 60, ttl % 60);
    200.  
    201.                 list[cnt].domain = (char *) calloc(strlen(aname) + 1, 1);
    202.                 memcpy(list[cnt].domain, aname, strlen(aname));
    203.  
    204.                 list[cnt].ip = (char *) calloc(strlen(ip) + 1, 1);
    205.                 memcpy(list[cnt].ip, ip, strlen(ip));
    206.  
    207.                 cnt++;
    208.             }
    209.  
    210.             ptr += datalen;
    211.         }
    212.     }
    213.     *domains = list;
    214.     ptr += 2;
    215.     return cnt;
    216. }
    217.  
    218.  
    219. int dns_client_commit(const char *domain) {
    220.     int sockfd = socket(AF_INET, SOCK_DGRAM, 0);
    221.     if (sockfd < 0) {
    222.         perror("create socket failed\n");
    223.         exit(-1);
    224.     }
    225.  
    226.     printf("url:%s\n", domain);
    227.  
    228.     struct sockaddr_in dest;
    229.     bzero(&dest, sizeof(dest));
    230.     dest.sin_family = AF_INET;
    231.     dest.sin_port = htons(53);
    232.     dest.sin_addr.s_addr = inet_addr(DNS_SVR);
    233.  
    234.     int ret = connect(sockfd, (struct sockaddr *) &dest, sizeof(dest));
    235.     printf("connect :%d\n", ret);
    236.  
    237.     struct dns_header header = {0};
    238.     dns_create_header(&header);
    239.  
    240.     struct dns_question question = {0};
    241.     dns_create_question(&question, domain);
    242.  
    243.     char request[1024] = {0};
    244.     int req_len = dns_build_request(&header, &question, request);
    245.     int slen = sendto(sockfd, request, req_len, 0, (struct sockaddr *) &dest, sizeof(struct sockaddr));
    246.  
    247.     char buffer[1024] = {0};
    248.     struct sockaddr_in addr;
    249.     size_t addr_len = sizeof(struct sockaddr_in);
    250.  
    251.     int n = recvfrom(sockfd, buffer, sizeof(buffer), 0, (struct sockaddr *) &addr, (socklen_t *) &addr_len);
    252.  
    253.     printf("recvfrom n : %d\n", n);
    254.     struct dns_item *domains = NULL;
    255.     dns_parse_response(buffer, &domains);
    256.  
    257.     return 0;
    258. }
    259.  
    260. char *domain[] = {
    261. //	"www.ntytcp.com",
    262.         "bojing.wang",
    263.         "www.baidu.com",
    264.         "tieba.baidu.com",
    265.         "news.baidu.com",
    266.         "zhidao.baidu.com",
    267.         "music.baidu.com",
    268.         "image.baidu.com",
    269.         "v.baidu.com",
    270.         "map.baidu.com",
    271.         "baijiahao.baidu.com",
    272.         "xueshu.baidu.com",
    273.         "cloud.baidu.com",
    274.         "www.163.com",
    275.         "open.163.com",
    276.         "auto.163.com",
    277.         "gov.163.com",
    278.         "money.163.com",
    279.         "sports.163.com",
    280.         "tech.163.com",
    281.         "edu.163.com",
    282.         "www.taobao.com",
    283.         "q.taobao.com",
    284.         "sf.taobao.com",
    285.         "yun.taobao.com",
    286.         "baoxian.taobao.com",
    287.         "www.tmall.com",
    288.         "suning.tmall.com",
    289.         "www.tencent.com",
    290.         "www.qq.com",
    291.         "www.aliyun.com",
    292.         "www.ctrip.com",
    293.         "hotels.ctrip.com",
    294.         "hotels.ctrip.com",
    295.         "vacations.ctrip.com",
    296.         "flights.ctrip.com",
    297.         "trains.ctrip.com",
    298.         "bus.ctrip.com",
    299.         "car.ctrip.com",
    300.         "piao.ctrip.com",
    301.         "tuan.ctrip.com",
    302.         "you.ctrip.com",
    303.         "g.ctrip.com",
    304.         "lipin.ctrip.com",
    305.         "ct.ctrip.com"
    306. };
    307.  
    308.  
    309. int main(int argc, char *argv[]) {
    310.     int begin, end;
    311.     begin = clock();    //计时开始
    312.  
    313.     int i;
    314.     for (i = 0; i < sizeof(domain) / sizeof(domain[0]); i++) {
    315.         dns_client_commit(domain[i]);
    316.     }
    317.  
    318.     end = clock();    //计时结束
    319.  
    320.     getchar();
    321.  
    322.     printf("\n\nRunning Time：%lfs\n", (double)(end-begin)/CLOCKS_PER_SEC);
    323. }

    DNS异步请求代码

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

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

     

    如果想用同步的编程方式去实现异步的性能，那么就需要用到协程的思想来进行改变这个异步请求池

    也就是我们异步请求池里的操作是，一个线程处理结果，另一个线程检测IO事件是否就绪，

    那么用协程的思想去操作的时候，那么就知道了，我们同步和异步的区别就是，发送请求

    等待结果，并处理结果，这里会等待，等待的原因是因为IO事件没有就绪，那么就可以用

    跳转，如果事件没有就绪，直接跳转，重新发送，并检测IO事件，往返多次，就达到了我

    们的异步的性能，主要是优化了等待资源就绪的时间！！！！ 

  • 相关阅读:
    详解卡尔曼滤波原理
    RabbitMQ实践——搭建多人聊天服务
    超声功率放大器在MEMS超声测试中的应用
    rxjs Observable of 操作符的单步调试分析
    Mysql之聚合函数
    office365 outlook邮件无法删除
    Java中如何通过路径表达式找值：XPath和JsonPath以及SpEL详解及对比
    kafka（六）：java API消费数据
    从模型到部署，教你如何用Python构建机器学习API服务
    数据分发服务 (DDS) 内置主题
  • 原文地址：https://blog.csdn.net/txh1873749380/article/details/134096146