【高并发服务器 02】——线程池与IO多路复用

线程池复习

• 线程池的好处：所有的池都是为了事先把资源准备好，在后续用的时候可以更加方便的拿到这个资源——不用去申请、释放资源

• 什么时候用线程池？

  • IO&事务并发较高：人在杭州，但是数据库在北京，想要查询数据库，需要通过互联网建立TCP三次握手，频繁地创建和销毁线程意味着频繁地进行TCP三次握手，四次挥手。如果实际查询数据库的时间不到建立连接时间的一般甚至更少。那么这种代价是我们不能忍受的。而池化技术，就是这样一种，创建好了线程，完成了TCP连接，不需要频繁创建和释放资源的技术。

• 线程的数量多少合适？

  • 小于等于CPU的核心数：我们知道线程是CPU调度的基本单位，调度的目的就是使他能够在CPU上运行，同一时间最多线程能够运行的个数取决于CPU的核心数。

  • IO密集型：进程来了之后就做一件事情——申请1次IO。就像我们去银行取钱，你要做的无非就是让柜台给我取钱。我们处在上层的应用发起一次请求，交给内核下层去做，内核做的过程中，有这种IO管理上的通道技术、IO控制器，所以并非是CPU在处理拿数据。如果线程是处于阻塞等的状态，那么这个线程就暂时处于失联了的状态，这个时候线程数量过少了也不行。但是如果是异步IO，非阻塞，这个时候就跟CPU密集型是一样的。因为不会在IO上发生等待。

  • CPU密集型：比如计算一个非常复杂的式子，很多个线程协同计算这件事情，CPU有多少潜能都能被榨干。这个时候，线程的数量一定不能大于CPU核心数。

• 线程池的实现：

  • 任务队列：队列的定义、初始化、push、pop、销毁
  • 一组线程：线程处理函数

• 死锁：两个或两个以上的进程因为推进顺序不当或资源数量受限而导致的互相等待的情况叫做死锁。

• 惊群效应：pthread_cond_wait这条语句之前必须上锁。执行到它时会首先把锁给解开，然后进去睡眠等待，直到有一个信号来唤醒它，唤醒了之后又去抢锁，这个时候，有一堆人抢锁，第一个抢到锁的人，率先把资源给吃了，接着把锁释放，剩下的线程继续抢锁，抢到锁之后发现队列还是空的，于是又把锁丢掉，进入了睡眠直到下一个信号把它唤醒。

编程技巧

• #ifdef宏定义

  • #ifdef是一个预处理指令，用于在编译时根据条件判断是否编译特定的代码块。它的作用是根据条件判断编译不同的代码，可以用来实现条件编译。当条件为真时，编译#ifdef和#endif之间的代码；当条件为假时，忽略#ifdef和#endif之间的代码。这样可以根据不同的编译条件，在不同的环境中编译不同的代码，实现对不同平台、不同需求的适配。

  • #ifndef 的作用就是确保只定义一次

#ifndef _HEAD_H
#define _HEAD_H
#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include "thread_pool.h"
#include "color.h"
#include "common.h" // 自己的库写在后面，每新写一个库就往后面放，确保引用顺序正确
#endif

• 编译调试命令：gcc test.c -I common/ -D _D

  • -I common/：这个选项告诉编译器在哪里查找头文件。编译器在编译程序时，除了查找标准的头文件位置外，还会在这里指定的目录中查找。在这个例子中，编译器会在名为common/的目录下查找头文件。这个目录通常包含了一些公共的或者共享的头文件。

  • -D _D：这个选项用于定义宏。-D选项后面跟着的是宏的名称（在这个例子中是_D），它相当于在源代码的最开始添加了一行#define _D。这样，你就可以在代码中通过预处理指令#ifdef、#ifndef或#if defined等来检查这个宏是否被定义，从而使得代码可以根据宏的定义与否来选择性地编译。这在条件编译中非常有用。

#ifdef _D
#define DBG(fmt, args...) printf(fmt, ##args)
#else
#define DBG(fmt, args...)
#endif

• 颜色调试
  我们希望在debug输出的时候能够以彩色的形式输出更多更丰富的类别信息。

#ifndef _COLOR_H
#define _COLOR_H

