zxyの神秘小屋

large picture of the cover

Linux端下的 并发服务器

  • 多进程服务器--创建多个进程提供服务

  • 多路复用服务器--通过捆绑并统一管理I/O对象提供服务

  • 多线程服务器--通过生成与客户端等量的线程提供服务

进程

wait() & waitpid()

在 Linux C++ 编程中，wait 和 waitpid 函数用于进程控制，特别是父进程等待其子进程的状态变化。以下是这两个函数的主要用途和区别：

wait 函数

用途：

wait 函数使父进程阻塞，直到任一子进程终止。它的功能是等待一个终止的子进程，并获取该子进程的终止状态。

#include <sys/types.h>
#include <sys/wait.h>

pid_t wait(int *status);

参数：

  • status：指向一个整数变量的指针，用于存储子进程的终止状态。如果不关心子进程的终止状态，可以传递 NULL。

返回值：

  • 返回终止的子进程的 PID。

  • 如果调用出错，返回 -1，并设置 errno 来指示错误。

waitpid 函数

用途：

waitpid 函数提供了更精细的控制，允许父进程等待指定的子进程，并可以选择是否阻塞。它可以用于等待特定的子进程终止，或者使用选项来非阻塞地等待进程状态变化。

原型：

#include <sys/types.h>
#include <sys/wait.h>

pid_t waitpid(pid_t pid, int *status, int options);

参数：

  • pid：指定要等待的子进程的 PID。

    • pid > 0：等待进程 ID 为 pid 的子进程。

    • pid = 0：等待与调用进程在同一进程组中的任何子进程。

    • pid < -1：等待进程组 ID 为 |pid| 的子进程。

    • pid = -1：等待任一子进程（与 wait 行为相同）。

  • status：指向一个整数变量的指针，用于存储子进程的终止状态。可以传递 NULL。

  • options：用于控制函数的行为。常用的选项包括：

    • WNOHANG：如果没有子进程终止，不阻塞并立即返回。

    • WUNTRACED：返回已停止的子进程状态，直到其被追踪或继续执行。

返回值：

  • 返回等待到的子进程的 PID。

  • 如果调用出错，返回 -1，并设置 errno 来指示错误。

示例代码

#include <sys/types.h>
#include <sys/wait.h>
#include <unistd.h>
#include <iostream>

int main() {
    pid_t pid = fork();
    
    if (pid == -1) {
        std::cerr << "fork failed" << std::endl;
        return 1;
    } else if (pid == 0) {
        // 子进程
        std::cout << "Child process: " << getpid() << std::endl;
        sleep(2); // 模拟子进程工作
        return 0;
    } else {
        // 父进程
        int status;
        pid_t child_pid = wait(&status);
        
        if (child_pid == -1) {
            std::cerr << "wait failed" << std::endl;
            return 1;
        }
        
        if (WIFEXITED(status)) {
            std::cout << "Child process " << child_pid << " exited with status " << WEXITSTATUS(status) << std::endl;
        } else {
            std::cout << "Child process " << child_pid << " did not exit normally" << std::endl;
        }
        
        return 0;
    }
}

上述代码展示了如何使用 wait 函数等待子进程终止并获取其终止状态。对于更精细的进程控制，可以使用 waitpid 函数。

僵尸进程

僵尸进程（Zombie process）是当子进程比父进程先结束，而父进程又没有回收子进程，释放子进程占用的资源，此时子进程将成为一个僵尸进程。如果父进程先退出 ，子进程被init接管，子进程退出后init会回收其占用的相关资源

  void Zombie_process() {
	pid_t pid = fork();
	if (pid > 0) {
		sleep(30);
		int status;
		waitpid(pid,&status,0);
	}
	else { //子进程会一直挂在后台
		printf("%s(%d):%s\n",__FILE__,__LINE__,__FUNCTION__);
		exit(-1);
	}
}

信号处理

在 Linux 和类 Unix 系统中，信号（Signal）是一种用于进程间通信的机制，用于通知进程某些事件的发生。信号处理（Signal Handling）指的是进程如何捕获和响应这些信号。以下是信号处理的简单介绍：

常见的信号

以下是一些常见的信号及其含义：

  • SIGINT：中断信号（通常由 Ctrl+C 触发），用于请求进程终止。

  • SIGTERM：终止信号，通常用于有礼貌地请求进程终止。

  • SIGKILL：杀死信号，用于强制终止进程，不能被捕获或忽略。

  • SIGSTOP：停止信号，暂停进程的执行，不能被捕获或忽略。

  • SIGCONT：继续信号，恢复被暂停的进程的执行。

  • SIGCHLD：当子进程状态变化（如终止或暂停）时发送给父进程。

  • SIGALRM：定时器信号，由 alarm 函数设定的定时器到期时发送。

  • SIGHUP：挂起信号，当控制终端关闭时发送。

