C语言跨进程AF_LOCAL socket通信的示例，其中server端是一个进程，client端是多个线程，并且使用epoll管理server端，其中client端通过read()接受数据，重复发送10次相同数据，数据中携带线程信息，并且每次发送前都重先建联socket然后发送完之后再断连socket。其中server端会根据client线程数目创建相同数量的线程专门监听该连接，并在线程中发送和接受数据。
server
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/socket.h>
#include <sys/epoll.h>
#include <sys/un.h>
#include <fcntl.h>
#include <errno.h>
#include <pthread.h>#define SOCKET_PATH "/tmp/example_socket"
#define BUFFER_SIZE 1024
#define MAX_EVENTS 10
#define MAX_CLIENTS 5typedef struct {long thread_id;int sequence_number;char message[BUFFER_SIZE];
} Message;void error_and_exit(const char *msg) {perror(msg);exit(EXIT_FAILURE);
}void *connection_thread(void *arg) {int client_socket = *(int *)arg;Message buffer;int bytes_read;// 读取客户端消息while ((bytes_read = read(client_socket, &buffer, sizeof(buffer))) > 0) {// 输出客户端消息printf("Received from thread %ld sequence number %d: %s\n", buffer.thread_id, buffer.sequence_number, buffer.message);// 发送响应const char *response = "Hello from server";if (write(client_socket, response, strlen(response)) < 0) {error_and_exit("Write failed");}// 更新序列号buffer.sequence_number++;}// 检查错误if (bytes_read == -1) {if (errno == EAGAIN || errno == EWOULDBLOCK) {// No data availableprintf("No data available\n");} else {perror("Read failed");}}// 关闭客户端套接字close(client_socket);// 释放线程参数free(arg);pthread_exit(NULL);
}int main() {int server_socket, epoll_fd, client_socket;struct sockaddr_un addr;struct epoll_event events[MAX_EVENTS];socklen_t len;int num_events;// 清理套接字文件if (unlink(SOCKET_PATH) == -1 && errno != ENOENT) {perror("Failed to remove socket file");exit(EXIT_FAILURE);}// 创建套接字if ((server_socket = socket(AF_LOCAL, SOCK_STREAM, 0)) < 0) {error_and_exit("Socket creation failed");}// 清空地址结构memset(&addr, 0, sizeof(addr));addr.sun_family = AF_LOCAL;strncpy(addr.sun_path, SOCKET_PATH, sizeof(addr.sun_path) - 1);// 绑定套接字if (bind(server_socket, (struct sockaddr *)&addr, sizeof(addr)) < 0) {error_and_exit("Bind failed");}// 监听连接if (listen(server_socket, MAX_CLIENTS) < 0) {error_and_exit("Listen failed");}// 创建epoll实例if ((epoll_fd = epoll_create1(0)) < 0) {error_and_exit("Epoll creation failed");}// 添加监听套接字到epollstruct epoll_event ev;ev.events = EPOLLIN | EPOLLET; // 设置为边缘触发ev.data.fd = server_socket;if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, server_socket, &ev) < 0) {error_and_exit("Epoll add failed");}// 设置为非阻塞模式if (fcntl(server_socket, F_SETFL, O_NONBLOCK) < 0) {error_and_exit("Non-blocking mode failed");}int num_client_threads = 0;while (1) {// 等待事件num_events = epoll_wait(epoll_fd, events, MAX_EVENTS, -1);if (num_events == -1) {error_and_exit("Epoll wait failed");}for (int i = 0; i < num_events; i++) {if (events[i].data.fd == server_socket) {// 接受连接len = sizeof(addr);if ((client_socket = accept(server_socket, (struct sockaddr *)&addr, &len)) < 0) {error_and_exit("Accept failed");}// 设置客户端套接字为非阻塞if (fcntl(client_socket, F_SETFL, O_NONBLOCK) < 0) {error_and_exit("Non-blocking mode failed");}// 读取客户端的第一个消息以获取线程信息Message thread_info;ssize_t bytes_received = recv(client_socket, &thread_info, sizeof(thread_info), MSG_DONTWAIT);if (bytes_received == -1) {if (errno == EAGAIN || errno == EWOULDBLOCK) {// No data available yetcontinue;} else {perror("Read failed");close(client_socket);continue;}}// 创建线程处理客户端连接int *client_socket_ptr = malloc(sizeof(int));*client_socket_ptr = client_socket;if (pthread_create(NULL, NULL, connection_thread, client_socket_ptr) != 0) {error_and_exit("Thread creation failed");}num_client_threads++; // 跟踪客户端线程数} else {// 不处理其他事件continue;}}// 如果客户端线程数已知并且所有客户端线程都已启动，则退出循环if (num_client_threads >= 5) {break;}}// 等待所有客户端线程完成for (int i = 0; i < num_client_threads; i++) {pthread_join(i, NULL);}close(server_socket);close(epoll_fd);return 0;
}client
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <pthread.h>
#include <errno.h>#define SOCKET_PATH "/tmp/example_socket"
#define BUFFER_SIZE 1024
#define NUM_REPETITIONS 10typedef struct {long thread_id;int sequence_number;char message[BUFFER_SIZE];
} Message;void error_and_exit(const char *msg) {perror(msg);exit(EXIT_FAILURE);
}void *client_thread(void *arg) {int client_socket;struct sockaddr_un addr;Message buffer;long thread_id = (long)arg;// 重复建立连接、发送数据、接收响应、然后断开连接for (int j = 0; j < NUM_REPETITIONS; j++) {// 创建套接字if ((client_socket = socket(AF_LOCAL, SOCK_STREAM, 0)) < 0) {error_and_exit("Socket creation failed");}// 清空地址结构memset(&addr, 0, sizeof(addr));addr.sun_family = AF_LOCAL;strncpy(addr.sun_path, SOCKET_PATH, sizeof(addr.sun_path) - 1);// 连接到服务器if (connect(client_socket, (struct sockaddr *)&addr, sizeof(addr)) < 0) {error_and_exit("Connect failed");}// 构造消息buffer.thread_id = thread_id;buffer.sequence_number = j;snprintf(buffer.message, sizeof(buffer.message), "Hello from client thread %ld repetition %d", thread_id, j);// 向服务器发送消息if (write(client_socket, &buffer, sizeof(buffer)) < 0) {error_and_exit("Write failed");}// 读取服务器的响应if (read(client_socket, &buffer, sizeof(buffer)) < 0) {error_and_exit("Read failed");}// 输出服务器的响应puts(buffer.message);// 断开连接close(client_socket);}pthread_exit(NULL);
}int main() {pthread_t threads[5];// 启动多个线程for (int i = 0; i < 5; i++) {if (pthread_create(&threads[i], NULL, client_thread, (void *)i) != 0) {error_and_exit("Thread creation failed");}}// 等待所有线程完成for (int i = 0; i < 5; i++) {if (pthread_join(threads[i], NULL) != 0) {error_and_exit("Thread join failed");}}return 0;
}
