【Linux】C++多线程：线程池构建与日志实践

前言

在构建高性能和高可用性的软件系统时，多线程编程已成为一个不可或缺的技术。它允许程序执行并发操作，从而提高资源利用率和响应速度。然而，多线程编程也引入了复杂性，尤其是在线程的创建、同步和销毁方面。为了有效管理这些线程，线程池模型被广泛采用。线程池不仅有助于减少线程创建和销毁的开销，还提供了一种优雅的方式来控制并发级别和任务调度。

此外，日志系统在软件开发中扮演着重要角色，尤其是在多线程环境下，日志记录的线程安全性和性能变得尤为重要。本文将深入探讨线程封装、多线程环境下的日志系统设计，以及线程池的概念、实现和应用。

1. 线程封装

在现代软件开发中，多线程编程已成为提高应用性能的关键技术之一。C++提供了丰富的多线程支持，但直接使用底层的线程库可能会使代码变得复杂和难以管理。因此，对线程进行封装，提供更高层次的抽象，可以简化多线程编程，提高代码的可读性和可维护性。

1.1 线程封装的基本概念

线程封装通常涉及以下几个方面：
线程的创建和管理：封装线程的创建过程，提供简洁的接口来启动和停止线程。
线程的同步：提供同步机制，如互斥锁、条件变量等，以协调线程间的协作。
线程安全的数据访问：确保线程间共享数据的访问是安全的。

1.2 实现线程封装

// Thread.hpp
#ifndef __THREAD_HPP__  // 预处理指令，防止头文件被重复包含
#define __THREAD_HPP__

#include <iostream>      // 标准输入输出流库
#include <string>        // 字符串类模板
#include <unistd.h>      // UNIX标准函数库，提供sleep等函数
#include <functional>   // 函数对象和回调
#include <pthread.h>     // POSIX线程库

namespace ThreadModule  // 命名空间，用于封装代码
{
    // 使用std::function定义一个函数类型，可以接受一个std::string参数，返回void
    using func_t = std::function<void(std::string)>;

    // 定义Thread类，用于封装线程的创建和管理
    class Thread
    {
    public:
        // 执行线程函数，调用成员函数_func，传入线程名称_threadname
        void Excute()
        {
            _func(_threadname);
        }

    public:
        // 构造函数，接受一个函数对象和线程名称，初始化成员变量
        Thread(func_t func, std::string name="none-name")
            : _func(func), _threadname(name), _stop(true)  // 默认停止标志为true
        {}

        // 静态成员函数，作为线程的入口函数
        static void *threadroutine(void *args) 
        {
            Thread *self = static_cast<Thread *>(args);  // 将传入的void*转换为Thread*
            self->Excute();  // 调用Excute成员函数执行线程任务
            return nullptr;  // 线程执行完毕返回nullptr
        }

        // 开始线程，创建并启动线程
        bool Start()
        {
            int n = pthread_create(&_tid, nullptr, threadroutine, this);  // 创建线程
            if(!n)  // 如果创建成功
            {
                _stop = false;  // 设置停止标志为false，表示线程正在运行
                return true;  // 返回true表示成功
            }
            else  // 如果创建失败
            {
                return false;  // 返回false表示失败
            }
        }

        // 将线程与创建它的线程分离，线程将独立运行
        void Detach()
        {
            if(!_stop)  // 如果线程正在运行
            {
                pthread_detach(_tid);  // 分离线程
            }
        }

        // 等待线程结束
        void Join()
        {
            if(!_stop)  // 如果线程正在运行
            {
                pthread_join(_tid, nullptr);  // 等待线程结束
            }
        }

        // 获取线程名称
        std::string name()
        {
            return _threadname;  // 返回线程名称
        }

        // 停止线程
        void Stop()
        {
            _stop = true;  // 设置停止标志为true
        }

        // 析构函数，清理资源
        ~Thread() 
        {}

    private:
        pthread_t _tid;       // 线程ID
        std::string _threadname;  // 线程名称
        func_t _func;         // 线程要执行的函数对象
        bool _stop;           // 停止标志，用于控制线程是否应该停止
    };
} // namespace ThreadModule

#endif // __THREAD_HPP__

1.3 使用线程封装

#include "Thread.hpp"  // 包含线程封装的头文件

