Linux下线程池详解与实现：提升多任务处理效率的关键

time_t 是 C/C++ 语言中用于表示时间的一个数据类型，通常是一个长整型（long int 或 long long int）。它用来存储从 1970 年 1 月 1 日 00:00:00（UTC，即协调世界时）起至今的秒数，这个起始时间点被称为 Unix 时间戳的纪元（epoch）。

time() 函数

函数原型：time_t time(time_t *tloc)

功能：返回从纪元（即格林尼治时间（GMT）1970年1月1日午夜0点）开始至今的秒数。如果 tloc 不是一个空指针，time 函数还会将返回值写入 tloc 所指向的位置。

返回值：一个 time_t 类型的值，表示自纪元以来的秒数。

localtime() 函数

函数原型：struct tm *localtime(const time_t *timeptr)

功能：将 time_t 类型的时间戳转换为本地时间的 tm 结构体。

返回值：一个指向 tm 结构体的指针，包含了转换后的本地时间信息。

gmtime() 函数

函数原型：struct tm *gmtime(const time_t *timeptr)

功能：将 time_t 类型的时间戳转换为协调世界时（UTC）的 tm 结构体。

返回值：一个指向 tm 结构体的指针，包含了转换后的 UTC 时间信息。

asctime() 函数

函数原型：char *asctime(const struct tm *timeptr)

功能：将 tm 日期时间结构体转换成日期时间字符串。

返回值：指向日期时间字符串的指针。

ctime() 函数

函数原型：char *ctime(const time_t *timeval)

功能：将 time_t 秒数日期时间格式直接转换成日期时间字符串格式输出。使用更加方便，因为不需要先转换为 tm 结构体。

返回值：指向日期时间字符串的指针。如果返回的时间在1970年1月1日 0:0:0（UTC时间）之前，则返回NULL。

struct tm

struct tm 是 C/C++ 语言中的一个结构体，用于表示和处理日期和时间。它的具体结构如下：

struct tm {
    int tm_sec;   // 秒，范围从 0 到 59
    int tm_min;   // 分，范围从 0 到 59
    int tm_hour;  // 时，范围从 0 到 23
    int tm_mday;  // 一个月中的日，范围从 1 到 31
    int tm_mon;   // 月份，范围从 0 到 11（其中0表示一月，11表示十二月）
    int tm_year;  // 自1900年起的年数
    int tm_wday;  // 一周中的日，范围从 0（周日）到 6（周六）
    int tm_yday;  // 一年中的日，范围从 0 到 365
    int tm_isdst; // 夏令时标识符，小于0表示没有夏令时，等于0表示不知道，大于0表示夏令时
};

        这个结构体包含了年、月、日、时、分、秒等时间信息，使得开发者能够方便地获取和处理当前时间、进行日期和时间的计算、格式化输出等操作。

        在编程中，你可以使用 time() 函数获取当前时间的时间戳（以秒为单位），然后使用 localtime() 或 gmtime() 函数将这个时间戳转换为 struct tm 结构体，从而获取详细的日期和时间信息。

        因此如果我们要让日志获得具体的时间并且打印出来可以写出如下函数：

    std::string TimeStampExLocalTime()
    {
        time_t currtime=time(nullptr);
        struct tm*curr=localtime(&currtime);
        char time_buffer[128];
        snprintf(time_buffer,sizeof(time_buffer),"%d-%d-%d %d:%d:%d",curr->tm_year+1900,curr->tm_mon+1,curr->tm_mday,curr->tm_hour,curr->tm_min,curr->tm_sec);
        return time_buffer;
    }

如何遍历可变参数列表

在C语言中，遍历可变参数列表通常需要使用va_list、va_start、va_arg和va_end这四个宏。这些宏在stdarg.h头文件中定义，允许你在函数中处理可变数量的参数。

以下是一个简单的步骤说明如何遍历可变参数列表：

定义函数原型：在函数原型中使用...来表示可变参数列表。

初始化va_list：在函数内部，使用va_start宏来初始化一个va_list类型的变量。va_start的第一个参数是你的va_list变量，第二个参数是可变参数列表之前的最后一个固定参数。

