当前位置：首页 > article >正文

从零开始学习Linux（14）---线程池

article 2024/11/14 2:33:41

1.线程池

线程池是一种多线程编程技术，它提供了一个线程队列，用于存储和管理可重用的线程。当需要执行任务时，线程池会从队列中取出一个空闲线程来执行任务，而不是创建一个新的线程。任务完成后，线程被放回线程池中，等待下一次任务的分配。

以下是线程池的一些关键特性：

线程复用：线程池中的线程可以被重复使用，而不是每次需要执行任务时都创建新的线程。这可以显著减少线程创建和销毁的开销。
任务分配：任务可以被分配给线程池中的空闲线程，这样可以提高系统性能，因为线程的创建和销毁是耗时的。
线程管理：线程池提供了一种机制来控制线程的数量，例如限制线程池中同时运行的线程数量。这有助于防止系统资源过度消耗。
任务队列：线程池通常包含一个任务队列，用于存储需要执行的任务。当线程完成一个任务后，它会从队列中取出下一个任务。
线程生命周期：线程池可以管理线程的生命周期，包括创建、运行和销毁。这有助于简化线程管理，并提高程序的稳定性。

1.可重入

重入：重入（Reentrancy）是操作系统和编程语言中的一个概念，它指的是一个函数或方法可以被多次调用，而不会因为这些调用而产生任何错误或异常。在多线程环境中，重入性尤其重要，因为它允许同一个线程多次进入同一个函数或方法，而不会导致死锁或其他同步问题。

以下是一些常见的不可重入情况：

递归调用：
- 递归调用是指一个函数或方法在其内部调用了自身。在递归过程中，如果函数或方法没有正确处理锁的释放和获取，可能会导致死锁。
- 例如，一个递归函数尝试在递归过程中获取同一个锁，但没有在递归结束时释放锁，这将导致死锁。
全局锁：
- 如果一个锁被全局地用于保护多个资源，而没有考虑资源的局部性，那么这个锁可能会变得不可重入。
- 例如，一个全局锁被用来保护一个共享资源和一个私有资源，如果一个线程在访问私有资源时持有全局锁，那么其他线程在访问共享资源时也将无法获取锁，从而导致死锁。
资源依赖：
- 如果一个函数或方法在执行过程中依赖外部资源，而这些资源在同一时间内只能被一个线程访问，那么这个函数或方法可能会变得不可重入。
- 例如，一个函数使用了一个外部锁来访问一个资源，如果该函数在执行过程中被另一个线程调用，而该线程也试图访问相同的资源，可能会导致死锁。
锁的递归性：
- 如果一个锁的实现不支持递归调用，那么任何尝试递归获取该锁的函数或方法都会导致死锁。
- 例如，一个互斥锁的实现不支持递归，如果一个函数在持有该锁时再次尝试获取它，将会导致死锁。

2.线程安全

线程安全性：线程安全是指程序在多线程环境中能够正确执行，即使多个线程同时访问和修改共享资源，也不会导致数据不一致或错误的结果。线程安全是多线程编程中的一个关键概念，它要求程序在多个线程同时运行时能够保持一致性和正确性。

以下是一些常见的线程不安全情况：

竞态条件（Race Condition）：
- 当两个或多个线程同时访问和修改同一变量时，如果没有适当的同步机制，可能会导致竞态条件。
- 例如，两个线程同时读取一个变量，然后其中一个线程修改了该变量，而另一个线程仍在使用旧值。
数据不一致：
- 当多个线程同时修改共享数据时，如果没有同步机制，可能会导致数据不一致。
- 例如，线程A读取一个变量，然后线程B修改了该变量，但线程A没有看到B的修改，因此继续使用旧值。
资源竞争：
- 当多个线程同时访问或修改同一资源时，可能会导致资源竞争。
- 例如，多个线程同时尝试打开同一文件，如果没有适当的同步机制，可能会导致文件操作失败或数据损坏。
死锁（Deadlock）：
- 当两个或多个线程同时持有对方需要的资源，并等待对方释放资源时，可能会导致死锁。
- 例如，线程A持有资源1并等待资源2，而线程B持有资源2并等待资源1。
内存泄漏：
- 当线程在分配内存后没有正确释放时，可能会导致内存泄漏。
- 例如，一个线程分配了内存，但另一个线程使用了该内存，而第一个线程没有释放它。
条件竞争：
- 当线程之间通过条件变量进行同步时，如果条件变量被错误地使用，可能会导致条件竞争。
- 例如，线程A和线程B都检查了一个条件，然后线程A修改了该条件，但线程B没有看到修改，因此继续等待。

