C/C++如何使用recv在AF_UNIX套接字中逐消息逐行读取数据

你遇到的问题和recv本身的行为无关，核心是两个认知偏差加代码逻辑漏洞：

1. AF_UNIX的SOCK_STREAM类型套接字是无消息边界的字节流，行为和TCP完全一致：一次recv调用可能返回半条未接收完整的消息，也可能一次性返回多条拼接在一起的消息，不存在“每次调用刚好读一条消息”的天然保证。
2. 原有代码的明确漏洞：
   • recv只会把实际读到的字节写入缓冲区，不会自动添加C风格字符串的结束符\0。你用std::string messageRecieved(buffer)构造字符串时，会从缓冲区起始地址一直向后遍历内存，直到碰到随机出现的\0才停止，属于未定义行为。如果新读到的内容比第一条消息短，后续内存里残留的旧数据没有被覆盖，构造出的字符串就会一直包含第一条消息的内容，这就是你观察到“重复读第一条”的核心原因。
   • 每次读到数据后只截取到第一个换行就丢弃缓冲区剩余内容，既会丢失同一次recv返回的后续多条消息，也无法处理一次recv没读到换行符的半包场景。
   • 流套接字不存在“需要刷新缓冲区”的说法，内核维护的读指针会随着recv调用自动后移，不存在“每次从起始位置读”的情况，你观察到的重复内容完全是用户态缓冲区处理错误导致的。

核心是维护一块用户态的残留缓冲区：每次调用recv读到新数据后，先把实际读到的字节追加到缓冲区末尾，再循环检查缓冲区中是否存在换行符：存在就切出换行前的完整消息交给业务逻辑处理，删掉缓冲区中已经处理完的前缀（含换行符），直到缓冲区里没有完整换行，再发起下一次recv调用。

C++ 手动维护缓冲区实现（兼容阻塞/非阻塞模式） #

这个实现逻辑和socat的行处理行为完全一致，支持自定义消息回调，也可以直接适配epoll/select的非阻塞场景：

#include <sys/socket.h>
#include <sys/un.h>
#include <unistd.h>
#include <string>
#include <iostream>
#include <functional>
#include <cstring>
#include <cerrno>

// 替换成你自己的消息处理逻辑
void handle(const std::string& msg) {
    std::cout << "Received message: " << msg << std::endl;
}

void recv_loop(int sock_fd, std::function<void(const std::string&)> msg_handler) {
    std::string read_buf; // 全局残留缓冲区，存未处理完的半包数据
    char tmp_buf[1024];   // 临时读缓冲，大小不需要和单条消息上限绑定

    while (true) {
        ssize_t n = recv(sock_fd, tmp_buf, sizeof(tmp_buf), 0);
        if (n > 0) {
            // 仅追加实际读到的n字节到残留缓冲区，避免脏数据
            read_buf.append(tmp_buf, n);

            // 循环切分当前缓冲区里所有完整的换行消息
            size_t newline_pos;
            while ((newline_pos = read_buf.find('\n')) != std::string::npos) {
                // 取出换行前的完整消息，如果是\r\n结尾可额外去掉末尾的\r
                std::string msg = read_buf.substr(0, newline_pos);
                // 从缓冲区移除已经处理完的消息和换行符
                read_buf.erase(0, newline_pos + 1);
                // 触发业务处理
                msg_handler(msg);
            }
        } else if (n == 0) {
            // 对端正常关闭连接
            std::cerr << "Peer closed connection" << std::endl;
            break;
        } else {
            if (errno == EINTR) continue; // 被信号中断直接重试
            // 非阻塞模式下碰到EAGAIN/EWOULDBLOCK直接跳出循环，等下次可读事件再读即可
            if (errno == EAGAIN || errno == EWOULDBLOCK) break;
            std::cerr << "Recv error: " << strerror(errno) << std::endl;
            break;
        }
    }
}

// AF_UNIX套接字连接示例
int connect_unix(const std::string& sock_path) {
    int fd = socket(AF_UNIX, SOCK_STREAM, 0);
    if (fd < 0) {
        perror("Create socket failed");
        return -1;
    }

    struct sockaddr_un addr;
    memset(&addr, 0, sizeof(addr));
    addr.sun_family = AF_UNIX;
    strncpy(addr.sun_path, sock_path.c_str(), sizeof(addr.sun_path) - 1);

    if (connect(fd, (struct sockaddr*)&addr, sizeof(addr)) < 0) {
        perror("Connect unix socket failed");
        close(fd);
        return -1;
    }
    return fd;
}

int main() {
    int sock = connect_unix("/tmp/myepicsocket.sock");
    if (sock < 0) return -1;
    recv_loop(sock, handle);
    close(sock);
    return 0;
}

极简C风格实现（基于fdopen+getline，仅阻塞场景） #

如果不需要非阻塞支持，可以直接把套接字描述符包装成C标准库文件流，用getline自动完成缓冲区维护、换行识别、半包拼接，代码量更小，行为和shell的while read完全一致：

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <unistd.h>

// 替换成你自己的消息处理逻辑
void handle(const char* line) {
    printf("Received message: %s\n", line);
}

int main() {
    int sock_fd = socket(AF_UNIX, SOCK_STREAM, 0);
    struct sockaddr_un addr;
    memset(&addr, 0, sizeof(addr));
    addr.sun_family = AF_UNIX;
    strncpy(addr.sun_path, "/tmp/myepicsocket.sock", sizeof(addr.sun_path) - 1);
    
    if (connect(sock_fd, (struct sockaddr*)&addr, sizeof(addr)) < 0) {
        perror("Connect failed");
        close(sock_fd);
        return -1;
    }

    // 将套接字包装为标准IO流
    FILE* sock_fp = fdopen(sock_fd, "r");
    if (!sock_fp) {
        perror("fdopen failed");
        close(sock_fd);
        return -1;
    }

    char* linebuf = NULL;
    size_t buflen = 0;
    ssize_t nread;
    // getline自动处理缓冲区扩容、半包拼接，直到读到换行/连接断开才返回
    while ((nread = getline(&linebuf, &buflen, sock_fp)) != -1) {
        // 去掉末尾的换行符
        if (nread > 0 && linebuf[nread-1] == '\n') {
            linebuf[nread-1] = '\0';
            // 适配\r\n换行的场景可额外去掉末尾\r
            if (nread > 1 && linebuf[nread-2] == '\r') linebuf[nread-2] = '\0';
        }
        handle(linebuf);
    }

    // 资源清理：fclose会自动关闭底层套接字，不需要重复调用close
    free(linebuf);
    fclose(sock_fp);
    return 0;
}

• 如果你用的是SOCK_DGRAM类型的AF_UNIX数据报套接字，recv会保证每次返回一条完整数据报，但如果单条消息大小超过你传入的缓冲长度，超出部分会被内核直接截断丢弃。你用socat+shell read的逻辑是基于流模式的，默认用SOCK_STREAM即可和原行为对齐。
• 任何时候操作套接字读缓冲，都要以recv返回的实际字节数为准，不要直接把裸字符数组当以\0结尾的C字符串处理，避免读到脏数据触发未定义行为。
• 非阻塞场景下不要用fdopen+getline的方案，标准IO流的缓冲机制在非阻塞下容易出现死锁或事件漏触发，手动维护缓冲区的实现可以无缝兼容IO多路复用场景。

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