遍历参数：使用va_arg宏来遍历参数列表。每次调用va_arg时，它都会返回列表中的下一个参数，并将va_list指针向前移动到下一个参数。你需要指定返回参数的类型。

结束遍历：在遍历完所有参数后，使用va_end宏来清理va_list变量。

下面是一个示例函数，它接受一个整数作为参数数量，然后遍历可变参数列表中的所有整数，并将它们打印出来：

#include <stdarg.h>
#include <stdio.h>

void print_numbers(int count, ...) {
    va_list args;
    int i;

    // 初始化va_list变量
    va_start(args, count);

    // 遍历参数列表
    for (i = 0; i < count; i++) {
        int number = va_arg(args, int); // 获取下一个int类型的参数
        printf("%d ", number); // 打印参数
    }

    // 清理va_list变量
    va_end(args);

    printf("\n");
}

int main() {
    print_numbers(3, 1, 2, 3); // 输出: 1 2 3
    print_numbers(5, 5, 10, 15, 20, 25); // 输出: 5 10 15 20 25
    return 0;
}

在上面的代码中，print_numbers函数首先使用va_start初始化va_list变量args。然后，它通过一个循环使用va_arg宏来遍历参数列表，并在每次迭代中打印一个整数。最后，它使用va_end来清理va_list变量。

请注意，由于C语言在编译时不会检查可变参数的类型或数量，因此在使用可变参数时应该格外小心，以避免类型不匹配或缓冲区溢出等问题。此外，确保传递给va_start的最后一个固定参数是正确的，因为va_start使用这个参数来确定可变参数列表的起始位置。

日志的实现

#pragma once

#include <iostream>
#include <fstream>
#include <string>
#include <cstdarg>
#include <ctime>
#include <unistd.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>

enum
{
    Debug = 0,
    Info,
    Warning,
    Error,
    Fatal
};

enum
{
    Screen = 10,
    OneFile,
    ClassFile
};

std::string LevelToString(int level)
{
    switch (level)
    {
    case Debug:
        return "Debug";
    case Info:
        return "Info";
    case Warning:
        return "Warning";
    case Error:
        return "Error";
    case Fatal:
        return "Fatal";
    default:
        return "Unknown";
    }
}

const int defaultstyle = Screen;
const std::string default_filename = "log.";
const std::string logdir = "log";

class Log
{
public:
    Log() : style(defaultstyle), filename(default_filename)
    {
        mkdir(logdir.c_str(), 0775);
    }
    void Enable(int sty) //
    {
        style = sty;
    }
    std::string TimeStampExLocalTime()
    {
        time_t currtime = time(nullptr);
        struct tm *curr = localtime(&currtime);
        char time_buffer[128];
        snprintf(time_buffer, sizeof(time_buffer), "%d-%d-%d %d:%d:%d",
                 curr->tm_year + 1900, curr->tm_mon + 1, curr->tm_mday,
                 curr->tm_hour, curr->tm_min, curr->tm_sec);
        return time_buffer;
    }
    void WriteLogToOneFile(const std::string &logname, const std::string &message)
    {
        umask(0);
        int fd = open(logname.c_str(), O_CREAT | O_WRONLY | O_APPEND, 0666);
        if(fd < 0) return;
        write(fd, message.c_str(), message.size());
        close(fd);
        // std::ofstream out(logname);
        // if (!out.is_open())
        //     return;
        // out.write(message.c_str(), message.size());
        // out.close();
    }
    void WriteLogToClassFile(const std::string &levelstr, const std::string &message)
    {
        std::string logname = logdir;
        logname += "/";
        logname += filename;
        logname += levelstr;
        WriteLogToOneFile(logname, message);
    }