3.死锁

死锁（Deadlock）是操作系统和多线程编程中的一个重要概念，指的是两个或多个线程在等待对方持有的资源时，导致所有线程都无法继续执行的状态。死锁通常是由于线程间的资源竞争和同步机制的不当使用所引起的。

死锁的四个必要条件是：

互斥条件：至少有一个资源是互斥的，即同一时间只能被一个线程使用。
持有和等待条件：线程持有至少一个资源，并且等待获取其他线程所持有的资源。
非抢占条件：线程持有的资源不能被其他线程抢占。
循环等待条件：存在一个线程-资源对的循环等待链，即每个线程都在等待下一个线程所持有的资源。

当这四个条件同时满足时，就可能发生死锁。例如，假设线程A持有资源1并等待资源2，线程B持有资源2并等待资源1，这时线程A和线程B都会被阻塞，因为它们都在等待对方释放资源，而对方又不会释放资源，从而形成了一个循环等待链。

以下是一些避免死锁的策略和最佳实践：

避免循环等待：
- 确保每个线程请求资源的顺序是固定的，这样就可以避免形成循环等待链。
- 例如，如果一个线程必须请求多个资源，它应该按照一个固定的顺序请求这些资源，并且一旦开始请求资源，就不改变这个顺序。
避免资源独占：
- 尽量使资源成为可共享的，或者在尽可能短的时间内释放资源。
- 如果资源必须独占使用，确保线程在释放资源之前不会再次请求其他资源。
避免长时间持有锁：
- 尽量减少线程持有锁的时间，尤其是全局锁。
- 如果一个线程必须持有多个锁，确保它尽快释放不再需要的锁。
使用锁的兼容性：
- 确保锁的兼容性，即一个线程持有的锁不会阻止其他线程获取其他锁。
- 例如，如果线程A持有锁A并请求锁B，而线程B持有锁B并请求锁A，这可能会导致死锁。
使用死锁检测和恢复：
- 一些操作系统和编程语言提供了死锁检测和恢复机制。
- 这些机制可以在检测到死锁时自动尝试恢复，例如通过撤销某个线程的锁或者重新启动线程。
使用锁的互斥性：
- 确保锁是互斥的，即同一时间只能被一个线程持有。
- 如果锁是可重入的，确保在递归调用时正确地处理锁的释放和获取。
使用最小粒度锁：
- 尽量使用最小的锁来保护共享资源，避免使用全局锁或大范围的锁。
- 这样可以减少锁的竞争，并降低发生死锁的可能性。

4.单例模式

单例模式（Singleton Pattern）是一种常用的设计模式，用于确保一个类只有一个实例，并提供一个全局访问点来获取这个实例。这种模式适用于那些需要全局访问、只有一个实例、或者创建实例很昂贵的类。

单例模式有多种实现方式，以下是一些常见的方法：

饿汉式（Eager Initialization）
- 类加载时就立即创建实例，并将其存储在一个私有静态变量中。
- 优点：线程安全，没有延迟加载。
- 缺点：实例在类加载时就创建，即使不需要也占用了资源。
懒汉式（Lazy Initialization）
- 类加载时不创建实例，而是在第一次调用 getInstance 方法时创建。
- 优点：延迟加载，只有在需要时才创建实例。
- 缺点：线程不安全，需要额外的同步机制来保证只有一个线程可以创建实例。
双重校验锁（Double-Checked Locking）
- 结合了懒汉式和同步锁的优点，通过双重检查加锁来保证线程安全。
- 优点：线程安全，延迟加载。
- 缺点：实现较为复杂，容易出错。
静态内部类（Static Inner Class）
- 使用静态内部类来创建实例，类加载时不会立即创建，而是在第一次调用 getInstance 方法时创建。
- 优点：线程安全，延迟加载。
- 缺点：实现较为复杂，容易出错。