#define NONE      "\e[0m"     // 清除颜色，即之后的打印为正常输出，之前的不受影响
#define BLACK     "\e[0;30m"  // 深黑
#define L_BLACK   "\e[1;30m"  // 亮黑，偏灰褐
#define RED       "\e[0;31m"  // 深红
#define L_RED     "\e[1;31m"  // 鲜红
#define GREEN     "\e[0;32m"  // 深绿
#define L_GREEN   "\e[1;32m"  // 鲜绿色
#define BROWN     "\e[0;33m"  // 深黄
#define YELLOW    "\e[1;33m"  // 鲜黄
#define BLUE      "\e[0;34m"  // 深蓝
#define L_BLUE    "\e[1;34m"  // 亮蓝
#define PINK      "\e[0;35m"  // 深粉
#define L_PINK    "\e[1;35m"  // 亮粉
#define CYAN      "\e[0;36m"  // 暗青色
#define L_CYAN    "\e[1;36m"  // 亮青色
#define GRAY      "\e[0;37m"  // 灰色
#define WHITE     "\e[1;37m"  // 白色，字体比正常大，比bold小
#define BOLD      "\e[1m"     // 白色，粗体
#define UNDERLINE "\e[4m"     // 下划线，白色，正常大小
#define BLINK     "\e[5m"     // 闪烁，白色，正常大小
#define REVERSE   "\e[7m"     // 反转，即字体背景为白色，字体为黑色
#define HIDE      "\e[8m"     // 隐藏
#define CLEAR     "\e[2J"     // 清除
#define CLRLINE   "\e[K"      // 清除行

#endif

代码实现：

/*************************************************************************
	> File Name: common.h
	> Author: jby
	> Mail: 
	> Created Time: Thu 21 Mar 2024 09:01:03 AM CST
 ************************************************************************/

#ifndef _COMMON_H // 如果没有定义_COMMON_H，那么执行下面的代码
#define _COMMON_H // 定义_COMMON_H，防止头文件被重复包含

// 定义一个全局数组，用于存储配置文件的答案或其他用途
char conf_ans[512];

// 声明socket_create函数，该函数用于创建一个监听socket
// 参数port是要绑定的端口号
int socket_create(int port);

// 声明socket_connect函数，该函数用于创建一个连接到指定IP和端口的socket
// 参数ip是要连接的目标IP地址，port是目标端口号
int socket_connect(const char *ip, int port);

// 声明make_block函数，该函数用于将指定的文件描述符设置为阻塞模式
// 参数fd是要设置的文件描述符
int make_block(int fd);

// 声明make_nonblock函数，该函数用于将指定的文件描述符设置为非阻塞模式
// 参数fd是要设置的文件描述符
int make_nonblock(int fd);

#endif // 结束条件编译，与#ifndef对应

/*************************************************************************
	> File Name: head.h
	> Author: jby
	> Mail: 
	> Created Time: Thu 21 Mar 2024 08:56:01 AM CST
 ************************************************************************/

#ifndef _HEAD_H
#define _HEAD_H
#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <unistd.h>
#include <fcntl.h>
#include <sys/epoll.h>
#include <arpa/inet.h>
#include <sys/select.h>
#include <sys/time.h>
#include "thread_pool.h"
#include "color.h"
#include "common.h"
#ifdef _D
#define DBG(fmt, args...) printf(fmt, ##args)
#else
#define DBG(fmt, args...)
#endif
#endif

/*************************************************************************
	> File Name: common.c
	> Author: jby
	> Mail: 
	> Created Time: Thu 21 Mar 2024 08:56:39 AM CST
 ************************************************************************/

#include "head.h" // 引入相关的头文件

// 创建一个监听socket的函数
int socket_create(int port) {
	int sockfd; // 用于存储socket文件描述符

	// 创建socket，AF_INET表示使用IPv4协议，SOCK_STREAM表示使用TCP协议
	// 成功时返回socket描述符，失败时返回-1
	if ((sockfd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
		return -1; // 创建socket失败，返回-1
	}

	struct sockaddr_in addr; // 定义IPV4的sock地址结构
	addr.sin_family = AF_INET; // 指定地址家族为IPv4
	addr.sin_port = htons(port); // 指定端口号，并使用htons函数将主机字节顺序转换为网络字节顺序
	addr.sin_addr.s_addr = inet_addr("0.0.0.0"); // 指定接收所有IP地址的连接

	int reuse = 1; // 定义重用标志
	// 设置socket选项，允许重用本地地址和端口，这对于防止“地址已在使用”错误很有帮助
	setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, (char *)&reuse, sizeof(int));

	// 将socket绑定到指定的IP地址和端口
	if (bind(sockfd, (struct sockaddr *)&addr, sizeof(addr)) < 0) {
		return -1; // 绑定失败，返回-1
	}

	// 开始监听，第二个参数为backlog，表示内核监听队列的最大长度
	if (listen(sockfd, 8) < 0) {
		return -1; // 监听失败，返回-1
	}

	return sockfd; // 成功创建并初始化socket，返回socket文件描述符
}