    void WriteLog(const std::string &levelstr, const std::string &message)
    {
        switch (style)
        {
        case Screen:
            std::cout << message;
            break;
        case OneFile:
            WriteLogToClassFile("all", message);
            break;
        case ClassFile:
            WriteLogToClassFile(levelstr, message);
            break;
        default:
            break;
        }
    }
    void LogMessage(int level, const char *format, ...) // 类C的一个日志接口
    {
        char leftbuffer[1024];
        std::string levelstr = LevelToString(level);
        std::string currtime = TimeStampExLocalTime();
        std::string idstr = std::to_string(getpid());

        char rightbuffer[1024];
        va_list args; // char *, void *
        va_start(args, format);
        // args 指向了可变参数部分
        vsnprintf(rightbuffer, sizeof(rightbuffer), format, args);
        va_end(args); // args = nullptr;
        snprintf(leftbuffer, sizeof(leftbuffer), "[%s][%s][%s] ",
                 levelstr.c_str(), currtime.c_str(), idstr.c_str());

        std::string loginfo = leftbuffer;
        loginfo += rightbuffer;
        WriteLog(levelstr, loginfo);
    }
    // void operator()(int level, const char *format, ...)
    // {
    //     LogMessage(int level, const char *format, ...)
    // }
    ~Log() {}

private:
    int style;
    std::string filename;
};

Log lg;

class Conf
{
public:
    Conf()
    {
        lg.Enable(Screen);
    }
    ~Conf()
    {}
};

Conf conf;

线程池

线程池的理解

        在Linux环境下，线程池是一种用于管理和复用线程的技术，旨在避免频繁地创建和销毁线程，从而提高系统的性能和资源利用率。线程池预先创建并维护一定数量的线程，这些线程在空闲时处于等待状态，当任务到来时，线程池会分配一个空闲线程来执行任务，任务完成后线程再次回到空闲状态，等待下一个任务的分配。

        线程池的主要组件包括任务队列和线程池管理器。任务队列用于存放待执行的任务，线程池管理器则负责线程的创建、销毁、调度和管理。线程池的大小可以根据实际需求进行调整，以达到最佳的性能和资源利用率。

        线程池在Linux下的应用场景非常广泛，尤其适用于需要处理大量并发任务或突发请求的情况。例如，Web服务器在处理大量用户请求时，可以使用线程池来管理和复用线程，提高系统的响应速度和吞吐量。此外，线程池还可以用于异步任务执行、定时任务调度等场景。

        总的来说，Linux下的线程池是一种高效、灵活的并发编程技术，能够降低系统资源消耗、提高系统性能，并方便对线程并发数进行管控。通过合理使用线程池，可以显著提升Linux系统的稳定性和可扩展性。

线程池的实现

实际上线程池的实现同生产者消费者模型之间是存在密切的关系的：

线程池为生产者消费者模型提供了线程管理和复用的机制，使得生产者和消费者能够并发地执行任务，提高了系统的并发性能。

生产者可以将生成的数据或任务提交给线程池，由线程池中的线程负责执行，实现了任务的异步处理。我们可以让多个生产者向线程池推送任务，提高推送任务的效率。

消费者可以从线程池中获取线程来处理从缓冲区中取出的数据或任务，确保了消费者的处理速度与生产者的生成速度相匹配，避免了资源的浪费。可以合理的配置线程池中的线程来提高线程池的处理效率。

线程池实现的大致图示如下：

实现如下（详细请看代码，已给出详细的注释，复用了之前文章的代码）：

#pragma once

#include <iostream>
#include <queue>
#include <vector>
#include <pthread.h>
#include <functional>
#include "Log.hpp"
#include "Thread.hpp"
#include "LockGuard.hpp"

static const int defaultnum = 5;//默认的处理任务的线程数量

class ThreadData  //用于传输线程的数据，这里就先传输一个线程名
{
public:
    ThreadData(const std::string &name) : threadname(name)
    {
    }

public:
    std::string threadname;
};

template <class T>
class ThreadPool
{
    public:
    ThreadPool(int thread_num=defaultnum):_thread_num(thread_num) //构造线程池，可以指定处理任务的线程数量
    {
        pthread_mutex_init(&_mutex, nullptr); //初始化锁和条件变量
        pthread_cond_init(&_cond, nullptr);

        for(int i=0;i<_thread_num;i++) //创建线程，这里我们引用了之前写的线程接口：Thread(const std::string &threadname, func_t<T> func, T &data)
        {
            std::string threadname ="thread-"; //构建线程名
            threadname+=std::to_string(i+1);
            ThreadData td(threadname); //线程要传入的数据

            _threads.emplace_back(threadname,std::bind(&ThreadPool<T>::ThreadRun,this,std::placeholders::_1),td); //std::vector<Thread<ThreadData>> _threads;利用vector容器将所有的线程都管理起来
            // 其中需要特别注意ThreadRun这个成员函数，他是整个线程池中的线程运行函数，后续如何详见ThreadRun函数！
            lg.LogMessage(Info, "%s is created...\n", threadname.c_str()); //日志信息，lg在日志类中已定义

        }
    }
    public:
    bool Start() //启动线程
    {
        for(auto &thread : _threads) //循环将所有线程启动
        {
            thread.Start(); //调用thread类中的启动线程
            lg.LogMessage(Info, "%s is running ...\n", thread.ThreadName().c_str());
        }
        return true;
    }