信号处理函数

进程可以通过设置信号处理函数来自定义如何处理某些信号。通常使用 signal 或 sigaction 函数来设置信号处理函数。

signal 函数

#include <csignal>

void (*signal(int signum, void (*handler)(int)))(int);

  • signum：要处理的信号编号。

  • handler：指向信号处理函数的指针，或特殊值 SIG_IGN（忽略信号）和 SIG_DFL（使用默认处理）。

示例代码

#include <csignal>
#include <iostream>
#include <unistd.h>

void signalHandler(int signum) {
    std::cout << "Interrupt signal (" << signum << ") received.\n";
    // 清理并关闭程序
    exit(signum);
}

int main() {
    // 注册 SIGINT 信号处理器
    signal(SIGINT, signalHandler);

    while (true) {
        std::cout << "Going to sleep..." << std::endl;
        sleep(1);
    }

    return 0;
}

#include<signal.h>
void signal_func(int sig) {
	switch (sig) {
	case SIGINT:
		printf("Ctrl+c press...\n");
		exit(0);
		break;
	case SIGALRM:
		printf("tid:%d pid:%d\n", gettid(), getpid());
		alarm(2);
		break;
	default:
		break;
	}
}

void regist_signal() {
	//printf("main thread tid:%d pid:%d\n", gettid(), getpid());
	signal(SIGINT,signal_func);
	signal(SIGALRM,signal_func);
	alarm(1);
	while (1) {
		printf("main thread tid:%d pid:%d\n", gettid(), getpid());
		sleep(6);
	}
}

sigaction 函数

sigaction 提供了更高级的信号处理功能，可以设置更多的信号处理选项。

#include <signal.h>

int sigaction(int signum, const struct sigaction *act, struct sigaction *oldact);

  • signum：要处理的信号编号。

  • act：指向一个 sigaction 结构体的指针，用于指定新的信号处理行为。

  • oldact：指向一个 sigaction 结构体的指针，用于保存以前的信号处理行为（如果不需要，可以传递 NULL）。

struct sigaction {
    void (*sa_handler)(int);
    void (*sa_sigaction)(int, siginfo_t *, void *);
    sigset_t sa_mask;
    int sa_flags;
    void (*sa_restorer)(void);
};
#include <iostream>
#include <csignal>
#include <unistd.h>

void signalHandler(int signum, siginfo_t *info, void *context) {
    std::cout << "Interrupt signal (" << signum << ") received." << std::endl;
    exit(signum);
}

int main() {
    struct sigaction action;
    action.sa_sigaction = signalHandler;
    action.sa_flags = SA_SIGINFO;

    sigaction(SIGINT, &action, NULL);

    while (true) {
        std::cout << "Going to sleep..." << std::endl;
        sleep(1);
    }

    return 0;
}

总结

信号处理是进程间通信和控制的重要机制。通过设置信号处理函数，进程可以自定义如何响应各种信号。使用 signal 函数可以简单地设置信号处理函数，而 sigaction 函数提供了更高级的信号处理选项。

基于多任务的并发服务器

多进程并发服务器

#include<unistd.h>
#include<arpa/inet.h>
#include<signal.h>
#include<sys/socket.h>
#include<sys/wait.h>

void hand_childProc(int sig) { //子进程的回收
	pid_t pid;
	int status{ 0 };
	pid = waitpid(-1,&status,WNOHANG);
	printf("%s(%d)%s removed sub proc:%d\r\n",__FILE__,__LINE__,__FUNCTION__,pid);
}

