如何实现事件驱动的POSIX线程？优化Linux C Socket监听逻辑

要解决线程A轮询占用资源的问题，核心是用Linux I/O多路复用机制替代轮询，让线程A仅在Socket有可读事件时被唤醒。同时结合POSIX线程的同步机制（互斥锁+条件变量）实现线程B的按需唤醒，且方案天然支持TCP和UDP Socket。

核心原理 #

I/O多路复用函数（如select()、poll()、epoll()）会让线程阻塞，直到监听的Socket描述符出现可读/可写等事件，完全避免空轮询的CPU消耗。其中select()是POSIX标准接口，兼容性最好，适合资源有限的平台。

线程同步方面，使用条件变量+互斥锁实现线程B的唤醒：线程B阻塞在条件变量等待上，线程A在收到符合要求的数据后，通过条件变量触发唤醒信号，同时保证同步安全。

具体实现步骤 #

1. Socket初始化（区分TCP/UDP） #

• TCP：需要创建监听Socket，绑定端口后调用listen()，后续通过accept()处理新连接，监听监听Socket和已连接Socket的可读事件。
• UDP：创建数据报Socket，绑定端口即可，直接监听该Socket的可读事件（UDP无连接，数据到达时触发可读）。

2. 线程A逻辑 #

• 初始化目标Socket后，使用select()阻塞等待Socket的可读事件。
• 当select()返回时，检查触发事件的Socket：
  • TCP场景：若为监听Socket，处理新连接；若为已连接Socket，读取数据。
  • UDP场景：直接读取数据。
• 验证数据是否符合要求，若符合则通过条件变量唤醒线程B。

3. 线程B逻辑 #

• 持续阻塞在pthread_cond_wait()调用上（该函数会自动释放互斥锁，被唤醒后重新获取锁）。
• 被唤醒后执行目标任务，完成后继续等待下一次唤醒。

代码示例 #

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <pthread.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <sys/select.h>
#include <string.h>

#define PORT 8080
#define BUFFER_SIZE 1024

// 线程同步对象
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
int active_socket = -1; // TCP连接Socket或UDP绑定Socket

// 线程B执行函数
void* thread_b_handler(void* arg) {
    while (1) {
        pthread_mutex_lock(&mutex);
        // 阻塞等待唤醒信号，自动释放互斥锁；唤醒后重新获取锁
        pthread_cond_wait(&cond, &mutex);

        // 此处添加业务逻辑处理
        printf("[Thread B] 已被唤醒，开始处理任务\n");

        pthread_mutex_unlock(&mutex);
    }
    return NULL;
}

// TCP模式下的线程A逻辑
void* thread_a_tcp_handler(void* arg) {
    int listen_fd = socket(AF_INET, SOCK_STREAM, 0);
    if (listen_fd < 0) {
        perror("TCP socket创建失败");
        exit(EXIT_FAILURE);
    }

    struct sockaddr_in server_addr = {
        .sin_family = AF_INET,
        .sin_port = htons(PORT),
        .sin_addr.s_addr = INADDR_ANY
    };

    if (bind(listen_fd, (struct sockaddr*)&server_addr, sizeof(server_addr)) < 0) {
        perror("TCP bind失败");
        close(listen_fd);
        exit(EXIT_FAILURE);
    }

    if (listen(listen_fd, 5) < 0) {
        perror("TCP listen失败");
        close(listen_fd);
        exit(EXIT_FAILURE);
    }

    fd_set read_fds;
    int max_fd = listen_fd;

    while (1) {
        FD_ZERO(&read_fds);
        FD_SET(listen_fd, &read_fds);
        if (active_socket != -1) {
            FD_SET(active_socket, &read_fds);
            max_fd = active_socket > listen_fd ? active_socket : listen_fd;
        }

        // 阻塞等待可读事件，无超时
        int ret = select(max_fd + 1, &read_fds, NULL, NULL, NULL);
        if (ret < 0) {
            perror("select调用失败");
            break;
        }

        // 处理新TCP连接
        if (FD_ISSET(listen_fd, &read_fds)) {
            struct sockaddr_in client_addr;
            socklen_t client_len = sizeof(client_addr);
            active_socket = accept(listen_fd, (struct sockaddr*)&client_addr, &client_len);
            if (active_socket < 0) {
                perror("accept失败");
                continue;
            }
            printf("[Thread A] 新客户端已连接：%s:%d\n", 
                   inet_ntoa(client_addr.sin_addr), ntohs(client_addr.sin_port));
        }

        // 处理TCP数据
        if (active_socket != -1 && FD_ISSET(active_socket, &read_fds)) {
            char buffer[BUFFER_SIZE];
            ssize_t recv_len = recv(active_socket, buffer, BUFFER_SIZE - 1, 0);
            if (recv_len <= 0) {
                printf("[Thread A] 客户端已断开连接\n");
                close(active_socket);
                active_socket = -1;
                continue;
            }
            buffer[recv_len] = '\0';
            printf("[Thread A] 收到TCP数据：%s\n", buffer);

            // 判断是否为触发唤醒的数据（示例：包含"trigger"字符串）
            if (strstr(buffer, "trigger") != NULL) {
                pthread_mutex_lock(&mutex);
                pthread_cond_signal(&cond); // 唤醒线程B
                pthread_mutex_unlock(&mutex);
                printf("[Thread A] 已触发线程B唤醒\n");
            }
        }
    }

    close(listen_fd);
    if (active_socket != -1) close(active_socket);
    return NULL;
}