    void ThreadWait(const ThreadData &td) //根据条件变量线程等待的封装
    {
        lg.LogMessage(Debug, "no task, %s is sleeping...\n", td.threadname.c_str());
        pthread_cond_wait(&_cond,&_mutex);

    }
    void ThreadWakeup() //唤醒根据条件变量线程等待的封装
    {
        pthread_cond_signal(&_cond);
    }
    void ThreadRun(ThreadData &td) //真正的线程运行的函数，在前面已经被_threads容器管理起来了
    {
        while(true)
        {
            T t; //任务的创建用于给对应的任务
            { //先给锁，保证原子性
                LockGuard lockguard(&_mutex);
                while(_q.empty()) //任务队列中为空则等待
                {
                    ThreadWait(td);
                    lg.LogMessage(Debug, "thread %s is wakeup\n", td.threadname.c_str());
                }
                t=_q.front(); //拿取任务
                _q.pop();
            }
            t(); //重载了() 就是运行任务的意思
            lg.LogMessage(Debug, "%s handler task %s done, result is : %s\n",
                          td.threadname.c_str(), t.PrintTask().c_str(), t.PrintResult().c_str());
        }
    }
    void Push(T &in) //给外部将任务推送进队列的接口
    {
         lg.LogMessage(Debug, "other thread push a task, task is : %s\n", in.PrintTask().c_str());
         LockGuard lockguard(&_mutex);
         _q.push(in);
         ThreadWakeup(); //推送完成了那么久唤醒正在等待的线程
    }
    ~ThreadPool() //析构时自动释放资源
    {
        pthread_mutex_destroy(&_mutex);
        pthread_cond_destroy(&_cond);
    }
    void Wait() //外部用于统一回收线程资源
    {
        for(auto &thread :_threads)
        {
            thread.Join();
        }
    }
private:
    std::queue<T> _q; //任务队列 
    std::vector<Thread<ThreadData>> _threads; //线程管理容器
    int _thread_num; //线程数量
    pthread_mutex_t _mutex; //互斥锁
    pthread_cond_t _cond; //条件变量
};

调用线程池的顺序：

        1、创建线程池。

        2、启动线程池。

        3、将任务传入线程池，线程池会自动处理。

如下为一个示例：

#include <iostream>
#include <memory>
#include <ctime>
#include "ThreadPool.hpp"
#include "Task.hpp"


int main()
{
    std::unique_ptr<ThreadPool<Task>> tp(new ThreadPool<Task>()); //利用智能指针创建线程池
    tp->Start(); //启动线程池

    srand((uint64_t)time(nullptr) ^ getpid());

    while (true)
    {

        int x = rand() % 100 + 1; /->此处开始到——
        usleep(1234);
        int y = rand() % 200;
        usleep(1234);
        char oper = opers[rand() % opers.size()]; 

        Task t(x, y, oper);
        std::cout << "make task: " << t.PrintTask() << std::endl; /<- 这段只是任务的前置工作

        tp->Push(t); //推送任务进入线程池的任务队列处理
        sleep(1);
    }

    tp->Wait(); //回收线程池资源
    return 0;
}

实现效果：

使用单例模式改造线程池

单例模式的理解

单例模式是一种创建型设计模式，它确保一个类仅有一个实例，并提供一个全局访问点来访问这个唯一实例。在软件设计中，单例模式用于限制某个类只能创建一个对象。这个类提供了一种访问其唯一对象的方法，可以直接访问而不需要实例化该类。

单例模式的主要优点是：

全局唯一实例：由于单例模式限制了类的实例化次数，因此可以确保全局只有一个唯一的实例。这有助于节省系统资源，避免不必要的重复创建和销毁对象。