void server85() {
	struct sigaction act;
	act.sa_flags = 0;
	act.sa_handler = hand_childProc;
	sigaction(SIGCHLD,&act,0); //当子进程结束时 触发信号

	struct sockaddr_in server_addr, client_addr;
	int server_socket = socket(PF_INET,SOCK_STREAM,0);

	memset(&server_addr,0,sizeof(server_addr));
	server_addr.sin_family = AF_INET;
	server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
	server_addr.sin_port = htons(2233);

	if (bind(server_socket, (struct sockaddr*)&server_addr, sizeof(server_addr)) == -1) {
		error_handling("bind failed");
	}

	if (listen(server_socket, 5)) {
		error_handling("listen failed");
	}
	int count{ 0 };
	while (1) {
		socklen_t len = sizeof(client_addr);
		int client = accept(server_socket, (sockaddr*)&client_addr, &len);
		if (client >= 0) {
			pid_t pid = fork();
			count++;
			if (pid == 0) {
				close(server_socket); //fork复制出来的
				char buffer[2048]{ "" };
				ssize_t length{ 0 };
				while ((length = read(client, buffer, sizeof(buffer))) > 0) {
					write(client, buffer, length);
				}
				close(client);
				printf("%s(%d)%s client is closed\n", __FILE__, __LINE__, __FUNCTION__);
				return;
			}
			if (pid < 0) {
				close(client);
				printf("%s(%d)%s fork is failed\n", __FILE__, __LINE__, __FUNCTION__);
				break;
			}
			close(client);
		}
		if (count >= 5) break;
	}
	
	close(server_socket);
	return;

}

void client85() {
	int client_socket = socket(PF_INET, SOCK_STREAM, 0);
	struct sockaddr_in servaddr;
	memset(&servaddr, 0, sizeof(servaddr));
	servaddr.sin_family = AF_INET; //设置Net
	servaddr.sin_addr.s_addr = inet_addr("127.0.0.1"); //要发送服务器的ip地址
	servaddr.sin_port = htons(2233);
	int ret_connect = connect(client_socket, (struct sockaddr*)&servaddr, sizeof(servaddr));
	while (ret_connect == 0) {
		char buffer[256]{ "zxy 666 你好@123!" };
		printf("%s(%d)%s 链接成功！\n", __FILE__, __LINE__, __FUNCTION__);

		size_t len = strlen(buffer);
		size_t send_len{ 0 };
		while (send_len < len) {
			ssize_t ret = write(client_socket, buffer + send_len, len - send_len); //向服务器分包发送数据
			if (ret <= 0) {
				fputs("write failed!\n", stdout);
				close(client_socket);
				printf("%s(%d)%s if (ret <= 0) done \n", __FILE__, __LINE__, __FUNCTION__);
				break;
			}
			printf("%s(%d)%s 发送成功！\n", __FILE__, __LINE__, __FUNCTION__);
			send_len += (size_t)ret;
		}
		memset(buffer, 0, sizeof(buffer));

		size_t read_len = 0;
		while (read_len < len) {
			ssize_t ret = read(client_socket, buffer + read_len, len - read_len); //接收数据
			if (ret <= 0) {
				printf("%s(%d)%s 接收到:%s\n", __FILE__, __LINE__, __FUNCTION__,buffer);
				close(client_socket);
				std::cout << "client done!" << std::endl;
				break;
			}
			read_len += (size_t)ret;
		}
		std::cout << "from server: " << buffer << std::endl;
		break;
	}
	//printf("%s(%d):%s %d\n",__FILE__,__LINE__,__FUNCTION__, ret_connect);
	sleep(2);
	close(client_socket);
	std::cout << "client done!" << std::endl;
}

void 多进程服务器() {
	pid_t pid = fork();
	if (pid == 0) {//子进程
		//启动服务器
		server85();
	}
	else if(pid > 0){//主进程
		//启动客户端
		sleep(1);
		printf("%s(%d)%s wait server invoking\n", __FILE__, __LINE__, __FUNCTION__);
		for (size_t i = 0; i < 5; i++) {
			pid = fork();
			if (pid > 0) {
				continue;
			}
			else if(pid == 0){
				//启动客户端
				client85();
				break;
			}
		}
	}
	else {
		printf("%s(%d)%s fork failed！\n", __FILE__, __LINE__, __FUNCTION__);

	}
}

int main(int argc, char* argv[]) {
	多进程服务器();
	return 0;
}

进程间通信

ICP (InterProcess Communication) 进程间通信，通过内核提供的缓冲区进行数据交换的机制。

进程间通信意味着两个不同进程间可以交换数据，为了完成这一点，操作系统应提供两个进程可以同时访问的内存空间。

pipe管道

匿名管道

#include<unistd.h>
int pipe(int filedes[2]);
//成功时返回0 失败返回-1
//filedes[0] 通过管道 接收数据 时使用的文件描述符，即管道出口
//filedes[1] 通过管道 传输数据 时使用的文件描述符，即管道入口