// UDP模式下的线程A逻辑
void* thread_a_udp_handler(void* arg) {
    active_socket = socket(AF_INET, SOCK_DGRAM, 0);
    if (active_socket < 0) {
        perror("UDP socket创建失败");
        exit(EXIT_FAILURE);
    }

    struct sockaddr_in server_addr = {
        .sin_family = AF_INET,
        .sin_port = htons(PORT),
        .sin_addr.s_addr = INADDR_ANY
    };

    if (bind(active_socket, (struct sockaddr*)&server_addr, sizeof(server_addr)) < 0) {
        perror("UDP bind失败");
        close(active_socket);
        exit(EXIT_FAILURE);
    }

    fd_set read_fds;
    int max_fd = active_socket;

    while (1) {
        FD_ZERO(&read_fds);
        FD_SET(active_socket, &read_fds);

        // 阻塞等待UDP数据
        int ret = select(max_fd + 1, &read_fds, NULL, NULL, NULL);
        if (ret < 0) {
            perror("select调用失败");
            break;
        }

        if (FD_ISSET(active_socket, &read_fds)) {
            char buffer[BUFFER_SIZE];
            struct sockaddr_in client_addr;
            socklen_t client_len = sizeof(client_addr);
            ssize_t recv_len = recvfrom(active_socket, buffer, BUFFER_SIZE - 1, 0,
                                       (struct sockaddr*)&client_addr, &client_len);
            if (recv_len <= 0) {
                perror("recvfrom失败");
                continue;
            }
            buffer[recv_len] = '\0';
            printf("[Thread A] 收到UDP数据（来自%s:%d）：%s\n",
                   inet_ntoa(client_addr.sin_addr), ntohs(client_addr.sin_port), buffer);

            // 判断是否为触发唤醒的数据
            if (strstr(buffer, "trigger") != NULL) {
                pthread_mutex_lock(&mutex);
                pthread_cond_signal(&cond);
                pthread_mutex_unlock(&mutex);
                printf("[Thread A] 已触发线程B唤醒\n");
            }
        }
    }

    close(active_socket);
    return NULL;
}

int main(int argc, char* argv[]) {
    pthread_t thread_b;
    if (pthread_create(&thread_b, NULL, thread_b_handler, NULL) != 0) {
        perror("创建线程B失败");
        exit(EXIT_FAILURE);
    }

    pthread_t thread_a;
    if (argc > 1 && strcmp(argv[1], "udp") == 0) {
        if (pthread_create(&thread_a, NULL, thread_a_udp_handler, NULL) != 0) {
            perror("创建UDP模式线程A失败");
            exit(EXIT_FAILURE);
        }
    } else {
        if (pthread_create(&thread_a, NULL, thread_a_tcp_handler, NULL) != 0) {
            perror("创建TCP模式线程A失败");
            exit(EXIT_FAILURE);
        }
    }

    // 等待线程退出（实际场景可根据需求调整，比如捕获信号后优雅退出）
    pthread_join(thread_a, NULL);
    pthread_join(thread_b, NULL);

    // 清理同步资源
    pthread_mutex_destroy(&mutex);
    pthread_cond_destroy(&cond);

    return 0;
}

关键注意事项 #

• I/O多路复用选择：资源有限的平台优先用select()或poll()，两者都是POSIX标准；若需高并发支持，可改用Linux特有的epoll()，其性能更优。
• 线程同步正确性：必须使用pthread_cond_wait()配合互斥锁，该函数会自动处理锁的释放与重新获取，避免死锁。单纯解锁互斥锁无法主动唤醒线程B，仅当B尝试获取锁时才会继续执行，不符合需求。
• TCP连接管理：若需支持多客户端，需维护已连接Socket的列表，每次调用select()前将所有活跃Socket加入监听集合。
• 错误处理：示例代码简化了错误处理，实际开发中需完善资源释放逻辑（如Socket关闭、同步对象销毁），避免资源泄漏。

内容的提问来源于stack exchange，提问作者user44055