/*************************************************************************
	> File Name: thread_pool.c
	> Author: jby
	> Mail: 
	> Created Time: Thu 21 Mar 2024 09:23:56 AM CST
 ************************************************************************/

#include "head.h" // 包含自定义头文件，可能包含所需的库文件和宏定义

// 初始化任务队列
void task_queue_init(struct task_queue* taskQueue, int size) {
	taskQueue->size = size; // 设置队列大小
	taskQueue->total = taskQueue->head = taskQueue->tail = 0; // 初始化队列的状态
	taskQueue->data = calloc(sizeof(void *), size); // 分配存储任务的内存空间
	pthread_mutex_init(&taskQueue->mutex, NULL); // 初始化互斥锁
	pthread_cond_init(&taskQueue->cond, NULL); // 初始化条件变量
	return ;
}

// 向任务队列中添加任务
void task_queue_push(struct task_queue *taskQueue, void *data) {
	pthread_mutex_lock(&taskQueue->mutex); // 加锁保护队列状态
	if (taskQueue->total == taskQueue->size) { // 队列已满，无法添加
		DBG(YELLOW"<push> taskQueue is full.\n"NONE);
		pthread_mutex_unlock(&taskQueue->mutex); // 解锁后返回
		return ;
	}
	taskQueue->data[taskQueue->tail] = data; // 将任务添加到队尾
	DBG(PINK"<push> push to %dth task.\n"NONE, taskQueue->tail);
	taskQueue->total++; // 更新队列中的任务总数
	if (++taskQueue->tail == taskQueue->size) { // 如果队尾指针到达队列末尾，重置为0
		DBG(PINK"<push> tail begins with 0.\n"NONE);
		taskQueue->tail = 0;
	}
	pthread_cond_signal(&taskQueue->cond); // 通知等待的线程有新任务添加
	pthread_mutex_unlock(&taskQueue->mutex); // 解锁
}

// 从任务队列中取出任务
void* task_queue_pop(struct task_queue *taskQueue) {
	pthread_mutex_lock(&taskQueue->mutex); // 加锁
	while (taskQueue->total == 0) { // 如果队列为空，则等待
		pthread_cond_wait(&taskQueue->cond, &taskQueue->mutex);
	}
	void *data = taskQueue->data[taskQueue->head]; // 从队头取出任务
	DBG(BLUE"<pop> pop data from %dth task.\n"NONE, taskQueue->head);
	taskQueue->total--; // 更新队列中的任务总数
	if (++taskQueue->head == taskQueue->size) { // 如果队头指针到达队列末尾，重置为0
		DBG(PINK"<pop> head begins with 0.\n"NONE);
		taskQueue->head = 0;
	}
	pthread_mutex_unlock(&taskQueue->mutex); // 解锁
	return data; // 返回取出的任务数据
}

// 线程执行函数
void* thread_run(void *arg) {
	pthread_detach(pthread_self()); // 设置线程为分离状态，结束时自动释放资源
	struct task_queue *taskQueue = (struct task_queue *)arg; // 获取任务队列
	while (1) {
		void* data = task_queue_pop(taskQueue); // 从任务队列中取出任务
		DBG(GREEN"<thread> got a task! pop the data %s"NONE, (char *)data);
		// 在这里处理任务，示例代码中仅打印消息
	}
}

/*************************************************************************
	> File Name: thread_pool.h
	> Author: jby
	> Mail: 
	> Created Time: Thu 21 Mar 2024 09:24:02 AM CST
 ************************************************************************/

#ifndef _THREAD_POOL_H // 预处理指令，防止头文件被多次重复包含
#define _THREAD_POOL_H

// 定义任务队列的结构体
struct task_queue {
	int head, tail, size, total; // 分别表示队列的头部、尾部、大小和当前任务总数
	void **data; // 指向任务数据的指针数组
	pthread_mutex_t mutex; // 互斥锁，用于同步对任务队列的访问
	pthread_cond_t cond; // 条件变量，用于线程间的同步
};

// 声明任务队列初始化函数
void task_queue_init(struct task_queue *taskQueue, int size);

// 声明任务队列添加任务函数
void task_queue_push(struct task_queue *taskQueue, void *data);

// 声明从任务队列中取出任务的函数
void *task_queue_pop(struct task_queue *taskQueue);

// 声明线程运行函数
void *thread_run(void *arg);

#endif // 结束预处理指令

/*************************************************************************
	> File Name: 2.select.c
	> Author: jby
	> Mail: 
	> Created Time: Thu 21 Mar 2024 09:41:51 PM CST
 ************************************************************************/