//单向管道
void pipe_fun() {
	int fds[2]{ -1,-1 };
	char str[64]{"send by sub process!\n"};
	char buffer[128]{""};
	pipe(fds);
	pid_t pid = fork();
	if (pid == 0) {
		//子进程
		write(fds[1],str,sizeof(str));
		printf("%s(%d)%s send str:%s \n", __FILE__, __LINE__, __FUNCTION__, str);
	}
	else {
		read(fds[0],buffer,sizeof(buffer));
		printf("%s(%d)%s read buffer:%s \n", __FILE__, __LINE__, __FUNCTION__,buffer);
	}
}
//双向管道
void pipe_fun() {
	int s2c[2]{-1,-1}, c2s[2]{ -1,-1 };
	char buffer[128]{ "" };
	char str_c2s[64]{ "send by client" };
	char str_s2c[64]{ "send by server" };

	if (pipe(c2s) == -1) {
		std::cout << "pipe failed" << std::endl;
	}
	if (pipe(s2c) == -1) {
		std::cout << "pipe failed" << std::endl;

	}
	pid_t pid = fork();

	if (pid == 0) {
		write(c2s[1],str_c2s,strlen(str_c2s));
		read(s2c[0],buffer,sizeof(buffer));
		printf("%s(%d)%s sub process read buffer:%s \n", __FILE__, __LINE__, __FUNCTION__, buffer);
	}
	else {
		memset(buffer,0,sizeof(buffer));
		read(c2s[0],buffer,sizeof(buffer));
		printf("%s(%d)%s read buffer:%s \n", __FILE__, __LINE__, __FUNCTION__, buffer);
		write(s2c[1], str_s2c, strlen(str_s2c));
	}
}

fifo管道

先进先出管道

命名管道

（linux操作系统中的一个fifo命名文件）

#include <sys/stat.h>
#include<sys/file.h>
void fifo_fun() {
	mkfifo("./a.fifo",0666);
	pid_t pid = fork();
	if (pid == 0){
		int fd = open("./a.fifo",O_RDONLY);
		char buffer[128]{ "" };
		ssize_t len = read(fd,buffer,sizeof(buffer));
		printf("%s(%d)%s read buffer:%s len:%d\n", __FILE__, __LINE__, __FUNCTION__, buffer, len);
		close(fd);
	}
	else {
		int fd = open("./a.fifo", O_WRONLY);
		char str[]{"你好 世界!\n"};
		write(fd, str , strlen(str));
		printf("%s(%d)%s\n", __FILE__, __LINE__, __FUNCTION__);
		close(fd);
	}
}

//read 端会阻塞等待write端打开open

共享内存

它允许多个进程访问同一块内存区域，以此作为进程间通信（IPC, Inter-Process Communication）的一种方式。这种方式相对于管道、套接字等通信手段，具有更高的效率，因为数据不需要在用户空间和内核空间之间进行复制，也不需要经过序列化和反序列化的复杂过程。

如果另一个进程，也通过上述方式，通过页表映射到同一段物理内存，那就实现了让多个进程看到同一段空间，这样当一个进程向这段物理空间写入数据，另一个进程就可以马上从这段空间读取数据了，就可以实现进程间的通信了


  1. 创建或者获取一段共享内存；

  2. 将上一步创建的共享内存映射到该进程的地址空间；

  3. 访问共享内存；

  4. 将共享内存从当前的进程地址空间分离；

  5. 删除这段共享内存；

void icp_shm() {
	pid_t pid = fork();

	if (pid == 0) {
		usleep(10000);//100ms等待父进程写入
		//子进程
		int shm_id = shmget(ftok(".", 1), sizeof(STUDENT), IPC_CREAT | 0666);
		//创建共享内存
		if (shm_id == -1) {
			printf("%s(%d)%s create shm failed\n", __FILE__, __LINE__, __FUNCTION__);
		}
		//映射
		P_STUDENT pStu = (P_STUDENT)shmat(shm_id, NULL, 0);
		while (pStu->signal != 99) {
			usleep(1000);
		}
		printf("id:%d age:%d sex:%d name:%s\n",pStu->id,pStu->age,pStu->sex,pStu->name);
		pStu->signal = 0;
		//删除地址映射
		shmdt(pStu);
		//删除共享内存
		shmctl(shm_id, IPC_RMID, NULL);
	}
	else {
		//主进程
		int shm_id = shmget(ftok(".",1), sizeof(STUDENT), IPC_CREAT | 0666); 
		//创建共享内存
		if (shm_id == -1) {
			printf("%s(%d)%s create shm failed\n", __FILE__, __LINE__, __FUNCTION__);
		}
		//映射
		P_STUDENT pStu = (P_STUDENT)shmat(shm_id,NULL,0);

		//写入数据
		pStu->id = 123456;
		strcpy(pStu->name,"赵旭阳");
		pStu->age = 22;
		pStu->sex = true;
		pStu->signal = 99;

		while (pStu->signal == 99) { //共享内存 被修改后拿到信号变化则跳出
			usleep(1000);
		}

		//删除地址映射
		shmdt(pStu);
		//删除共享内存
		shmctl(shm_id,IPC_RMID,NULL);

	}
}