// 定义一个简单的函数，它将作为线程要执行的任务
void PrintThreadName(const std::string& name) {
    std::cout << "Thread " << name << " is running." << std::endl;
}

int main() {
    // 创建一个线程对象，线程函数是PrintThreadName，线程名称是"TestThread"
    ThreadModule::Thread myThread(PrintThreadName, "TestThread");

    // 启动线程
    if (myThread.Start()) {
        std::cout << "Thread " << myThread.name() << " started successfully." << std::endl;
    } else {
        std::cerr << "Failed to start thread " << myThread.name() << "." << std::endl;
        return -1;  // 启动失败，退出程序
    }

    // 可以选择让主线程在这里做其他事情，或者直接等待线程完成
    sleep(1);  // 假设我们想等待1秒让线程执行

    // 等待线程结束
    myThread.Join();
    std::cout << "Thread " << myThread.name() << " finished." << std::endl;

    return 0;  // 正常退出程序
}

2. 多线程环境下的日志系统设计

在软件开发过程中，日志系统是不可或缺的一部分，它帮助开发者监控程序运行状态、调试问题以及追踪错误。在多线程应用中，日志系统的实现需要考虑线程安全和性能。

2.1 日志系统需求

多级别日志：支持不同级别的日志记录，如DEBUG、INFO、WARNING、ERROR和FATAL。
时间戳：每条日志应包含时间戳。
线程安全：在多线程环境下安全地记录日志。
灵活的输出：日志可以输出到控制台或保存到文件。

2.2 日志系统设计

2.2.1. 日志级别

定义一个枚举Level来表示不同的日志级别，方便管理和使用。

enum Level {
    DEBUG = 0,
    INFO,
    WARNING,
    ERROR,
    FATAL
};

2.2.2. 日志格式与输出

日志信息包括时间戳、日志级别、进程ID、文件名、行号和日志内容。使用LogMessage函数格式化日志消息，并根据需要输出到控制台或保存到文件。

void LogMessage(std::string filename, int line, bool issave, int level, const char *format, ...);

2.2.3. 线程安全

使用互斥锁pthread_mutex_t保证日志操作的线程安全。结合LockGuard对象自动管理锁的加解锁。

pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;

2.2.4. 日志输出控制

通过宏LOG定义日志记录的简便接口，并使用EnableFile和EnableScreen宏控制日志的输出方式。

#define LOG(level, format, ...) do { ... } while (0)
#define EnableFile() do { gIsSave = true; } while (0)
#define EnableScreen() do { gIsSave = false; } while (0)

2.3 日志系统实现

2.3.1 时间戳获取

实现GetTimeString函数，获取当前时间并格式化为字符串。

std::string GetTimeString();

2.3.2 日志级别转换

将日志级别转换为对应的字符串，方便阅读。

std::string LevelToString(int level);

2.3.3 日志保存

实现SaveFile函数，将日志信息追加到文件中。

void SaveFile(const std::string &filename, const std::string &message);

2.3.4 宏定义简化日志记录

使用宏定义简化日志记录过程，自动填充文件名、行号和日志级别。

#define LOG(level, format, ...) do { ... } while (0)

2.3.4 代码

// Log.hpp
#pragma once  // 确保头文件只被包含一次

#include <iostream>
#include <fstream>
#include <cstdio>
#include <string>
#include <ctime>
#include <cstdarg>
#include <sys/types.h>
#include <unistd.h>
#include <pthread.h>
#include "LockGuard.hpp"

// 定义全局变量，控制日志是否保存到文件
bool gIsSave = false;
// 日志文件的默认名称
const std::string logname = "log.txt";

// 日志级别定义
enum Level {
    DEBUG,  // 调试信息
    INFO,   // 一般信息
    WARNING, // 警告信息
    ERROR,  // 错误信息
    FATAL   // 严重错误信息，通常是程序不能恢复的错误
};

// 将日志信息保存到文件的函数
void SaveFile(const std::string &filename, const std::string &message) {
    std::ofstream out(filename, std::ios::app);  // 以追加模式打开文件
    if (!out.is_open()) return;  // 如果文件无法打开，直接返回
    out << message;  // 写入日志信息
    out.close();  // 关闭文件
}

// 将日志级别转换为字符串的函数
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";
    }
}