#include "head.h"
#define MAX 100
#define INS 5
int main (int argc, char **argv) {
	if (argc < 2) {
		fprintf(stderr, "Usage: %s port", argv[0]);
		exit(1);
	}
	int server_listen, port, sockfd;
	pthread_t tid[INS];
	int clients[MAX] = {0};
	char buff[MAX][1024];
	struct task_queue *taskQueue = (struct task_queue *)malloc(sizeof(struct task_queue)); 
	task_queue_init(taskQueue, MAX);
	for (int i = 0; i < INS; i++) {
		pthread_create(&tid[i], NULL, thread_run, (void *)taskQueue);
	}
	port = atoi(argv[1]);
	if ((server_listen = socket_create(port)) < 0) {
		perror("server_port");
		exit(1);
	}
	DBG(GREEN"connect to client , server_listen fd = %d.\n"NONE, server_listen);
	fd_set rfds;
	int max_fd;
	max_fd = server_listen;
	clients[server_listen] = server_listen;
	while (1) {
		FD_ZERO(&rfds);
		FD_SET(server_listen, &rfds);
		for (int i = 3; i < max_fd + 1; i++) { // 每次都要重新注册
			if (clients[i] == -1) continue;
			FD_SET(clients[i], &rfds);
			DBG(GREEN"SET %d in rfds.\n", clients[i]);
		}
		int ret = select(max_fd + 1, &rfds, NULL, NULL, NULL); 
		if (ret < 0) {
			perror("select");
			exit(1);
		}
		if (FD_ISSET(server_listen, &rfds)) {
			if ((sockfd = accept(server_listen, NULL, NULL)) < 0) {
				perror("accpet");
				exit(1);
			}
			if (sockfd > max_fd) max_fd = sockfd;
			ret--;
			clients[sockfd] = sockfd; // 文件描述符每次选最小的数字
			DBG(CYAN"clients[%d] = %d.\n"NONE, sockfd, clients[sockfd]);
		}
		for (int i = 0; i < max_fd + 1; i++) {
			if (clients[i] == server_listen) continue;
			if (FD_ISSET(clients[i], &rfds)) {
				int rsize = recv(clients[i], buff[i], 1024, 0);
				DBG(CYAN"clients[%d], rsize = %d.\n"NONE, clients[i], rsize);
				if (rsize <= 0) {
					close(clients[i]);
					clients[i] = -1;
				} else {
					task_queue_push(taskQueue, buff[i]);
				}
				if (--ret == 0) break; // 提高效率，避免全部遍历
			}
		}		
	}
}

/*************************************************************************
	> File Name: thread_pool test.c
	> Author: jby
	> Mail: 
	> Created Time: Thu 21 Mar 2024 09:09:44 AM CST
 ************************************************************************/
 
#include"head.h" // 包含头文件，这个头文件可能包含了程序需要的其他库文件、宏定义、函数声明等
#define INS 5 // 定义常量INS，表示创建的线程数为5
#define MAX 100 // 定义常量MAX，表示缓冲区可以存储的最大行数为100

int main () {
	FILE *fp; // 定义文件指针fp，用于打开和读取文件
	pthread_t tid[INS]; // 定义线程ID数组，数组大小为INS，用于存储线程ID
	char buff[MAX][1024]; // 定义一个二维字符数组buff，用于存储从文件中读取的每行数据
	struct task_queue* taskQueue = (struct task_queue *)malloc(sizeof(struct task_queue)); // 动态分配任务队列结构体的内存
	task_queue_init(taskQueue, MAX); // 初始化任务队列

	for (int i = 0; i < INS; i++) { // 循环创建INS个线程
		pthread_create(&tid[i], NULL, thread_run, (void *)taskQueue); // 创建线程，线程执行的函数为thread_run，参数为任务队列的指针
	}
	int sub = 0; // 定义变量sub，用于记录当前读取到buff数组的哪一行
	while (1) { // 无限循环，不断从文件读取数据并处理
		int sub = 0; // 每次循环开始时重置sub为0
		if ((fp = fopen("./a.txt", "r")) == NULL) { // 尝试打开文件a.txt，如果失败，则打印错误信息并退出程序
			perror("fopen");
			exit(1);
		}
		while (fgets(buff[sub], 1024, fp) != NULL) { // 从文件中读取一行，存储到buff[sub]中，直到文件结束
			task_queue_push(taskQueue, buff[sub]); // 将读取到的行数据推送到任务队列中
			// sleep(1); // 可以根据需要取消注释，使程序在每次推送后暂停1秒
			if (++sub == MAX) sub = 0; // 如果sub达到MAX，则重置为0，实现循环使用buff数组
		}
		fclose(fp); // 关闭文件
	}
    return 0; // 程序正常退出
}