信号量

用户进程可以通过使用操作系统提供的一对原语来对信号量进行操作，从而很方便的实现了进程互斥、进程同步。

void icp_shm() {
	pid_t pid = fork();

	if (pid == 0) {
		//子进程

		//初始化信号量
		key_t key = ftok(".", 2);
		int sem_id = semget(key, 2, IPC_CREAT);

		int shm_id = shmget(ftok(".", 1), sizeof(STUDENT), IPC_CREAT | 0666);
		//创建共享内存
		if (shm_id == -1) {
			printf("%s(%d)%s create shm failed\n", __FILE__, __LINE__, __FUNCTION__);
			return;
		}

		sembuf sop = {
			.sem_num = 0,
			.sem_op = -1,//P操作
		};
		semop(sem_id, &sop, 1);		//P操作
		//映射
		P_STUDENT pStu = (P_STUDENT)shmat(shm_id, NULL, 0);
		printf("id:%d age:%d sex:%d name:%s\n", pStu->id, pStu->age, pStu->sex, pStu->name);
		//信号量
		sop.sem_num = 1;
		sop.sem_op = 1;//V操作
		semop(sem_id, &sop, 1);		//V操作

		//删除地址映射
		shmdt(pStu);
		//删除共享内存
		shmctl(shm_id, IPC_RMID, NULL);
		
	}
	else {
		//主进程
		key_t key = ftok(".",2);
		int sem_id = semget(key,2,IPC_CREAT);
		//初始化信号量
		semctl(sem_id, 0, SETVAL, 0);
		semctl(sem_id, 1, SETVAL, 0);

		int shm_id = shmget(ftok(".",1), sizeof(STUDENT), IPC_CREAT | 0666); 
		//创建共享内存
		if (shm_id == -1) {
			printf("%s(%d)%s create shm failed\n", __FILE__, __LINE__, __FUNCTION__);
		}
		//映射
		P_STUDENT pStu = (P_STUDENT)shmat(shm_id,NULL,0);
		//写入数据
		pStu->id = 123456;
		strcpy(pStu->name,"赵旭阳");
		pStu->age = 22;
		pStu->sex = true;

		//信号量
		sembuf sop = {
			.sem_num = 0,
			.sem_op = 1,
		};
		semop(sem_id,&sop,1);		//V操作
		sop.sem_num = 1;
		sop.sem_op = -1;
		semop(sem_id, &sop, 1);		//P操作

		//删除共享内存地址映射
		shmdt(pStu);
		shmctl(shm_id,IPC_RMID,NULL);

		//删除信号量
		semctl(sem_id, 0, IPC_RMID);
		semctl(sem_id, 1, IPC_RMID);

		int status{};
		wait(&status);
	}
}

消息队列

消息队列提供了一种从一个进程向另一个进程发送一个数据块的方法。每个数据块都被认为含有一个类型，接收进程可以独立地接收含有不同类型的数据结构。我们可以通过发送消息来避免命名管道的同步和阻塞问题。

但消息队列与命名管道一样，每个数据块都有一个最大长度限制。

void 消息队列() {
	pid_t pid = fork();
	if (pid > 0) {
		int msg_id = msgget(ftok(".", 3), IPC_CREAT | 0666);//创建
		printf("int msg_id :%d\n",msg_id);
		MSG msg;
		memset(&msg,0,sizeof(msg));
		while (1) {
			ssize_t ret = msgrcv(msg_id, &msg, sizeof(msg.data), 0, 0); //接收数据
			if (ret != -1) {
				break;
			}
			sleep(1);
			printf("error!");
		}
		printf("id:%d name:%s age:%d message:%s\n",msg.data.id,msg.data.name,msg.data.age,msg.data.message);
		getchar();
		msgctl(msg_id,IPC_RMID,0); //删除

		int status;
		wait(&status);
	}
	else {
		int msg_id = msgget(ftok(".", 3),0666);//创建
		MSG msg;memset(&msg, 0, sizeof(msg));
		msg.type = 1;
		msg.type = 1;
		msg.data.id = 2233;
		msg.data.age = 22;
		strcpy(msg.data.name, "zxy");
		strcpy(msg.data.message, "这是子进程发送过去的消息");

		msgsnd(msg_id,&msg,sizeof(msg.data),0);
		msgctl(msg_id,IPC_RMID,0);

	}
}