下面是用单例模式实现的线程池：

#pragma once

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

using namespace ThreadModule;

const static int gdefaultthreadnum = 10;

// 日志
template <typename T>
class ThreadPool
{
private:
    void LockQueue()
    {
        pthread_mutex_lock(&_mutex);
    }
    void UnlockQueue()
    {
        pthread_mutex_unlock(&_mutex);
    }
    void ThreadSleep()
    {
        pthread_cond_wait(&_cond, &_mutex);
    }
    void ThreadWakeup()
    {
        pthread_cond_signal(&_cond);
    }
    void ThreadWakeupAll()
    {
        pthread_cond_broadcast(&_cond);
    }
    // 是要有的，必须是私有的
    ThreadPool(int threadnum = gdefaultthreadnum) : _threadnum(threadnum), _waitnum(0), _isrunning(false)
    {
        pthread_mutex_init(&_mutex, nullptr);
        pthread_cond_init(&_cond, nullptr);
        LOG(INFO, "ThreadPool Construct()");
    }
    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, std::placeholders::_1), 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) // 类的成员方法，也可以成为另一个类的回调方法，方便我们继续类级别的互相调用！
    {
        LOG(INFO, "%s is running...", name.c_str());
        while (true)
        {
            // 1. 保证队列安全
            LockQueue();
            // 2. 队列中不一定有数据
            while (_task_queue.empty() && _isrunning)
            {
                _waitnum++;
                ThreadSleep();
                _waitnum--;
            }
            // 2.1 如果线程池已经退出了 && 任务队列是空的
            if (_task_queue.empty() && !_isrunning)
            {
                UnlockQueue();
                break;
            }
            // 2.2 如果线程池不退出 && 任务队列不是空的
            // 2.3 如果线程池已经退出 && 任务队列不是空的 --- 处理完所有的任务，然后在退出
            // 3. 一定有任务, 处理任务
            T t = _task_queue.front();
            _task_queue.pop();
            UnlockQueue();
            LOG(DEBUG, "%s get a task", name.c_str());
            // 4. 处理任务，这个任务属于线程独占的任务
            t();
            LOG(DEBUG, "%s handler a task, result is: %s", name.c_str(), t.ResultToString().c_str());
        }
    }
    // 复制拷贝禁用
    ThreadPool<T> &operator=(const ThreadPool<T> &) = delete;
    ThreadPool(const ThreadPool<T> &) = delete;

public:
    static ThreadPool<T> *GetInstance()
    {
        // 如果是多线程获取线程池对象下面的代码就有问题了！！
        // 只有第一次会创建对象，后续都是获取
        // 双判断的方式，可以有效减少获取单例的加锁成本，而且保证线程安全
        if (nullptr == _instance) // 保证第二次之后，所有线程，不用在加锁，直接返回_instance单例对象
        {
            LockGuard lockguard(&_lock);
            if (nullptr == _instance)
            {
                _instance = new ThreadPool<T>();
                _instance->InitThreadPool();
                _instance->Start();
                LOG(DEBUG, "创建线程池单例");
                return _instance;
            }
        }
        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)
    {
        bool ret = false;
        LockQueue();
        if (_isrunning)
        {
            _task_queue.push(t);
            if (_waitnum > 0)
            {
                ThreadWakeup();
            }
            LOG(DEBUG, "enqueue task success");
            ret = true;
        }
        UnlockQueue();
        return ret;
    }
    ~ThreadPool()
    {
        pthread_mutex_destroy(&_mutex);
        pthread_cond_destroy(&_cond);
    }

private:
    int _threadnum;
    std::vector<Thread> _threads; // for fix, int temp
    std::queue<T> _task_queue;
    pthread_mutex_t _mutex;
    pthread_cond_t _cond;

    int _waitnum;
    bool _isrunning;

    // 添加单例模式
    static ThreadPool<T> *_instance;
    static pthread_mutex_t _lock;
};

template <typename T>
ThreadPool<T> *ThreadPool<T>::_instance = nullptr;

template <typename T>
pthread_mutex_t ThreadPool<T>::_lock = PTHREAD_MUTEX_INITIALIZER;