// 获取当前时间的字符串形式的函数
std::string GetTimeString() {
    time_t curr_time = time(nullptr);  // 获取当前时间
    struct tm *format_time = localtime(&curr_time);  // 将时间转换为本地时间
    if (format_time == nullptr) return "None";  // 如果转换失败，返回"None"

    char time_buffer[1024];  // 时间字符串缓冲区
    snprintf(time_buffer, sizeof(time_buffer),  // 格式化时间字符串
             "%d-%d-%d %d:%d:%d",
             format_time->tm_year + 1900, format_time->tm_mon + 1, format_time->tm_mday,
             format_time->tm_hour, format_time->tm_min, format_time->tm_sec);
    return time_buffer;  // 返回格式化的时间字符串
}

// 全局互斥锁，用于保证日志操作的线程安全
pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;

// 日志消息输出函数
void LogMessage(std::string filename, int line, bool issave, int level, const char *format, ...) {
    // 构造日志消息前缀
    std::string levelstr = LevelToString(level);
    std::string timestr = GetTimeString();
    pid_t selfid = getpid();  // 获取当前进程ID

    // 使用可变参数列表构建日志消息
    char buffer[1024];
    va_list arg;
    va_start(arg, format);
    vsnprintf(buffer, sizeof(buffer), format, arg);
    va_end(arg);

    // 构造完整的日志消息
    std::string message = "[" + timestr + "]" + "[" + levelstr + "]" +
                          "[" + std::to_string(selfid) + "]" +
                          "[" + filename + "]" + "[" + std::to_string(line) + "] " + buffer + "\n";

    // LockGuard对象，自动管理互斥锁的加解锁
    LockGuard lockguard(&lock);

    // 根据gIsSave标志决定日志输出位置
    if (!issave) {
        std::cout << message;  // 输出到控制台
    } else {
        SaveFile(logname, message);  // 保存到文件
    }
}

// 宏定义，简化日志记录操作
#define LOG(level, format, ...) do { \
    LogMessage(__FILE__, __LINE__, gIsSave, level, format, ##__VA_ARGS__); \
} while (0)

// 宏定义，控制日志是否保存到文件
#define EnableFile() do { gIsSave = true; } while (0)
#define EnableScreen() do { gIsSave = false; } while (0)

2.4 使用日志系统

#include "Log.hpp"  // 假设您的日志系统代码保存在LogSystem.hpp文件中

// 定义一个示例函数，用于演示日志记录
void exampleFunction() {
    // 使用DEBUG级别记录一条消息
    LOG(DEBUG, "This is a debug message from exampleFunction.");

    // 使用INFO级别记录一条消息
    LOG(INFO, "This function is called with INFO level logging.");

    // 模拟一些操作...
    // ...

    // 使用ERROR级别记录一条消息
    LOG(ERROR, "An error occurred in exampleFunction!");
}

int main() {
    // 启用日志信息保存到文件
    EnableFile();

    // 使用INFO级别记录一条启动信息
    LOG(INFO, "Application is starting...");

    // 调用示例函数
    exampleFunction();

    // 禁用文件保存，仅在屏幕上显示日志
    EnableScreen();

    // 使用WARNING级别记录一条消息
    LOG(WARNING, "This is a warning message displayed on screen.");

    return 0;
}

3. 线程池

在现代软件开发中，多线程是提升应用性能、实现并发处理的关键技术。线程池作为管理线程的一种高效机制，能够显著减少线程创建和销毁的开销，同时提高资源利用率和系统稳定性。本文将详细介绍线程池的概念、实现原理以及一个基于C++的线程池实现示例。

3.1 线程池的概念

线程池维护着一组工作线程，这些线程可以并发执行多个任务。线程池的核心优势包括：
资源节约：避免频繁创建和销毁线程，减少系统开销。
并发控制：限制最大并发线程数量，避免系统过载。
任务调度：智能地调度任务，提高执行效率。

3.2 线程池的实现原理

线程池的实现涉及以下关键组件：

3.2.1 任务队列

线程池包含一个任务队列，用于存储待处理的任务。工作线程从队列中获取任务并执行。

3.2.2 工作线程

线程池管理着一组工作线程，这些线程循环等待、获取并执行任务队列中的任务。

3.2.3 同步机制

为保证多线程环境下的线程安全，线程池使用互斥锁和条件变量来同步线程间的操作。