多线程并发服务器

//********************
//网络编程+多线程+线程同步的聊天服务器和客户端

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

int clnt_socks[100] = { 0 };
int clnt_cnt{ 0 };
pthread_mutex_t mutex;
char name[64] = {"DEFAULT"};
sem_t semid;

void send_msg(const char* msg, ssize_t str_len) {

	pthread_mutex_lock(&mutex);
	for (int i = 0; i < clnt_cnt; i++) {
		if(clnt_socks[i] >= 0)
			write(clnt_socks[i],msg,str_len);
	}
	pthread_mutex_unlock(&mutex);
}

//线程处理函数
void* handle_cient(void* arg) {
	pthread_detach(pthread_self());
	int client_socket = *((int*)arg);
	char msg[1024];
	ssize_t str_len{ 0 };
	while((str_len = read(client_socket, msg, sizeof(msg))) > 0)
		send_msg(msg,str_len);

	pthread_mutex_lock(&mutex);
	//for (int i = 0; i < clnt_cnt; i++) { //todo
	//	if (client_socket == clnt_socks[i]) {
	//		clnt_socks[i] = -1;
	//		break;
	//	}
	//}
	*((int*)arg) = -1;
	pthread_mutex_unlock(&mutex);
	close(client_socket);
	pthread_exit(NULL);
}

void server98() {
	int server_socket, client_socket;
	struct sockaddr_in server_addr,client_addr;

	int serAddSZ = sizeof(sockaddr);
	socklen_t clientlen = sizeof(client_addr);

	server_socket = socket(PF_INET,SOCK_STREAM,0);
	memset(&server_addr,0,serAddSZ);
	server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
	server_addr.sin_family = AF_INET;
	server_addr.sin_port = htons(2233);

	pthread_mutex_init(&mutex,NULL);

	if (bind(server_socket, (struct sockaddr*)&server_addr, serAddSZ) == -1) {
		printf("%s(%d)%s server bind failed:%s\n", __FILE__, __LINE__, __FUNCTION__,strerror(errno));
		return;
	}
	if (listen(server_socket, 5) == -1) {
		printf("%s(%d)%s server listen failed:%s\n", __FILE__, __LINE__, __FUNCTION__, strerror(errno));
		return;
	}
	while (1) {
		client_socket = accept(server_socket, (struct sockaddr*)&client_addr,&clientlen);
		if (client_socket == -1) {
			printf("%s(%d)%s server accept client failed:%s\n", __FILE__, __LINE__, __FUNCTION__, strerror(errno));
			break;
		}

		pthread_mutex_lock(&mutex);
		clnt_socks[clnt_cnt++] = client_socket;
		pthread_mutex_unlock(&mutex);

		//客户端处理交给线程处理
		pthread_t tid;
		pthread_create(&tid,NULL,handle_cient,&clnt_socks[clnt_cnt-1]);
	}

	close(server_socket);
	pthread_mutex_destroy(&mutex);
}

void* client_send_msg(void* arg) {
	pthread_detach(pthread_self());

	int sock = *((int*)arg);
	char msg[256]{ "" };
	char buffer[1024]{""};
	while (1) {
		memset(buffer, 0, sizeof(buffer));
		fgets(msg,sizeof(msg),stdin);
		if ((strcmp(msg, "q\n") == 0) || (strcmp(msg, "Q\n") == 0)) {
			break;
		}
		snprintf(buffer,sizeof(buffer),"%s %s",name,msg);
		write(sock, buffer,strlen(buffer));
	}
	sem_post(&semid);
	pthread_exit(NULL);
}

void* client_recv_msg(void* arg) {
	pthread_detach(pthread_self());

	int sock = *((int*)arg);
	char msg[256]{ "" };
	while (1) {
		ssize_t str_len = read(sock, msg, sizeof(msg));
		if (str_len <= 0)
			break;
		fputs(msg,stdout);
	}

	sem_post(&semid);
	pthread_exit(NULL);
}



