C语言如何实现异步Socket

创作时间:

作者:

@小白创作中心

C语言如何实现异步Socket

引用

来源

https://docs.pingcode.com/baike/1531899

异步Socket编程是网络编程中的一个重要概念，它允许程序在等待I/O操作完成时继续执行其他任务，从而提高程序的并发性和响应性能。本文将详细介绍如何在C语言中实现异步Socket，重点讲解多路复用的方法。

一、非阻塞I/O

非阻塞I/O是实现异步Socket的基础。通过设置Socket为非阻塞模式，I/O操作将在数据尚未准备好时立即返回，而不是阻塞等待。

1. 设置非阻塞模式

使用fcntl函数设置Socket为非阻塞模式：

#include <fcntl.h>
#include <unistd.h>

int set_nonblocking(int sockfd) {
    int flags = fcntl(sockfd, F_GETFL, 0);
    if (flags == -1) {
        return -1;
    }
    return fcntl(sockfd, F_SETFL, flags | O_NONBLOCK);
}

2. 非阻塞读写

在非阻塞模式下，读写操作可能返回EAGAIN或EWOULDBLOCK，表示数据尚未准备好。需要处理这些错误并继续操作。

ssize_t nonblocking_read(int sockfd, void *buf, size_t len) {
    ssize_t n;
    while ((n = read(sockfd, buf, len)) == -1 && errno == EAGAIN) {
        // 数据尚未准备好，继续尝试读取
    }
    return n;
}

ssize_t nonblocking_write(int sockfd, const void *buf, size_t len) {
    ssize_t n;
    while ((n = write(sockfd, buf, len)) == -1 && errno == EAGAIN) {
        // 数据尚未准备好，继续尝试写入
    }
    return n;
}

二、使用多线程

通过多线程可以在每个线程中处理一个Socket连接，从而实现异步I/O。然而，多线程编程复杂且开销大，不适合大量并发连接的场景。

1. 创建线程

使用pthread库创建线程来处理每个Socket连接：

#include <pthread.h>

void *handle_connection(void *arg) {
    int sockfd = *(int *)arg;
    char buf[1024];
    ssize_t n;
    // 处理连接
    while ((n = read(sockfd, buf, sizeof(buf))) > 0) {
        // 处理数据
        write(sockfd, buf, n);
    }
    close(sockfd);
    return NULL;
}

void start_server(int listen_sock) {
    pthread_t tid;
    int conn_sock;
    while ((conn_sock = accept(listen_sock, NULL, NULL)) != -1) {
        pthread_create(&tid, NULL, handle_connection, &conn_sock);
        pthread_detach(tid);
    }
}

三、使用多路复用

多路复用是处理大量并发连接的高效方法。常用的多路复用机制有select、poll和epoll。下面将详细介绍如何使用select和epoll实现异步Socket。

1. 使用select

select是一个标准的多路复用函数，适用于小规模并发连接。

#include <sys/select.h>
#include <unistd.h>

void start_server_select(int listen_sock) {
    fd_set readfds;
    int maxfd, conn_sock;
    struct sockaddr_in client_addr;
    socklen_t client_len = sizeof(client_addr);
    char buf[1024];
    ssize_t n;
    // 初始化
    FD_ZERO(&readfds);
    FD_SET(listen_sock, &readfds);
    maxfd = listen_sock;
    while (1) {
        fd_set tempfds = readfds;
        if (select(maxfd + 1, &tempfds, NULL, NULL, NULL) == -1) {
            perror("select");
            break;
        }
        // 处理新连接
        if (FD_ISSET(listen_sock, &tempfds)) {
            conn_sock = accept(listen_sock, (struct sockaddr *)&client_addr, &client_len);
            if (conn_sock == -1) {
                perror("accept");
                continue;
            }
            set_nonblocking(conn_sock);
            FD_SET(conn_sock, &readfds);
            if (conn_sock > maxfd) {
                maxfd = conn_sock;
            }
        }
        // 处理现有连接
        for (int i = 0; i <= maxfd; ++i) {
            if (FD_ISSET(i, &tempfds)) {
                n = read(i, buf, sizeof(buf));
                if (n <= 0) {
                    // 连接关闭或出错
                    close(i);
                    FD_CLR(i, &readfds);
                } else {
                    // 处理数据
                    write(i, buf, n);
                }
            }
        }
    }
}

2. 使用epoll

epoll是Linux特有的多路复用机制，适用于大规模并发连接。

#include <sys/epoll.h>
#include <unistd.h>

void start_server_epoll(int listen_sock) {
    int epfd, conn_sock;
    struct epoll_event ev, events[1024];
    struct sockaddr_in client_addr;
    socklen_t client_len = sizeof(client_addr);
    char buf[1024];
    ssize_t n;
    epfd = epoll_create1(0);
    if (epfd == -1) {
        perror("epoll_create1");
        return;
    }
    // 注册监听Socket
    ev.events = EPOLLIN;
    ev.data.fd = listen_sock;
    if (epoll_ctl(epfd, EPOLL_CTL_ADD, listen_sock, &ev) == -1) {
        perror("epoll_ctl");
        close(epfd);
        return;
    }
    while (1) {
        int nfds = epoll_wait(epfd, events, 1024, -1);
        if (nfds == -1) {
            perror("epoll_wait");
            break;
        }
        for (int i = 0; i < nfds; ++i) {
            if (events[i].data.fd == listen_sock) {
                // 处理新连接
                conn_sock = accept(listen_sock, (struct sockaddr *)&client_addr, &client_len);
                if (conn_sock == -1) {
                    perror("accept");
                    continue;
                }
                set_nonblocking(conn_sock);
                ev.events = EPOLLIN | EPOLLET;
                ev.data.fd = conn_sock;
                if (epoll_ctl(epfd, EPOLL_CTL_ADD, conn_sock, &ev) == -1) {
                    perror("epoll_ctl");
                    close(conn_sock);
                    continue;
                }
            } else {
                // 处理现有连接
                n = read(events[i].data.fd, buf, sizeof(buf));
                if (n <= 0) {
                    // 连接关闭或出错
                    close(events[i].data.fd);
                } else {
                    // 处理数据
                    write(events[i].data.fd, buf, n);
                }
            }
        }
    }
    close(epfd);
}