3.2.4 单例模式

线程池常采用单例模式实现，确保全局只有一个线程池实例，便于统一管理和资源共享。

3.3 线程池的实现

// ThreadPool.hpp
#pragma once  // 确保头文件只被包含一次

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

using namespace ThreadModule;  // 假设ThreadModule是包含日志和线程操作的命名空间

const static int gdefaultthreadnum = 10;  // 默认线程数量

// 线程池模板类，用于执行任意类型的任务
template <typename T>
class ThreadPool {
private:
    // 互斥锁和条件变量用于同步访问任务队列
    pthread_mutex_t _mutex;
    pthread_cond_t _cond;

    // 线程池中的线程数量
    int _threadnum;
    // 存储线程池中的线程
    std::vector<Thread> _threads;
    // 任务队列，存储待执行的任务
    std::queue<T> _task_queue;
    // 记录等待任务的线程数量
    int _waitnum;
    // 标记线程池是否正在运行
    bool _isrunning;

    // 私有构造函数，接受线程数量，默认使用gdefaultthreadnum
    ThreadPool(int threadnum = gdefaultthreadnum) : _threadnum(threadnum), _waitnum(0), _isrunning(false) {
        // 初始化互斥锁和条件变量
        pthread_mutex_init(&_mutex, nullptr);
        pthread_cond_init(&_cond, nullptr);
        LOG(INFO, "ThreadPool Construct()");  // 记录日志，ThreadPool正在构造
    }

    // 禁止拷贝赋值操作符
    ThreadPool<T> &operator=(const ThreadPool<T> &) = delete;
    // 禁止拷贝构造函数
    ThreadPool(const ThreadPool<T> &) = delete;

    // 初始化线程池，创建线程但不启动
    void InitThreadPool() {
        for (int num = 0; num < _threadnum; ++num) {
            std::string name = "thread-" + std::to_string(num + 1);
            // 创建线程，并将线程的名字和任务处理函数绑定
            _threads.emplace_back(std::bind(&ThreadPool::HandlerTask, this, name), name);
            LOG(INFO, "init thread %s done", name.c_str());
        }
        _isrunning = true;  // 线程池初始化完成，设置为运行状态
    }

    // 启动线程池中的所有线程
    void Start() {
        for (auto &thread : _threads) {
            thread.Start();  // 启动每个线程
        }
    }

    // 线程任务处理函数
    void HandlerTask(std::string name) {
        while (true) {
            LockQueue();  // 加锁访问任务队列
            while (_task_queue.empty() && _isrunning) {  // 如果任务队列为空且线程池正在运行
                _waitnum++;  // 增加等待线程计数
                ThreadSleep();  // 等待条件变量
                _waitnum--;  // 减少等待线程计数
            }
            if (_task_queue.empty() && !_isrunning) {  // 如果任务队列为空且线程池不再运行
                UnlockQueue();  // 解锁任务队列
                break;  // 退出循环，结束线程
            }
            // 从任务队列中取出任务
            T task = _task_queue.front();
            _task_queue.pop();
            UnlockQueue();  // 解锁任务队列

            LOG(DEBUG, "%s get a task", name.c_str());  // 记录日志，线程获取任务

            // 执行任务
            task();
            // 任务执行完毕后记录日志
            LOG(DEBUG, "%s handler a task, result is: %s", name.c_str(), task.ResultToString().c_str());
        }
    }

    // 线程池单例实例和锁
    static ThreadPool<T> *_instance;
    static pthread_mutex_t _lock;

public:
    // 获取线程池单例的静态方法
    static ThreadPool<T> *GetInstance() {
        if (_instance == nullptr) {  // 如果单例未创建
            LockGuard lockguard(&_lock);  // 加锁
            if (_instance == nullptr) {  // 再次检查单例是否创建，确保线程安全
                _instance = new ThreadPool<T>();  // 创建线程池实例
                _instance->InitThreadPool();  // 初始化线程池
                _instance->Start();  // 启动线程池
                LOG(DEBUG, "创建线程池单例");
            }
        }
        LOG(DEBUG, "获取线程池单例");
        return _instance;  // 返回单例指针
    }

    // 停止线程池
    void Stop() {
        LockQueue();
        _isrunning = false;  // 设置线程池为非运行状态
        ThreadWakeupAll();  // 唤醒所有等待的线程
        UnlockQueue();  // 解锁任务队列
    }