void client98() {
	int socket_client = socket(PF_INET,SOCK_STREAM,0);
	struct sockaddr_in server_addr;
	memset(&server_addr, 0, sizeof(server_addr));
	server_addr.sin_addr.s_addr = inet_addr("127.0.0.1");
	server_addr.sin_family = AF_INET;
	server_addr.sin_port = htons(2233);

	if (connect(socket_client, (struct sockaddr*)&server_addr, sizeof(server_addr)) == -1) {
		printf("%s(%d)%s client connect failed:%s\n", __FILE__, __LINE__, __FUNCTION__, strerror(errno));
		return;
	}
	//消息发送 消息接收
	pthread_t send_thread_id,recv_thread_id;

	sem_init(&semid,0,-1);

	pthread_create(&send_thread_id,NULL,client_send_msg,(void*)&socket_client);
	pthread_create(&recv_thread_id,NULL,client_recv_msg,(void*)&socket_client);

	sem_wait(&semid);
	close(socket_client);
}

void 多线程并发服务器(const char* arg) {
	if (strcmp(arg, "s") == 0) {
		server98();
	}
	else {
		client98();
	}
}

int main(int argc, char* argv[]) {
	多线程并发服务器(argv[1]);
	return 0;
}

Select(io复用)并发服务器

//select 实现I\O复用服务器
#include<sys/select.h>
#include<sys/times.h>

void server() {
	int server_socket, client_socket;
	struct sockaddr_in server_addr,client_addr;
	socklen_t client_size = sizeof(client_addr);
	socklen_t server_size = sizeof(server_addr);
	memset(&server_addr,0,server_size);
	server_addr.sin_family = AF_INET;
	server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
	server_addr.sin_port = htons(2233);

	server_socket = socket(PF_INET,SOCK_STREAM,0);

	if ((bind(server_socket, (struct sockaddr*)&server_addr, server_size)) == -1) {
		printf("%s(%d)%s server bind failed:%s\n", __FILE__, __LINE__, __FUNCTION__, strerror(errno));
		close(server_socket);
		return;
	}

	if ((listen(server_socket, 5)) == -1) {
		printf("%s(%d)%s server listen failed:%s\n", __FILE__, __LINE__, __FUNCTION__, strerror(errno));
		close(server_socket);
		return;
	}

	fd_set reads,copy_reads;
	FD_ZERO(&reads);
	FD_SET(server_socket,&reads);
	timeval timeout = {0,500000};
	int max_sock = server_socket;

	while (1) {
		copy_reads = reads;
		int fd_num = select(max_sock + 1, &copy_reads, NULL, NULL, &timeout);
		if (fd_num == -1) {
			printf("%s(%d)%s server select failed:%s\n", __FILE__, __LINE__, __FUNCTION__, strerror(errno));
			close(server_socket);
			return;
		}

		if (fd_num == 0) continue;

		for (int i = 0; i < max_sock +1; i++) {
			if (FD_ISSET(i, &copy_reads)) {
				if (i == server_socket) {
					client_socket = accept(server_socket,(struct sockaddr*)&client_addr,&client_size);
					FD_SET(client_socket,&reads);
					if (server_socket < client_socket) {
						max_sock = client_socket;
					}
					printf("%s(%d)%s client is connected:%d\n", __FILE__, __LINE__, __FUNCTION__,client_socket);
				}
				else {
					char buffer[256]{""};
					ssize_t strlen = read(i,buffer,sizeof(buffer));
					if (strlen == 0) {
						FD_CLR(i,&reads);
						close(i);
						printf("%s(%d)%s client is disconnect:%s\n", __FILE__, __LINE__, __FUNCTION__, i);
					}
					else {
						write(i,buffer,strlen);
					}
				}
			}
		}
	}
	close(server_socket);
}

void client() {
	int client_socket = socket(PF_INET,SOCK_STREAM,0);
	struct sockaddr_in server_addr;
	socklen_t server_size = sizeof(server_addr);
	memset(&server_addr, 0, server_size);
	server_addr.sin_family = AF_INET;
	server_addr.sin_addr.s_addr = inet_addr("127.0.0.1");
	server_addr.sin_port = htons(2233);

	if ((connect(client_socket, (struct sockaddr*)&server_addr, server_size)) == -1) {
		printf("%s(%d)%s client connect error!\n", __FILE__, __LINE__, __FUNCTION__);
		close(client_socket);
		return;
	}
	char message[256]{""};
	while (true) {
		printf("Input message(q/Q):");
		fgets(message,sizeof(message),stdin);
		if (strcmp(message, "q\n") == 0 || strcmp(message, "Q\n") == 0) break;

		write(client_socket, message, strlen(message));
		memset(message, 0, strlen(message));
		read(client_socket, message, sizeof(message));
		printf("server:%s\n", message);
	}
	close(client_socket);
}


