目录
版本四：线程池
注意事项
文件：Task.hpp -- 任务单独为一个文件
组件：日志修改
新函数：vprintf()
可变参数的提取逻辑
vfprintf()的工作原理
初始化一个va_list
日志准备
获取时间小知识
日志初版
日志启动测试
TCP通用服务器(守护进程) *
新指令1：jobs -- 查看进程作业
新指令2：fg -- foreground(前台)
新指令3：bg(Ctrl + Z) -- background(后台)
创建新的会话
新接口2：daemon() -- 选用(本文不用)
新接口3：setsid() *
小细节1：null文件 -- 任何请求都可以被接收，但是都会被丢弃
接口1：open() -- 打开文件
接口2：dup2 -- 重定向
接口3：chdir() -- 进程执行路径更改(选填)
文件：daemon.hpp(守护进程)
测试：守护进程
测试服务器的正常回显
日志修改：日志写入文件中保存
全部代码文件
daemon.hpp -- 守护进程
log.hpp -- 日志文件
makefile
tcpClient.cc -- 客户端1
tcpClient.hpp -- 客户端2
tcpServer.cc -- 服务端1
tcpServer.hpp -- 服务端2
Task.hpp -- 形成任务
Thread.hpp -- 线程池1
ThreadPool.hpp -- 线程池2
LockGuard.hpp -- 线程池(加锁部分)
接上文：
版本四：线程池
至此，多线程与多进程的版本完成，但是由于无论是线程的创建还是进程的创建，都是事情到来的时候(链接到来的时候)才创建任务，所以会存在一个频繁创建的问题。还有一个问题，就是客户有多少，就需要多少个进程或者线程，那一个服务器的效率就不会很高，应付几十、几百可以，但是一但多一点就不行了，为此这里就引入一个基于管道式的线程池的组件来实现第四个版本
一个主线程打开的文件，新线程是可以看到的，也就是说线程池对于资源共享有天然的优势，于是可以把新链接构成一个新任务，把任务传递给线程池来统一执行
注意这里就直接引入一个线程池组件进行使用，就不再讲解了
关于线程池的介绍可以参考文章 -- 待更新
这里的线程池是一个单例模式
注意事项
注意线程池的版本对于代码是有整体修改的，所以下面的代码会进行一定程度的整理和修改
因为这里的serviceIO服务和类里面是没有任何关系的，所以代码可以直接拿出去，将其放入Task.hpp文件中去
文件：Task.hpp -- 任务单独为一个文件
#pragma once
#include <iostream>
#include <string>
#include <cstdio>
#include <functional>
// 因为这里的服务和类里面是没有任何关系的，所以代码可以直接拿出去
void serviceIO(int sock)
{
char buffer[1024];
while (true)
{
ssize_t n = read(sock, buffer, sizeof(buffer) - 1);
if (n > 0)
{
// 目前我们把读到的数据当成字符串, 截止目前
buffer[n] = 0;
std::cout << "recv message: " << buffer << std::endl;
std::string outbuffer = buffer;
outbuffer += " server[echo]";
write(sock, outbuffer.c_str(), outbuffer.size()); // 多路转接
}
else if (n == 0)
{
// 代表client退出
logMessage(NORMAL, "client quit, me too!");
break;
}
}
close(sock);
}
class Task
{
using func_t = std::function<void(int)>;
public:
Task()
{
}
Task(int sock, func_t func)
: _sock(sock), _callback(func)
{
}
void operator()()
{
_callback(_sock);
}
private:
int _sock;
func_t _callback;
};组件：日志修改
新函数：vprintf()
头文件：<stdarg.h>
首先使用函数前要搞明白可变参数列表
可变参数
提出问题：实现下面的写法
logMessage(NORMAL, "create socket success: %d", _listensock);
为了至此这种可变参数，需要几个宏
va_list
参考文献：va_list_百度百科
可变参数的提取逻辑
关于可变参数的提取模拟网上也大有代码可以提供参考，这里就不再赘述了
vfprintf()的工作原理
vfprintf()会直接向文件中写入，为了后序显示打印结果，这里直接使用vsnfprintf(),工作原理大差不差
初始化一个va_list
日志准备
获取时间小知识
获取时间的方式有很多种比如：
日志初版
#pragma once
#include <iostream>
#include <string>
#include <stdarg.h>
#include <ctime>
#include <unistd.h>
// 定义五种不同的信息
#define DEBUG 0
#define NORMAL 1
#define WARNING 2
#define ERROR 3 // 一种不影响服务器的错误
#define FATAL 4 // 致命错误
const char *to_levelstr(int level)
{
switch (level) // 这里直接return了
{
case DEBUG:
return "DEBUG";
case NORMAL:
return "NORMAL";
case WARNING:
return "WARNING";
case ERROR:
return "ERROR";
case FATAL:
return "FATAL";
default:
return nullptr;
}
}
// voidlogMessage(DEBUG, "hello %f, %d, %c", 3.14, 10, 'C');
void logMessage(int level, const char *format, ...)
{
// 格式如下
// [日志等级] [时间戳/时间] [pid] [message]
// [FATAL0] [2023-06-11 16:46:07] [123] [创建套接字失败]
// 可变参数的提取逻辑
/* va_list start;
va_start(start);
while (*p)
{
switch (*p)
{
case '%':
p++;
if (*p == 'f')
arg = va_arg(start, float);
...
}
}
va_end(start); */
#define NUM 1024
// 获取前缀信息[日志等级] [时间戳/时间] [pid]
char logprefix[NUM];
snprintf(logprefix, sizeof(logprefix), "[%s][%ld][pid: %d]",
to_levelstr(level), (long int)time(nullptr), getpid());
// 获取内容
char logcontent[NUM];
va_list arg;
va_start(arg, logcontent); // 因为压栈是反过来的，所以直接使用左边那个参数就行了
vsnprintf(logcontent, sizeof(logcontent), format, arg); // 第三个参数是格式， 第四个就是初始化好的可变参数
std::cout << logprefix << logcontent << std::endl;
}日志启动测试
这里就像一个日志的格式的样子了，稍后将其写入文件中
TCP通用服务器(守护进程) *
这时候一个基本的TCP服务器已经完成，但是有一个很糟糕的显现会出现，我这里是使用Xshell来进行远程链接的，但是一旦关闭Xshell窗口的时候，服务器就自动退出了，显然实际中的服务器是不能这样的，于是我们需要引入守护进程(精灵进程)的概念，利用这个方法去解决这个问题，将它守护进程化
新指令1：jobs -- 查看进程作业
新指令2：fg -- foreground(前台)
新指令3：bg(Ctrl + Z) -- background(后台)
有且只有一个前台任务！
创建新的会话
当Xshell退出的时候，会话窗口就会退出，那么任务就会被自动清理，所以需要创建新的会话
新接口2：daemon() -- 选用(本文不用)
Linux里面提供了创建守护进程的方式，但是下面就模拟实现一个daemon，来方便学习
为了完成守护进程我们满足下面四点
#pragma once
#include <unistd.h>
void daemonSelf()
{
// 1. 让调用进程忽略异常的信号 -- 否则一些进程碰到就直接挂了
// 2. 如何让自己不是组长， setsid
// 3. 守护进程是脱离终端的，关闭或者重定向以前进程默认打开的文件
// 4. 可选：进程执行路径发生更改
}
新接口3：setsid() *
小细节1：null文件 -- 任何请求都可以被接收，但是都会被丢弃
守护进程是脱离终端的，关闭或者重定向以前进程默认打开的文件，因为 0 1 2 这几个文件已经被打开使用了，是不能用的，且不能关闭 0 1 2 这几个文件 -- 一旦出现一个打印，就相当于向一个不存在的文件里面写入了，立马报错，然后崩溃，所以我们就可以利用一个特殊的文件中写入 -- null
接口1：open() -- 打开文件
返回值
open函数的返回值如果操作成功，它将返回一个文件描述符，如果操作失败，它将返回-1。
参数含义：
1、pathname:
在open函数中第一个参数pathname是指向想要打开的文件路径名，或者文件名。我们需要注意的是，这个路径名是绝对路径名。文件名则是在当前路径下的。
2、flags:
flags参数表示打开文件所采用的操作，我们需要注意的是：必须指定以下三个常量的一种，且只允许指定一个
- O_RDONLY：只读模式
- O_WRONLY：只写模式
- O_RDWR：可读可写
参考文献：linux open函数详解_open 函数具体做了什么
接口2：dup2 -- 重定向
接口3：chdir() -- 进程执行路径更改(选填)
文件：daemon.hpp(守护进程)
#pragma once
#include <unistd.h>
#include <signal.h>
#include <cstdlib>
#include <cassert>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#define DEV "/dev/null"
void daemonSelf(const char *currPath = nullptr)
{
// 1. 让调用进程忽略异常的信号 -- 否则一些进程碰到就直接挂了
signal(SIGPIPE, SIG_IGN); // 意思是说，如果客户端提前关闭了，服务器还在写入就会导致错误写入导致服务器崩掉
// 2. 如何让自己不是组长， setsid
if (fork() > 0)
exit(0);
// 子进程 -- 守护进程(精灵进程)：本质就是孤儿进程的一种
pid_t n = setsid();
assert(n != -1);
// 3. 守护进程是脱离终端的，关闭或者重定向以前进程默认打开的文件
// 因为 0 1 2 这几个文件已经被打开使用了，是不能用的
// 且不能关闭 0 1 2 这几个文件 -- 一旦出现一个打印，就相当于向一个不存在的文件里面写入了，立马报错，然后崩溃
// 所以我们就可以利用一个特殊的文件中写入 -- null
// 这样重定向到nuill中，这样进程的写入和读取就不会报错了
int fd = open(DEV, O_RDWR);
if(fd >= 0)
{
dup2(fd, 0); // 本来从 0 1 2 中读，现在从fd中读，也就是dev/null 中去读
dup2(fd, 1);
dup2(fd, 2);
close(fd);
}
else
{
close(0);
close(1);
close(2);
}
// 4. 可选：进程执行路径发生更改
if(currPath) chdir(currPath); // 如果currPath被设置，这里就进行更改
}测试：守护进程
测试服务器的正常回显
日志修改：日志写入文件中保存
文章来源：https://www.toymoban.com/news/detail-488434.html
文章来源地址https://www.toymoban.com/news/detail-488434.html
全部代码文件
daemon.hpp -- 守护进程
#pragma once
#include <unistd.h>
#include <signal.h>
#include <cstdlib>
#include <cassert>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#define DEV "/dev/null"
void daemonSelf(const char *currPath = nullptr)
{
// 1. 让调用进程忽略异常的信号 -- 否则一些进程碰到就直接挂了
signal(SIGPIPE, SIG_IGN); // 意思是说，如果客户端提前关闭了，服务器还在写入就会导致错误写入导致服务器崩掉
// 2. 如何让自己不是组长， setsid
if (fork() > 0)
exit(0);
// 子进程 -- 守护进程(精灵进程)：本质就是孤儿进程的一种
pid_t n = setsid(); // 走到这里，就已经是一个 画手进程，组长进程了，这时基于第二步产生的结果
assert(n != -1);
// 3. 守护进程是脱离终端的，关闭或者重定向以前进程默认打开的文件
// 因为 0 1 2 这几个文件已经被打开使用了，是不能用的
// 且不能关闭 0 1 2 这几个文件 -- 一旦出现一个打印，就相当于向一个不存在的文件里面写入了，立马报错，然后崩溃
// 所以我们就可以利用一个特殊的文件中写入 -- null
// 这样重定向到nuill中，这样进程的写入和读取就不会报错了
int fd = open(DEV, O_RDWR);
if(fd >= 0)
{
dup2(fd, 0); // 本来从 0 1 2 中读，现在从fd中读，也就是dev/null 中去读
dup2(fd, 1);
dup2(fd, 2);
close(fd);
}
else
{
close(0);
close(1);
close(2);
}
// 4. 可选：进程执行路径发生更改
if(currPath) chdir(currPath); // 如果currPath被设置，这里就进行更改
}log.hpp -- 日志文件
#pragma once
#include <iostream>
#include <string>
#include <stdarg.h>
#include <ctime>
#include <unistd.h>
#define LOG_NORMAL "log.txt" // 前三个放入这里，后两个信息放入下面的文件中去
#define LOG_ERR "log.error"
// 定义五种不同的信息
#define DEBUG 0
#define NORMAL 1
#define WARNING 2
#define ERROR 3 // 一种不影响服务器的错误
#define FATAL 4 // 致命错误
const char *to_levelstr(int level)
{
switch (level) // 这里直接return了
{
case DEBUG:
return "DEBUG";
case NORMAL:
return "NORMAL";
case WARNING:
return "WARNING";
case ERROR:
return "ERROR";
case FATAL:
return "FATAL";
default:
return nullptr;
}
}
// voidlogMessage(DEBUG, "hello %f, %d, %c", 3.14, 10, 'C');
void logMessage(int level, const char *format, ...)
{
// 格式如下
// [日志等级] [时间戳/时间] [pid] [message]
// [FATAL0] [2023-06-11 16:46:07] [123] [创建套接字失败]
// 可变参数的提取逻辑
/* va_list start;
va_start(start);
while (*p)
{
switch (*p)
{
case '%':
p++;
if (*p == 'f')
arg = va_arg(start, float);
...
}
}
va_end(start); */
#define NUM 1024
// 获取前缀信息[日志等级] [时间戳/时间] [pid]
char logprefix[NUM];
snprintf(logprefix, sizeof(logprefix), "[%s][%ld][pid: %d]",
to_levelstr(level), (long int)time(nullptr), getpid());
// 获取内容
char logcontent[NUM];
va_list arg;
va_start(arg, format); // 因为压栈是反过来的，所以直接使用左边那个参数就行了
vsnprintf(logcontent, sizeof(logcontent), format, arg); // 第三个参数是格式， 第四个就是初始化好的可变参数
// std::cout << logprefix << logcontent << std::endl;
FILE *log = fopen(LOG_NORMAL, "a"); // 追加式写入
FILE *err = fopen(LOG_ERR, "a");
if(log != nullptr && err != nullptr)
{
FILE *curr = nullptr;
if(level == DEBUG || level == NORMAL || level == WARNING) curr = log;
if(level == ERROR || level == FATAL) curr = err;
if(curr) fprintf(curr, "%s%s\n", logprefix, logcontent);
fclose(log);
fclose(err);
}
}makefile
cc=g++
.PHONY:all
all:tcpserver tcpclient
tcpclient:tcpClient.cc
$(cc) -o $@ $^ -std=c++11
tcpserver:tcpServer.cc
$(cc) -o $@ $^ -std=c++11 -lpthread
.PHONY:clean
clean:
rm -f tcpserver tcpclienttcpClient.cc -- 客户端1
#include "tcpClient.hpp"
#include <memory>
using namespace std;
static void Usage(string proc)
{
cout << "Usage:\n\t" << proc << " serverip serverport\n\n"; // 命令提示符
}
// ./tcpclient serverip serverport 调用逻辑
int main(int argc, char *argv[])
{
if(argc != 3)
{
Usage(argv[0]);
exit(1);
}
string serverip = argv[1];
uint16_t serverport = atoi(argv[2]);
unique_ptr<TcpClient> tcli(new TcpClient(serverip, serverport));
tcli->initClient();
tcli->start();
return 0;
}tcpClient.hpp -- 客户端2
#pragma once
#include <iostream>
#include <string>
#include <cstring>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#define NUM 1024
class TcpClient
{
public:
TcpClient(const std::string &serverip, const uint16_t &port)
: _sock(1), _serverip(serverip), _serverport(port)
{
}
void initClient()
{
// 1. 创建socket
_sock = socket(AF_INET, SOCK_STREAM, 0);
if (_sock < 0)
{
// 客户端也可以有日志，不过这里就不再实现了，直接打印错误
std::cout << "socket create error" << std::endl;
exit(2);
}
// 2. tcp的客户端要不要bind？ 要的！ 但是不需要显示bind，这里的client port要让OS自定！
// 3. 要不要listen？ -- 不需要！客户端不需要建立链接
// 4. 要不要accept？ -- 不要！
// 5. 要什么？ 要发起链接！
}
void start()
{
struct sockaddr_in server;
memset(&server, 0, sizeof(server));
server.sin_family = AF_INET;
server.sin_port = htons(_serverport);
server.sin_addr.s_addr = inet_addr(_serverip.c_str());
if (connect(_sock, (struct sockaddr *)&server, sizeof(server)) != 0)
{
std::cerr << "socket connect error" << std::endl;
}
else
{
std::string msg;
while (true)
{
std::cout << "Enter# ";
std::getline(std::cin, msg);
write(_sock, msg.c_str(), msg.size());
char buffer[NUM];
int n = read(_sock, buffer, sizeof(buffer) - 1);
if (n > 0)
{
// 目前我们把读到的数据当成字符串， 截至目前
buffer[n] = 0;
std::cout << "Server回显# " << buffer << std::endl;
}
else
{
break;
}
}
}
}
~TcpClient()
{
if(_sock >= 0) close(_sock); //不写也行，因为文件描述符的生命周期随进程，所以进程退了，自然也就会自动回收了
}
private:
int _sock;
std::string _serverip;
uint16_t _serverport;
};tcpServer.cc -- 服务端1
#include "tcpServer.hpp"
#include "daemon.hpp"
#include <memory>
using namespace server;
using namespace std;
static void Usage(string proc)
{
cout << "Usage:\n\t" << proc << " local_port\n\n"; // 命令提示符
}
// tcp服务器，启动上和udp server一模一样
// ./tcpserver local_port
int main(int argc, char *argv[])
{
if (argc != 2)
{
Usage(argv[0]);
exit(USAGE_ERR);
}
// 测试 守护进程
uint16_t port = atoi(argv[1]);
unique_ptr<TcpServer> tsvr(new TcpServer(port));
tsvr->initServer();
daemonSelf();
tsvr->start();
return 0;
}tcpServer.hpp -- 服务端2
#pragma once
#include <iostream>
#include <string>
#include <cstring>
#include <cstdlib>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <sys/wait.h>
#include <signal.h>
#include <pthread.h>
#include "log.hpp"
#include "ThreadPool.hpp"
#include "Task.hpp"
namespace server
{
enum
{
USAGE_ERR = 1,
SOCKET_ERR,
BIND_ERR,
LISTEN_ERR
};
static const uint16_t gport = 8080;
static const int gbacklog = 5; // 10、20、50都可以，但是不要太大比如5千，5万
class TcpServer; // 声明
// 用以线程传参
class ThreadData
{
public:
ThreadData(TcpServer *self, int sock) : _self(self), _sock(sock)
{
}
public:
TcpServer *_self;
int _sock;
};
class TcpServer
{
public:
TcpServer(const uint16_t &port = gport) : _listensock(-1), _port(port)
{
}
void initServer()
{
// 1. 创建socket文件套接字对象 -- 流式套接字
_listensock = socket(AF_INET, SOCK_STREAM, 0); // 第三个参数默认 0
if (_listensock < 0)
{
logMessage(FATAL, "create socket error");
exit(SOCKET_ERR);
}
logMessage(NORMAL, "create socket success: %d", _listensock);
// 2.bind绑定自己的网路信息 -- 注意包含头文件
struct sockaddr_in local;
memset(&local, 0, sizeof(local));
local.sin_family = AF_INET;
local.sin_port = htons(_port); // 这里有个细节，我们会发现当我们接受数据的时候是不需要主机转网路序列的，因为关于IO类的接口，内部都帮我们实现了这一功能，这里不帮我们做是因为我们传入的是一个结构体，系统做不到
local.sin_addr.s_addr = INADDR_ANY; // 接受任意ip地址
if (bind(_listensock, (struct sockaddr *)&local, sizeof(local)) < 0)
{
logMessage(FATAL, "bind socket error");
exit(BIND_ERR);
}
logMessage(NORMAL, "bind socket success");
// 3. 设置socket 为监听状态 -- TCP与UDP不同，它先要建立链接之后，TCP是面向链接的，后面还会有“握手”过程
if (listen(_listensock, gbacklog) < 0) // 第二个参数backlog后面再填这个坑
{
logMessage(FATAL, "listen socket error");
exit(LISTEN_ERR);
}
logMessage(NORMAL, "listen socket success");
}
void start()
{
// version 4. 线程池初始化
ThreadPool<Task>::getInstance()->run(); // 让它跑起来
logMessage(NORMAL, "Thread init success");
for (;;) // 一个死循环
{
// 4. server 获取新链接
// sock 和client 进行通信的fd
struct sockaddr_in peer;
socklen_t len = sizeof(peer);
int sock = accept(_listensock, (struct sockaddr *)&peer, &len);
if (sock < 0)
{
logMessage(ERROR, "accept error, next"); // 这个不影响服务器的运行，用ERROR，就像张三不会因为没有把人招呼进来就不干了
continue;
}
logMessage(NORMAL, "accept a new link success, get new sock: %d", sock); // 因为支持可变参数了
// 日志测试
logMessage(DEBUG, "accept error, next");
logMessage(WARNING, "accept error, next");
logMessage(FATAL, "accept error, next");
logMessage(NORMAL, "accept error, next");
logMessage(NORMAL, "accept a new link success, get new sock: %d", sock); // ?
logMessage(DEBUG, "accept error, next");
logMessage(WARNING, "accept error, next");
logMessage(FATAL, "accept error, next");
logMessage(NORMAL, "accept error, next");
logMessage(NORMAL, "accept a new link success, get new sock: %d", sock); // ?
logMessage(DEBUG, "accept error, next");
logMessage(WARNING, "accept error, next");
logMessage(FATAL, "accept error, next");
logMessage(NORMAL, "accept error, next");
// version 4 线程池 -- 默认启动10个
ThreadPool<Task>::getInstance()->push(Task(sock, serviceIO));
}
}
~TcpServer() {}
private:
int _listensock; // 修改二：改为listensock 不是用来进行数据通信的，它是用来监听链接到来，获取新链接的！
uint16_t _port;
};
} // namespace serverTask.hpp -- 形成任务
#pragma once
#include <iostream>
#include <string>
#include <cstdio>
#include <functional>
// 因为这里的服务和类里面是没有任何关系的，所以代码可以直接拿出去
void serviceIO(int sock)
{
char buffer[1024];
while (true) // 其实这里的任务是不适合线程池处理的，一个任务来了就会拿走一个线程，所以线程池处理的应该是一个很快就可以结束的服务
{
ssize_t n = read(sock, buffer, sizeof(buffer) - 1);
if (n > 0)
{
// 目前我们把读到的数据当成字符串, 截止目前
buffer[n] = 0;
std::cout << "recv message: " << buffer << std::endl;
std::string outbuffer = buffer;
outbuffer += " server[echo]";
write(sock, outbuffer.c_str(), outbuffer.size()); // 多路转接
}
else if (n == 0)
{
// 代表client退出
logMessage(NORMAL, "client quit, me too!");
break;
}
}
close(sock); // 在内部关闭就行了
}
class Task
{
using func_t = std::function<void(int)>;
public:
Task()
{
}
Task(int sock, func_t func)
: _sock(sock), _callback(func)
{
}
void operator()()
{
_callback(_sock);
}
private:
int _sock;
func_t _callback;
};
Thread.hpp -- 线程池1
#pragma once
#include <iostream>
#include <string>
#include <cstring>
#include <cassert>
#include <functional>
#include <pthread.h>
namespace ThreadNs
{
typedef std::function<void *(void *)> func_t;
const int num = 1024;
class Thread
{
private:
// 在类内创建线程，想让线程执行对应的方法，需要将方法设置成为static
static void *start_routine(void *args) // 类内成员，有缺省参数！
{
Thread *_this = static_cast<Thread *>(args);
return _this->callback();
}
public:
Thread()
{
char namebuffer[num];
snprintf(namebuffer, sizeof namebuffer, "thread-%d", threadnum++);
name_ = namebuffer;
}
void start(func_t func, void *args = nullptr)
{
func_ = func;
args_ = args;
int n = pthread_create(&tid_, nullptr, start_routine, this); // TODO
assert(n == 0);
(void)n;
}
void join()
{
int n = pthread_join(tid_, nullptr);
assert(n == 0);
(void)n;
}
std::string threadname()
{
return name_;
}
~Thread()
{
// do nothing
}
void *callback() { return func_(args_);}
private:
std::string name_;
func_t func_;
void *args_;
pthread_t tid_;
static int threadnum;
};
int Thread::threadnum = 1;
} // end namespace ThreadNsThreadPool.hpp -- 线程池2
#pragma once
#include "Thread.hpp"
#include "LockGuard.hpp"
#include "log.hpp"
#include <vector>
#include <queue>
#include <mutex>
#include <pthread.h>
#include <unistd.h>
using namespace ThreadNs;
const int gnum = 10;
template <class T>
class ThreadPool;
template <class T>
class ThreadData
{
public:
ThreadPool<T> *threadpool;
std::string name;
public:
ThreadData(ThreadPool<T> *tp, const std::string &n) : threadpool(tp), name(n)
{
}
};
template <class T>
class ThreadPool
{
private:
static void *handlerTask(void *args)
{
ThreadData<T> *td = (ThreadData<T> *)args;
while (true)
{
T t;
{
LockGuard lockguard(td->threadpool->mutex());
while (td->threadpool->isQueueEmpty())
{
td->threadpool->threadWait();
}
t = td->threadpool->pop(); // pop的本质，是将任务从公共队列中，拿到当前线程自己独立的栈中
}
t();
}
delete td;
return nullptr;
}
ThreadPool(const int &num = gnum) : _num(num)
{
pthread_mutex_init(&_mutex, nullptr);
pthread_cond_init(&_cond, nullptr);
for (int i = 0; i < _num; i++)
{
_threads.push_back(new Thread());
}
}
void operator=(const ThreadPool &) = delete;
ThreadPool(const ThreadPool &) = delete;
public:
void lockQueue() { pthread_mutex_lock(&_mutex); }
void unlockQueue() { pthread_mutex_unlock(&_mutex); }
bool isQueueEmpty() { return _task_queue.empty(); }
void threadWait() { pthread_cond_wait(&_cond, &_mutex); }
T pop()
{
T t = _task_queue.front();
_task_queue.pop();
return t;
}
pthread_mutex_t *mutex()
{
return &_mutex;
}
public:
void run()
{
for (const auto &t : _threads)
{
ThreadData<T> *td = new ThreadData<T>(this, t->threadname());
t->start(handlerTask, td);
//std::cout << t->threadname() << " start ..." << std::endl;
logMessage(DEBUG, "%s start ...", t->threadname().c_str());
}
}
void push(const T &in)
{
LockGuard lockguard(&_mutex);
_task_queue.push(in);
pthread_cond_signal(&_cond);
}
~ThreadPool()
{
pthread_mutex_destroy(&_mutex);
pthread_cond_destroy(&_cond);
for (const auto &t : _threads)
delete t;
}
static ThreadPool<T> *getInstance()
{
if (nullptr == tp)
{
_singlock.lock();
if (nullptr == tp)
{
tp = new ThreadPool<T>();
}
_singlock.unlock();
}
return tp;
}
private:
int _num;
std::vector<Thread *> _threads;
std::queue<T> _task_queue;
pthread_mutex_t _mutex;
pthread_cond_t _cond;
static ThreadPool<T> *tp;
static std::mutex _singlock;
};
template <class T>
ThreadPool<T> *ThreadPool<T>::tp = nullptr;
template <class T>
std::mutex ThreadPool<T>::_singlock;
LockGuard.hpp -- 线程池(加锁部分)
#pragma once
#include <iostream>
#include <pthread.h>
class Mutex
{
public:
Mutex(pthread_mutex_t *lock_p = nullptr): lock_p_(lock_p)
{}
void lock()
{
if(lock_p_) pthread_mutex_lock(lock_p_);
}
void unlock()
{
if(lock_p_) pthread_mutex_unlock(lock_p_);
}
~Mutex()
{}
private:
pthread_mutex_t *lock_p_;
};
class LockGuard
{
public:
LockGuard(pthread_mutex_t *mutex): mutex_(mutex)
{
mutex_.lock(); //在构造函数中进行加锁
}
~LockGuard()
{
mutex_.unlock(); //在析构函数中进行解锁
}
private:
Mutex mutex_;
};
到了这里，关于简单的TCP网络程序·线程池(后端服务器)的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