    // 等待所有线程完成它们的任务
    void Wait() {
        for (auto &thread : _threads) {
            thread.Join();  // 等待每个线程完成
            LOG(INFO, "%s is quit...", thread.name().c_str());  // 记录日志，线程退出
        }
    }

    // 向线程池添加任务
    bool Enqueue(const T &t) {
        LockQueue();
        if (!_isrunning) {  // 如果线程池不在运行状态，返回失败
            UnlockQueue();
            return false;
        }
        _task_queue.push(t);  // 将任务添加到任务队列
        if (_waitnum > 0) {  // 如果有线程正在等待任务
            ThreadWakeup();  // 唤醒一个等待的线程
        }
        LOG(DEBUG, "enqueue task success");  // 记录日志，任务添加成功
        UnlockQueue();
        return true;
    }

    // 析构函数，清理线程池资源
    ~ThreadPool() {
        Stop();  // 停止线程池
        Wait();  // 等待所有线程完成
        pthread_mutex_destroy(&_mutex);  // 销毁互斥锁
        pthread_cond_destroy(&_cond);  // 销毁条件变量
    }
};

// 初始化静态成员变量
template <typename T>
ThreadPool<T> *ThreadPool<T>::_instance = nullptr;
template <typename T>
pthread_mutex_t ThreadPool<T>::_lock = PTHREAD_MUTEX_INITIALIZER;

3.4 线程池的使用

#include "ThreadPool.hpp"  // 包含线程池的声明
#include "Task.hpp"         // 包含任务类的定义
#include "Log.hpp"          // 包含日志系统的声明
#include <iostream>         // 标准输入输出流
#include <string>           // 字符串类
#include <memory>           // 智能指针相关
#include <ctime>            // 时间相关函数

int main() {
    // 记录日志，表示程序已经加载
    LOG(DEBUG, "程序已经加载");
    
    // 休眠3秒，模拟程序启动延时
    sleep(3);
    
    // 获取线程池的单例对象
    ThreadPool<Task>::GetInstance();
    
    // 再次休眠2秒，观察单例对象是否重复创建（它不应该被重复创建）
    sleep(2);

    // 再次获取线程池单例对象，这将返回同一个实例
    ThreadPool<Task>::GetInstance();
    
    // 休眠2秒
    sleep(2);

    // 再次获取线程池单例对象
    ThreadPool<Task>::GetInstance();
    
    // 休眠2秒
    sleep(2);

    // 再次获取线程池单例对象
    ThreadPool<Task>::GetInstance();
    
    // 休眠2秒
    sleep(2);

    // 获取线程池单例对象，并调用Wait方法等待所有任务完成
    ThreadPool<Task>::GetInstance()->Wait();
    
    // 休眠2秒，等待所有任务确保已经完成
    sleep(2);

    return 0;  // 程序结束
}

总结

本文详细介绍了多线程编程中的三个关键概念：线程封装、日志系统设计，以及线程池的实现和使用。

• 线程封装：通过封装线程的创建和管理，我们简化了多线程编程的复杂性，提供了线程的启动、同步和停止的简洁接口，增强了代码的可读性和可维护性。

• 多线程环境下的日志系统设计：设计了一个支持多级别日志记录、时间戳、线程安全和灵活输出的日志系统。通过宏定义简化了日志记录的过程，并通过互斥锁确保了日志操作的线程安全。

• 线程池：重点介绍了线程池的概念和实现原理。线程池通过维护一组工作线程来执行多个任务，显著减少了线程创建和销毁的开销。线程池的实现涉及任务队列、工作线程、同步机制和单例模式。通过一个基于C++的线程池实现示例，展示了如何创建、启动、停止和使用线程池。

线程池的使用示例进一步说明了如何通过单例模式获取线程池实例，如何向线程池提交任务，以及如何等待所有任务完成。这些技术不仅提高了程序的性能，还增强了程序的稳定性和可扩展性。

通过本文的深入探讨和代码示例，读者应该能够更好地理解多线程编程中的高级概念，并将其应用于实际的软件开发项目中。随着多核处理器的普及和并发需求的增加，掌握线程封装、日志系统设计和线程池实现的技术将变得越来越重要。