其中，a.txt文件的生成：文章来源地址https://www.toymoban.com/news/detail-842638.html

shell命令：

while [[ 1 ]] do
cat test.c > a.txt
done

ctrl + c停止

/*************************************************************************
	> File Name: 2.select.c
	> Author: jby
	> Mail: 
	> Created Time: Thu 21 Mar 2024 09:41:51 PM CST
 ************************************************************************/

#include "head.h" // 包含自定义的头文件，可能包括必要的库文件和宏定义
#define MAX 100 // 定义最大客户端数量
#define INS 5 // 定义线程池中线程的数量
int main (int argc, char **argv) {
	if (argc < 2) { // 检查命令行参数数量，确保提供了端口号
		fprintf(stderr, "Usage: %s port", argv[0]); // 如果没有提供端口号，打印使用方法
		exit(1); // 退出程序
	}
	int server_listen, port, sockfd; // 分别用于存储监听套接字、端口号和客户端套接字
	pthread_t tid[INS]; // 存储线程ID的数组
	int clients[MAX] = {0}; // 存储客户端套接字的数组，初始化为0
	char buff[MAX][1024]; // 存储接收数据的缓冲区
	struct task_queue *taskQueue = (struct task_queue *)malloc(sizeof(struct task_queue)); 
	task_queue_init(taskQueue, MAX); // 初始化任务队列
	for (int i = 0; i < INS; i++) {
		pthread_create(&tid[i], NULL, thread_run, (void *)taskQueue); // 创建处理任务的线程
	}
	port = atoi(argv[1]); // 从命令行参数获取端口号
	if ((server_listen = socket_create(port)) < 0) { // 创建监听套接字
		perror("server_port"); // 如果创建失败，打印错误信息
		exit(1); // 退出程序
	}
	DBG(GREEN"connect to client , server_listen fd = %d.\n"NONE, server_listen); // 打印监听套接字的文件描述符
	fd_set rfds; // 定义文件描述符集合
	int max_fd; // 存储文件描述符的最大值
	max_fd = server_listen; // 初始化最大文件描述符为监听套接字
	clients[server_listen] = server_listen; // 将监听套接字加入客户端数组
	while (1) {
		FD_ZERO(&rfds); // 清空文件描述符集合
		FD_SET(server_listen, &rfds); // 将监听套接字加入集合
		for (int i = 3; i < max_fd + 1; i++) { // 遍历所有可能的文件描述符
			if (clients[i] == -1) continue; // 如果客户端已关闭，跳过
			FD_SET(clients[i], &rfds); // 将活跃的客户端套接字加入集合
			DBG(GREEN"SET %d in rfds.\n", clients[i]);
		}
		int ret = select(max_fd + 1, &rfds, NULL, NULL, NULL); // 调用select等待活动的文件描述符
		if (ret < 0) {
			perror("select"); // 如果select调用失败，打印错误信息
			exit(1); // 退出程序
		}
		if (FD_ISSET(server_listen, &rfds)) { // 检查监听套接字是否有新的连接请求
			if ((sockfd = accept(server_listen, NULL, NULL)) < 0) {
				perror("accpet"); // 如果接受连接失败，打印错误信息
				exit(1); // 退出程序
			}
			if (sockfd > max_fd) max_fd = sockfd; // 更新最大文件描述符
			ret--; // 减少待处理的文件描述符数量
			clients[sockfd] = sockfd; // 将新客户端的套接字加入数组
			DBG(CYAN"clients[%d] = %d.\n"NONE, sockfd, clients[sockfd]);
		}
		for (int i = 0; i < max_fd + 1; i++) { // 遍历所有文件描述符处理数据
			if (clients[i] == server_listen) continue; // 跳过监听套接字
			if (FD_ISSET(clients[i], &rfds)) { // 检查该文件描述符是否有数据可读
				int rsize = recv(clients[i], buff[i], 1024, 0); // 接收数据
				DBG(CYAN"clients[%d], rsize = %d.\n"NONE, clients[i], rsize);
				if (rsize <= 0) { // 如果接收到的数据大小小于等于0，表示客户端关闭连接或出错
					close(clients[i]); // 关闭套接字
					clients[i] = -1; // 标记客户端已关闭
				} else {
					task_queue_push(taskQueue, buff[i]); // 将接收到的数据加入任务队列
				}
				if (--ret == 0) break; // 如果已处理完所有活动的文件描述符，退出循环
			}
		}		
	}
}

到了这里，关于【高并发服务器 02】——线程池与IO多路复用的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！