目录

版本四：线程池

注意事项

文件：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被设置，这里就进行更改
}

测试：守护进程

​

测试服务器的正常回显

日志修改：日志写入文件中保存

全部代码文件

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 tcpclient

tcpClient.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 server

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;
};

​

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 ThreadNs

ThreadPool.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_;
};

​