int main(int argc, char* argv[]) {
	
	if (strcmp(argv[1], "s") == 0) {
		server();
	}
	else {
		client();
	}

	return 0;
}

Epoll 服务器

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

void server() {
	int server_socket, client_socket;
	struct sockaddr_in server_addr, client_addr;
	socklen_t client_size = sizeof(client_addr);
	socklen_t server_size = sizeof(server_addr);
	memset(&server_addr, 0, server_size);
	server_addr.sin_family = AF_INET;
	server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
	server_addr.sin_port = htons(2233);
	char buffer[256]{""};

	server_socket = socket(PF_INET, SOCK_STREAM, 0);

	if ((bind(server_socket, (struct sockaddr*)&server_addr, server_size)) == -1) {
		printf("%s(%d)%s server bind failed:%s\n", __FILE__, __LINE__, __FUNCTION__, strerror(errno));
		close(server_socket);
		return;
	}

	if ((listen(server_socket, 5)) == -1) {
		printf("%s(%d)%s server listen failed:%s\n", __FILE__, __LINE__, __FUNCTION__, strerror(errno));
		close(server_socket);
		return;
	}
	//epoll
	epoll_event event; //事件
	int epfd, event_cnt;
	epfd = epoll_create(2048);
	if (epfd == -1) {
		printf("%s(%d)%s epoll create failed:%s\n", __FILE__, __LINE__, __FUNCTION__, strerror(errno));
		close(server_socket);
		return;
	}
	epoll_event* all_events = new epoll_event[100];

	event.events = EPOLLIN;
	event.data.fd = server_socket;
	epoll_ctl(epfd,EPOLL_CTL_ADD,server_socket,&event);
	while (true) {
		event_cnt = epoll_wait(epfd,all_events,100,1000);
		if (event_cnt == -1) {
			printf("%s(%d)%s epoll_wait error:%s\n", __FILE__, __LINE__, __FUNCTION__, strerror(errno));
			break;
		}
		if (event_cnt == 0) continue;

		for (int i = 0; i < event_cnt; i++) {
			if (all_events[i].data.fd == server_socket) {
				client_socket = accept(server_socket,(struct sockaddr*)&client_socket,&client_size);
				event.events = EPOLLIN;
				event.data.fd = client_socket;
				epoll_ctl(epfd,EPOLL_CTL_ADD,client_socket,&event);
				printf("%s(%d)%s client is connected!:%s\n", __FILE__, __LINE__, __FUNCTION__, strerror(errno));
			}
			else {
				ssize_t len = read(all_events[i].data.fd,buffer,sizeof(buffer));
				if (len <= 0) {
					epoll_ctl(epfd,EPOLL_CTL_DEL,all_events[i].data.fd,NULL);
					close(client_socket);
					printf("%s(%d)%s client is closed!\n", __FILE__, __LINE__, __FUNCTION__);

				}
				else {
					write(all_events[i].data.fd,buffer,len);
				}
			}
		}
	}

	delete[] all_events;
	close(server_socket);
	close(epfd);
}

void client() {
	int client_socket = socket(PF_INET, SOCK_STREAM, 0);
	struct sockaddr_in server_addr;
	socklen_t server_size = sizeof(server_addr);
	memset(&server_addr, 0, server_size);
	server_addr.sin_family = AF_INET;
	server_addr.sin_addr.s_addr = inet_addr("127.0.0.1");
	server_addr.sin_port = htons(2233);

	if ((connect(client_socket, (struct sockaddr*)&server_addr, server_size)) == -1) {
		printf("%s(%d)%s client connect error!\n", __FILE__, __LINE__, __FUNCTION__);
		close(client_socket);
		return;
	}
	char message[256]{ "" };
	while (true) {
		printf("Input message(q/Q):");
		fgets(message, sizeof(message), stdin);
		if (strcmp(message, "q\n") == 0 || strcmp(message, "Q\n") == 0) break;

		write(client_socket, message, strlen(message));
		memset(message, 0, strlen(message));
		read(client_socket, message, sizeof(message));
		printf("server:%s\n", message);
	}
	close(client_socket);
}


int main(int argc, char* argv[]) {
	
	if (strcmp(argv[1], "s") == 0) {
		server();
	}
	else {
		client();
	}

	return 0;
}

site logo

zxyの神秘小屋

© 2025 zxyの神秘小屋