Linux下Unix Socket进程间通信的互斥处理方法咨询

针对Linux下多进程通过Unix Socket向单进程发送数据的场景，以下是几种实用的互斥与同步方案，覆盖发送端、接收端的不同需求：

一、发送端互斥：避免数据交织（针对Stream Socket） #

Unix Stream Socket是字节流协议，多个进程同时调用send()会导致不同进程的数据被无边界拼接，接收端无法区分独立消息。可通过以下方式实现发送互斥：

• 文件锁（flock）：
  所有发送进程共享一个锁文件，发送前获取排他锁，发送完成后释放。这种方式无需额外依赖，跨进程天然有效：

  // 发送进程代码片段
  int lock_fd = open("/tmp/socket_send_lock", O_CREAT | O_RDWR, 0666);
  if (lock_fd == -1) { /* 错误处理 */ }
  
  // 阻塞等待获取排他锁
  flock(lock_fd, LOCK_EX);
  send(sock_fd, data_buf, data_len, 0);
  // 释放锁
  flock(lock_fd, LOCK_UN);
  
  close(lock_fd);
  

• System V信号量：
  创建全局信号量集，初始值设为1，发送进程通过P/V操作实现互斥。适合需要更细粒度同步控制的场景：

  // 初始化信号量（仅执行一次，比如在第一个发送进程中）
  key_t sem_key = ftok("/tmp/sem_socket_key", 1);
  int sem_id = semget(sem_key, 1, IPC_CREAT | 0666);
  semctl(sem_id, 0, SETVAL, 1);
  
  // 发送进程代码片段
  struct sembuf sem_op;
  // P操作：获取锁
  sem_op.sem_num = 0;
  sem_op.sem_op = -1;
  sem_op.sem_flg = 0;
  semop(sem_id, &sem_op, 1);
  
  send(sock_fd, data_buf, data_len, 0);
  
  // V操作：释放锁
  sem_op.sem_op = 1;
  semop(sem_id, &sem_op, 1);
  

二、接收端互斥：多线程处理单套接字的场景 #

如果接收进程采用多线程共享同一个套接字进行recv()，内核无法保证线程间的接收顺序和数据完整性，需在接收端加互斥：

• POSIX互斥锁：
  初始化全局互斥锁，每个线程在调用recv()前加锁，处理完成后释放：

  // 全局互斥锁初始化
  pthread_mutex_t recv_mutex = PTHREAD_MUTEX_INITIALIZER;
  
  // 线程处理函数
  void* recv_handler(void* arg) {
      int sock_fd = *(int*)arg;
      char recv_buf[1024];
      ssize_t recv_len;
  
      while (1) {
          pthread_mutex_lock(&recv_mutex);
          recv_len = recv(sock_fd, recv_buf, sizeof(recv_buf), 0);
          pthread_mutex_unlock(&recv_mutex);
  
          if (recv_len <= 0) break;
          // 处理接收的数据
      }
      return NULL;
  }
  

三、更简便的替代方案：使用Datagram Unix Socket #

Unix数据报套接字（SOCK_DGRAM）是面向消息的，内核保证每个sendto()的消息会被完整接收，即使多个进程同时发送，接收端也能通过recvfrom()拿到独立的完整消息，无需额外互斥逻辑。适合不需要严格字节流顺序的场景：

• 发送端示例：

  int sock_fd = socket(AF_UNIX, SOCK_DGRAM, 0);
  struct sockaddr_un server_addr;
  memset(&server_addr, 0, sizeof(server_addr));
  server_addr.sun_family = AF_UNIX;
  strncpy(server_addr.sun_path, "/tmp/udp_unix_socket", sizeof(server_addr.sun_path)-1);
  
  sendto(sock_fd, data_buf, data_len, 0, (struct sockaddr*)&server_addr, sizeof(server_addr));
  close(sock_fd);
  

• 接收端示例：

  int sock_fd = socket(AF_UNIX, SOCK_DGRAM, 0);
  struct sockaddr_un server_addr;
  memset(&server_addr, 0, sizeof(server_addr));
  server_addr.sun_family = AF_UNIX;
  strncpy(server_addr.sun_path, "/tmp/udp_unix_socket", sizeof(server_addr.sun_path)-1);
  
  bind(sock_fd, (struct sockaddr*)&server_addr, sizeof(server_addr));
  
  char recv_buf[1024];
  ssize_t recv_len;
  while ((recv_len = recvfrom(sock_fd, recv_buf, sizeof(recv_buf), 0, NULL, NULL)) > 0) {
      // 处理完整的单条消息
  }
  close(sock_fd);
  

四、经典C/S模型：接收端为每个连接分配独立处理单元 #

如果使用Stream Socket，接收端可以通过accept()为每个发送进程建立独立的连接套接字，然后用单独的线程或子进程处理该连接。这种方式天然避免了单套接字的竞争问题，是最常用的高并发处理方案：

• 接收端主进程代码：

  int listen_fd = socket(AF_UNIX, SOCK_STREAM, 0);
  // 绑定、监听套接字步骤省略
  
  while (1) {
      int conn_fd = accept(listen_fd, NULL, NULL);
      if (conn_fd == -1) continue;
  
      // 创建线程处理该连接
      pthread_t tid;
      pthread_create(&tid, NULL, conn_handler, &conn_fd);
      pthread_detach(tid); // 自动回收线程资源
  }
  

• 连接处理线程：

  void* conn_handler(void* arg) {
      int conn_fd = *(int*)arg;
      char recv_buf[1024];
      ssize_t recv_len;
  
      while ((recv_len = recv(conn_fd, recv_buf, sizeof(recv_buf), 0)) > 0) {
          // 处理当前连接的数据
      }
      close(conn_fd);
      return NULL;
  }
  

内容的提问来源于stack exchange，提问作者Miguel Perez Avila