建公司网站要多少钱,网站设计团队发展,泰安飞讯网络有限公司,图库 网站 源码线程池复习 线程池的好处#xff1a;所有的池都是为了事先把资源准备好#xff0c;在后续用的时候可以更加方便的拿到这个资源——不用去申请、释放资源 什么时候用线程池#xff1f; IO事务并发较高#xff1a;人在杭州#xff0c;但是数据库在北京#xff0c;想要…线程池复习 线程池的好处所有的池都是为了事先把资源准备好在后续用的时候可以更加方便的拿到这个资源——不用去申请、释放资源 什么时候用线程池 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文件描述符// 创建socketAF_INET表示使用IPv4协议SOCK_STREAM表示使用TCP协议// 成功时返回socket描述符失败时返回-1if ((sockfd socket(AF_INET, SOCK_STREAM, 0)) 0) {return -1; // 创建socket失败返回-1}struct sockaddr_in addr; // 定义IPV4的sock地址结构addr.sin_family AF_INET; // 指定地址家族为IPv4addr.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(YELLOWpush taskQueue is full.\nNONE);pthread_mutex_unlock(taskQueue-mutex); // 解锁后返回return ;}taskQueue-data[taskQueue-tail] data; // 将任务添加到队尾DBG(PINKpush push to %dth task.\nNONE, taskQueue-tail);taskQueue-total; // 更新队列中的任务总数if (taskQueue-tail taskQueue-size) { // 如果队尾指针到达队列末尾重置为0DBG(PINKpush tail begins with 0.\nNONE);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(BLUEpop pop data from %dth task.\nNONE, taskQueue-head);taskQueue-total--; // 更新队列中的任务总数if (taskQueue-head taskQueue-size) { // 如果队头指针到达队列末尾重置为0DBG(PINKpop head begins with 0.\nNONE);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(GREENthread got a task! pop the data %sNONE, (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(GREENconnect to client , server_listen fd %d.\nNONE, 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(GREENSET %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(CYANclients[%d] %d.\nNONE, 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(CYANclients[%d], rsize %d.\nNONE, 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************************************************************************/#includehead.h // 包含头文件这个头文件可能包含了程序需要的其他库文件、宏定义、函数声明等
#define INS 5 // 定义常量INS表示创建的线程数为5
#define MAX 100 // 定义常量MAX表示缓冲区可以存储的最大行数为100int main () {FILE *fp; // 定义文件指针fp用于打开和读取文件pthread_t tid[INS]; // 定义线程ID数组数组大小为INS用于存储线程IDchar 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为0if ((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文件的生成
shell命令while [[ 1 ]] do
cat test.c a.txt
donectrl 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}; // 存储客户端套接字的数组初始化为0char 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(GREENconnect to client , server_listen fd %d.\nNONE, 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(GREENSET %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(CYANclients[%d] %d.\nNONE, 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(CYANclients[%d], rsize %d.\nNONE, clients[i], rsize);if (rsize 0) { // 如果接收到的数据大小小于等于0表示客户端关闭连接或出错close(clients[i]); // 关闭套接字clients[i] -1; // 标记客户端已关闭} else {task_queue_push(taskQueue, buff[i]); // 将接收到的数据加入任务队列}if (--ret 0) break; // 如果已处理完所有活动的文件描述符退出循环}} }
}