简化访问：通过提供一个全局访问点，可以方便地获取到类的唯一实例，无需每次使用时都进行实例化。

安全性：在某些场景下，如数据库连接、文件操作等，需要保证只有一个实例进行操作，以避免资源冲突或数据不一致的问题。单例模式可以确保这种安全性。

然而，单例模式也存在一些潜在的问题和缺点：

扩展性问题：由于单例模式限制了类的实例化次数，因此可能不适用于需要多个实例的场景。这可能导致代码的可扩展性受到限制。

测试困难：由于单例模式的唯一实例特性，可能导致在单元测试中难以模拟和替换对象，从而增加了测试的复杂性。

隐藏依赖：使用单例模式的代码可能隐式地依赖于全局唯一实例的存在，这可能导致代码之间的耦合度增加，降低代码的可维护性。

单例模式的分类主要包括以下三种：

饿汉式单例：这种单例模式在类被加载的时候，就创建了唯一实例。因此，它天然就是线程安全的。其优点在于没有线程安全的问题，但由于实例在初始化时就已经建好，可能会浪费一些内存空间。

懒汉式单例：与饿汉式不同，懒汉式单例在类被加载时并不会创建实例，而是在首次调用getInstance()方法时才进行创建。因此，懒汉式单例在延迟加载和节省内存方面有一定的优势。但是，懒汉式单例在实现时需要注意线程安全问题，否则可能会出现多个线程同时创建实例的情况，破坏单例的唯一性。

登记式单例：这种单例模式是通过将类本身作为一个键，将类的唯一实例存储在某个静态的存储结构中，如Map，以便随时访问。这种方式更为灵活，但实现起来相对复杂一些。

在实现单例模式时，需要注意线程安全问题。在多线程环境下，如果没有采取适当的同步措施，可能会导致多个线程同时创建实例，从而破坏单例模式的唯一性。因此，在实现单例模式时，需要确保线程安全，例如通过使用双重检查锁定（double-checked locking）等技术。（本文会有体现）

单例模式的实现

本文主要实现的是懒汉式单例，也是根据上面的线程池进行改造而得：

我们将构造函数放到了私有但是没有进行改动，改动较大的是我们将拷贝构造以及赋值操作给删除了，这也是为了符合单例模式的特效：全局唯一实例。我们要实现当该线程池还没有被创建出来的时候创建，如果已经创建好了那么就返回创建好的指针即可：因此，需要设置一个最开始就已经建立好的指针也就是说要么是全局类型的要么是static的，由于要调用类内的变量，因此我们创建为ThreadPool<T> *ThreadPool<T>::instance;首先设置为nullptr。接着通过给一个公共的GetInstance()（由于成员变量为static类型，因此函数也需要为static类型）接口用于外部调用。在根据懒汉式的要求首次调用getInstance()方法时才进行创建，否则直接返回指针。需要注意这里有多线程访问的安全问题，因此需要加上一个锁（由于函数为static类型，锁也需要为static类型）。后续为了线程的效率问题，如果每次调用都需要加锁才能调用，那么效率会很低，因此加上一句 if (instance == nullptr) 判断是否已经创建该单例来节省操作，直接返回指针即可。如下为具体的实现：

#pragma once

#include <iostream>
#include <queue>
#include <vector>
#include <pthread.h>
#include <functional>
#include "Log.hpp"
#include "Thread.hpp"
#include "LockGuard.hpp"

static const int defaultnum = 5; // 默认的处理任务的线程数量

class ThreadData // 用于传输线程的数据，这里就先传输一个线程名
{
public:
    ThreadData(const std::string &name) : threadname(name)
    {
    }

public:
    std::string threadname;
};

