高效通信的新范式：探索异步请求在网络设计中的创新思路
【摘要】 本文将探讨一种新的网络设计范式，即使用异步请求来提高通信效率。传统的网络通信中，请求和响应是同步进行的，即发送请求后必须等待响应才能继续下一步操作，这样会导致通信效率低下。而通过引入异步请求的思想，可以在发送请求后不需要等待响应，而是可以立即进行其他操作，等到响应返回时再处理。这种设计思路可以大大提高通信效率，特别适用于网络延迟较高、带宽有限的情况。
一、前言
Linux网络设计是指在Linux操作系统环境下，构建和优化计算机网络的架构和方案。它对于现代社会中的各个领域都至关重要，包括互联网、云计算、物联网等。
重要性：
可靠性和稳定性：作为一种开源的操作系统，Linux以其强大的稳定性和可靠性而闻名。因此，在网络设计中采用Linux能够提供更高的可靠性和稳定性，确保数据传输和网络连接的稳固性。
可扩展性：Linux网络设计提供了灵活的模块化结构，使得网络可以根据需求进行扩展和定制。这使得在变化多端的网络环境中能够自由地添加新的功能和服务，适应不断增长的业务需求。
安全性：Linux以其出色的安全性而闻名，具备强大的防护机制和安全特性。在网络设计中采用Linux可以提供高级的安全措施，保护数据和资源免受威胁和攻击。
开放性和自由度：作为开源平台，Linux鼓励用户参与和贡献代码，享有更大的自由度。这意味着在网络设计中，可以根据实际需求进行源代码定制和修改，满足特定需求并与其他系统和设备无缝集成。
挑战：
复杂性：Linux网络设计涉及多个组件和协议，需要深入理解各种技术和架构原理。此复杂性对于初学者或非专业人士来说可能是一种挑战，需要耐心学习和实践。
网络规模和负载：随着网络规模的持续增长和复杂性的提高，网络设计可能面临更大的挑战。
异步请求作为提升通信效率的创新思路在Linux网络设计中具有重要意义。传统的阻塞模型在进行网络通信时，通常需要等待请求的返回结果才能继续执行后续操作，这会导致资源的闲置和效率的降低。而异步请求则通过将请求发送后立即返回，并通过回调函数或事件通知的方式处理响应结果，实现了非阻塞式的通信机制。
使用异步请求的优势包括：
- 提高系统的并发性：通过异步请求，可以在等待响应的同时处理其他任务，充分利用系统资源，提高系统的并发性能。
- 减少等待时间：由于异步请求不需要等待响应结果就能继续执行后续操作，能够减少等待时间，提高通信效率。
- 可扩展性：异步请求使得系统能够同时处理多个请求，有助于应对大量并发请求的情况，提高系统的可扩展性。
然而，异步请求也面临一些挑战：
- 处理复杂性：异步请求需要设计合适的回调函数或事件处理机制，以确保正确处理返回结果并保持代码的清晰可读性。
- 线程安全性：多个异步请求可能会同时访问共享资源，因此需要采取合适的线程同步机制，确保数据的一致性和安全性。
- 调试和排错：异步请求的复杂性会增加调试和排错的难度，因为请求和响应的处理不再是线性的。
二、同步和异步的概念
用于形容两者的关系。
同步： 所谓同步，就是发起一个请求时，在返回结果前，该调用不会返回。类似串行的概念。
异步： 异步的概念和同步相对，当发起一个请求时，该调用立刻返回，不等待结果，实际返回的结果由另外的线程 / 进程处理。类似并行的概念。
使用异步的典型例子：NTP时间同步服务器。
同步和异步是用来描述不同的操作方式或通信模式的概念。
同步（Synchronous）：在同步操作中，调用者发起一个请求后，需要等待该操作完成并返回结果，然后才能继续执行后续操作。这意味着调用者会阻塞等待操作的完成。同步操作通常按照顺序依次执行，每个操作必须等待前一个操作完成才能开始。
异步（Asynchronous）：在异步操作中，调用者发起一个请求后，不需要等待该操作的完成，可以继续执行后续操作。异步操作会在后台进行处理，并通过回调函数、事件通知或轮询等方式，在适当的时机将结果返回给调用者。异步操作不会阻塞调用者的执行流程，而是充分利用等待时间去执行其他任务。
对比：
- 同步操作强调顺序执行，调用者必须等待结果返回后才能继续执行。
- 异步操作强调非阻塞执行，调用者不需要等待结果返回，可以先执行后续操作。
选择使用同步还是异步操作取决于具体的应用场景和需求。同步操作简单直观，适用于简单任务或依赖前后顺序的操作。异步操作适合处理复杂、耗时较长或需要同时处理多个任务的情况，可以提高系统的并发性和响应能力。
三、异步的优缺点
优点： 异步比同步的性能要好。因为异步的发起请求和处理结果是分开的，不像同步需要等待有结果才返回调用，所以性能比同步高。
缺点： 异步不易理解，流程不清晰。
优点：
- 提高系统并发性能：异步操作可以在等待某个任务的同时执行其他任务，从而充分利用系统资源，提高系统的并发性能和吞吐量。
- 改善响应性能：由于异步操作不需要等待结果返回，调用者可以立即执行后续操作，使得系统能够更快地响应用户请求，提高用户体验。
- 资源利用率高：在等待IO操作完成时，异步操作可以释放CPU等资源，避免资源的闲置浪费，提高系统的资源利用率。
- 高度可扩展性：通过异步操作，系统能够同时处理大量的并发请求，适应高负载场景，并具有更好的扩展性。
缺点：
- 处理复杂性增加：异步操作通常需要设计回调函数或事件处理机制，代码逻辑较同步操作更加复杂，需要额外的开发和维护工作。
- 调试和排错困难：异步操作的执行流程相对复杂，当出现错误时，调试和排错会更加困难，需要细致地跟踪异步操作的过程。
- 可能引发并发问题：由于异步操作通常涉及共享资源的访问，如果对共享资源的并发访问控制不当，可能引发竞态条件、死锁等并发问题。
- 不适用于简单任务：对于一些简单的任务或者依赖前后顺序的操作，使用异步操作可能会引入额外的复杂性，反而降低代码的可读性和易用性。
四、异步实现逻辑
异步的实现主要分为四个部分。
4.1、初始化环境
实现一个初始化上下文环境的函数。主要是创建一个处理返回结果的线程。
示例：
int async_client_init(struct async_context *ctx)
{
if (ctx == NULL)
return -EINVAL;
ctx->epfd = epoll_create(1);// IO多路复用器
// 开启处理结果的线程
int ret = pthread_create(&ctx->thid, NULL, dns_async_client_callback, ctx);
if (ret)
{
perror("pthread_create error");
return -1;
}
return 0;
}
4.2、处理返回结果
实现一个线程回调函数，主要用于使用IO多路复用器监测是否有数据返回、接收数据和处理数据。
伪代码：
void * async_client_callback(void *arg)
{
struct async_context *ctx = (struct async_context *)arg;
while (1)
{
// 监测可读事件
//...
// 获取与socketfd关联的数据
//...
// 从读缓冲区中读取数据
//...
// 解析数据
//...
// 删除事件监听
//...
//释放内存
//...
}
}
4.3、发起请求
实现一个请求函数。主要是创建socket、建立连接、准备协议、发送协议、添加fd到epoll等IO多路复用器中。
示例：
int async_client_commit(struct async_context *ctx,const char *domain, async_result_cb callback)
{
// 创建 socket
int sockfd = socket(AF_INET, SOCK_DGRAM, 0);
if (sockfd < 0) {
perror("create socket failed\n");
exit(-1);
}
// 配置socket相关信息
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目标，探路
int ret = connect(sockfd, (struct sockaddr*)&dest, sizeof(dest));
// 准备协议
char request[1024] = { 0 };
//...
// 发送数据
int slen = sendto(sockfd, request, request_length, 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 = callback;
// 加入epoll中，监测结果返回
struct epoll_event ev;
ev.events = EPOLLIN;
ev.data.ptr = eparg;
return epoll_ctl(ctx->epfd, EPOLL_CTL_ADD, sockfd, &ev);
}
4.4、销毁
关闭相关的文件描述符，释放申请的内存块。
示例：
int async_client_destory(struct async_context *ctx)
{
if (ctx == NULL)
return -EINVAL;
pthread_cancel(ctx->thid);
close(ctx->epfd);
return 0;
}
五、异步请求实现示例
以DNS请求为例，遵循实现异步的四个过程。
示例代码：
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <errno.h>
#include <fcntl.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <sys/epoll.h>
#include <netdb.h>
#include <arpa/inet.h>
#include <pthread.h>
struct dns_item {
char *domain;
char *ip;
};
// 异步相关的结构体
#define ASYNC_EVENT_LENGTH 1024
typedef void(*async_result_cb)(struct dns_item *list, int count);
struct ep_arg {
int sockfd;
async_result_cb cb;
};
struct async_context {
int epfd;
pthread_t thid;
};
/*
*
* DNS 协议相关实现
*
*/
// ..............................
char *domain[] = {
"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"
};
/*
*
* 异步请求池的实现
*
*/
void dns_async_client_free_domains(struct dns_item *list, int count) {
int i = 0;
for (i = 0; i < count; i++) {
free(list[i].domain);
free(list[i].ip);
}
free(list);
}
void * dns_async_client_callback(void *arg)
{
struct async_context *ctx = (struct async_context *)arg;
while (1)
{
struct epoll_event events[ASYNC_EVENT_LENGTH] = { 0 };
// 监测可读事件
int nready = epoll_wait(ctx->epfd, events, ASYNC_EVENT_LENGTH, -1);
if (nready < 0)
{
if (errno == EINTR || errno == EAGAIN)
continue;
else
break;
}
else if (nready == 0)
continue;
printf("nready:%d\n", nready);
int i = 0;
for (i = 0; i < nready;i++)
{
// 获取与socketfd关联的数据
struct ep_arg *data = (struct ep_arg *)events[i].data.ptr;
int clientfd = data->sockfd;
if (events[i].events &EPOLLIN)
{
char buffer[1024] = { 0 };
struct sockaddr_in addr;
size_t addr_len = sizeof(struct sockaddr_in);
// 从读缓冲区中读取数据
int n = recvfrom(clientfd, buffer, sizeof(buffer), 0, (struct sockaddr*)&addr, (socklen_t*)&addr_len);
printf("recvfrom n : %d\n", n);
struct dns_item *domains = NULL;
// 解析数据
int count = dns_parse_response(buffer, &domains);
data->cb(domains, count);
// 删除事件监听
epoll_ctl(ctx->epfd, EPOLL_CTL_DEL, clientfd, NULL);
// 关闭fd
close(clientfd);
//释放内存
dns_async_client_free_domains(domains, count);
free(data);
}
}
}
}
int dns_async_client_init(struct async_context *ctx)
{
if (ctx == NULL)
return -EINVAL;
// 创建一个epoll对象
ctx->epfd = epoll_create(1);
// 开启线程，检测结果返回
int ret = pthread_create(&ctx->thid, NULL, dns_async_client_callback, ctx);
if (ret)
{
perror("pthread_create error");
close(ctx->epfd);
return -1;
}
usleep(1); //child go first
return 0;
}
int dns_async_client_destory(struct async_context *ctx)
{
if (ctx == NULL)
return -EINVAL;
// 关闭线程
pthread_cancel(ctx->thid);
// 关闭epoll对象
close(ctx->epfd);
return 0;
}
int dns_async_client_commit(struct async_context *ctx,const char *domain, async_result_cb callback)
{
// 创建 socket
int sockfd = socket(AF_INET, SOCK_DGRAM, 0);
if (sockfd < 0) {
perror("create socket failed\n");
exit(-1);
}
printf("url:%s\n", domain);
// 配置socket相关信息
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目标，探路
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));
// 传递数据
struct ep_arg *eparg = (struct ep_arg *)calloc(1, sizeof(struct ep_arg));
if (eparg == NULL)
return -1;
eparg->sockfd = sockfd;
eparg->cb = callback;
// 加入epoll中，监测结果返回
struct epoll_event ev;
ev.events = EPOLLIN;
ev.data.ptr = eparg;
return epoll_ctl(ctx->epfd, EPOLL_CTL_ADD, sockfd, &ev);
}
static void dns_async_client_result_callback(struct dns_item *list,int count)
{
int i = 0;
for (i = 0; i < count; i++)
{
printf("name: %s, ip: %s\n", list[i].domain, list[i].ip);
}
}
int main(int argc, char *argv[]) {
int count = sizeof(domain) / sizeof(domain[0]);
int i = 0;
// 申请一个保存上下文的内存块
struct async_context *ctx=calloc(1,sizeof(struct async_context));
if (ctx == NULL)
return -1;
// 初始化
int ret = dns_async_client_init(ctx);
if (ret < 0)
{
free(ctx);
return -2;
}
for (i = 0; i < count; i++) {
// 发起请求
dns_async_client_commit(ctx, domain[i],dns_async_client_result_callback);
}
getchar();
printf("exit\n");
// 销毁
dns_async_client_destory(ctx);
free(ctx);
}
DNS协议部分的实现，协议的准备，协议的处理等。
#define DNS_SVR "114.114.114.114"
#define DNS_HOST 0x01
#define DNS_CNAME 0x05
struct dns_header {
unsigned short id;
unsigned short flags;
unsigned short qdcount;
unsigned short ancount;
unsigned short nscount;
unsigned short arcount;
};
struct dns_question {
int length;
unsigned short qtype;
unsigned short qclass;
char *qname;
};
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;
}
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;
}
}
}
}
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;
}
六、思考
1、UDP为什么需要调用connect()函数？
connect()函数会发送一个数据包，UDP主要用它作探测 / 探路。
2、函数的返回值有哪些形式？
（1）将数据结构作为参数，传给函数（推荐）。这种方式，变量的内存在函数外申请，由函数调用的地方进行管理。例如：
int func(struct value *arg)
{
//......
return 0;
};
（2）内部创建数据结构变量，返回值返回。这种方式由函数申请内存，调用方管理释放；这种方式不注意的话容易造成内存泄漏。例如：
struct value * func(void)
{
struct value *arg=alloc(1,sizeof(struct value));
//......
return arg;
};
（3）将数据结构作为参数，传给函数，同时内部创建数据结构变量，返回值返回（不推荐）。这种方式会添加函数的复杂性。例如：
struct value * func(struct value *arg)
{
struct value *arg1=alloc(1,sizeof(struct value));
//......
return arg1;
};
总结
异步将请求和接收分离，一个线程负责发起请求，一个线程负责接收返回，减少了等待结果的过程，性能比较高；但不符合人类的理解思维，因此理解起来不那么容易。
实现示例代码中可以作为一个异步请求池的模板，只要将DNS协议部分换成其他的协议实现即可（如HTTP）。
【版权声明】本文为华为云社区用户原创内容，未经允许不得转载，如需转载请自行联系原作者进行授权。如果您发现本社区中有涉嫌抄袭的内容，欢迎发送邮件进行举报，并提供相关证据，一经查实，本社区将立刻删除涉嫌侵权内容，举报邮箱：
cloudbbs@huaweicloud.com
- 点赞
- 收藏
- 关注作者
评论（0）