template <class T>
class ThreadPool
{
private:
    ThreadPool(int thread_num = defaultnum) : _thread_num(thread_num) // 构造线程池，可以指定处理任务的线程数量
    {
        pthread_mutex_init(&_mutex, nullptr); // 初始化锁和条件变量
        pthread_cond_init(&_cond, nullptr);

        for (int i = 0; i < _thread_num; i++) // 创建线程，这里我们引用了之前写的线程接口：Thread(const std::string &threadname, func_t<T> func, T &data)
        {
            std::string threadname = "thread-"; // 构建线程名
            threadname += std::to_string(i + 1);
            ThreadData td(threadname); // 线程要传入的数据

            _threads.emplace_back(threadname, std::bind(&ThreadPool<T>::ThreadRun, this, std::placeholders::_1), td); // std::vector<Thread<ThreadData>> _threads;利用vector容器将所有的线程都管理起来
            // 其中需要特别注意ThreadRun这个成员函数，他是整个线程池中的线程运行函数，后续如何详见ThreadRun函数！
            lg.LogMessage(Info, "%s is created...\n", threadname.c_str()); // 日志信息，lg在日志类中已定义
        }
    }

    ThreadPool(const ThreadPool<T> &tp) = delete;

    ThreadPool<T> &operator=(const ThreadPool<T> &cp) = delete;

public:
    static ThreadPool<T> *GetInstance()
    {
        if (instance == nullptr)
        {
            LockGuard lockguard(&sig_lock);
            if (instance == nullptr)
            {
                lg.LogMessage(Info, "创建单例成功...\n");
                instance = new ThreadPool<T>();
            }
        }
        return instance;
    }

    bool Start() // 启动线程
    {
        for (auto &thread : _threads) // 循环将所有线程启动
        {
            thread.Start(); // 调用thread类中的启动线程
            lg.LogMessage(Info, "%s is running ...\n", thread.ThreadName().c_str());
        }
        return true;
    }

    void ThreadWait(const ThreadData &td) // 根据条件变量线程等待的封装
    {
        lg.LogMessage(Debug, "no task, %s is sleeping...\n", td.threadname.c_str());
        pthread_cond_wait(&_cond, &_mutex);
    }
    void ThreadWakeup() // 唤醒根据条件变量线程等待的封装
    {
        pthread_cond_signal(&_cond);
    }
    void ThreadRun(ThreadData &td) // 真正的线程运行的函数，在前面已经被_threads容器管理起来了
    {
        while (true)
        {
            T t; // 任务的创建用于给对应的任务
            {    // 先给锁，保证原子性
                LockGuard lockguard(&_mutex);
                while (_q.empty()) // 任务队列中为空则等待
                {
                    ThreadWait(td);
                    lg.LogMessage(Debug, "thread %s is wakeup\n", td.threadname.c_str());
                }
                t = _q.front(); // 拿取任务
                _q.pop();
            }
            t(); // 重载了() 就是运行任务的意思
            lg.LogMessage(Debug, "%s handler task %s done, result is : %s\n",
                          td.threadname.c_str(), t.PrintTask().c_str(), t.PrintResult().c_str());
        }
    }
    void Push(T &in) // 给外部将任务推送进队列的接口
    {
        lg.LogMessage(Debug, "other thread push a task, task is : %s\n", in.PrintTask().c_str());
        LockGuard lockguard(&_mutex);
        _q.push(in);
        ThreadWakeup(); // 推送完成了那么久唤醒正在等待的线程
    }
    ~ThreadPool() // 析构时自动释放资源
    {
        pthread_mutex_destroy(&_mutex);
        pthread_cond_destroy(&_cond);
    }
    void Wait() // 外部用于统一回收线程资源
    {
        for (auto &thread : _threads)
        {
            thread.Join();
        }
    }

private:
    std::queue<T> _q;                         // 任务队列
    std::vector<Thread<ThreadData>> _threads; // 线程管理容器
    int _thread_num;                          // 线程数量
    pthread_mutex_t _mutex;                   // 互斥锁
    pthread_cond_t _cond;                     // 条件变量

    static ThreadPool<T> *instance;
    static pthread_mutex_t sig_lock;
};

template <class T>
ThreadPool<T> *ThreadPool<T>::instance = nullptr;

template <class T>
pthread_mutex_t ThreadPool<T>::sig_lock = PTHREAD_MUTEX_INITIALIZER;
// 扩展1：

// int _task_num;

// int _thread_num_low_water;  // 3
// int _thread_num_high_water; // 10
// int _task_num_low_water;    // 0
// int _task_num_high_water;   // 30

感谢你耐心的看到这里ღ( ´･ᴗ･` )比心，如有哪里有错误请踢一脚作者o(╥﹏╥)o！