多线程(线程池，读者写者，自旋锁)

发布时间：2024-06-20 08:01

文章目录

1、线程池的概念
2、线程池的应用场景
3、线程池的种类
4、线程池代码实例
5、单例模式
- 5.1 饿汉方式和懒汉方式
- 5.2 懒汉方式实现单例模式
6、读写者问题
7、自旋锁
- 7.1 自旋锁的概念
- 7.2 自旋锁的相关接口
8、其他常见的各种锁

1、线程池的概念

线程池是一种线程使用模式。线程过多会带来调度开销，进而影响缓存局部性和整体性能。而线程池维护着多个线程，等待着监督管理者分配可并发执行的任务。这避免了在处理短时间任务时创建与销毁线程的代价。线程池不仅能够保证内核的充分利用，还能防止过分调度。可用线程数量应该取决于可用的并发处理器、处理器内核、内存、网络sockets等的数量
如果我们合理的使用线程池，则可以避免把系统搞崩的窘境。总得来说，使用线程池可以带来以下几个好处：

降低资源消耗。通过重复利用已创建的线程，降低线程创建和销毁造成的消耗。
提高响应速度。当任务到达时，任务可以不需要等到线程创建就能立即执行。
增加线程的可管理型。线程是稀缺资源，使用线程池可以进行统一分配，调优和监控

$\"多线程(线程池，读者写者，自旋锁)_第1张图片\"$

没有线程池好比银行业务窗口只有一个，那么人越多，等待时间越长，效率越低。内核就好比单一窗口里面的服务人员，长时间大量处理业务，难免会出现问题。线程池的概念就好比多开放几个窗口，来节省时间，提高效率。
在服务器接收来自用户发来的任务时，需要创建线程去处理任务，而这都需要花时间为代价，如果任务队列接受一个任务创建一个线程的去处理任务，那么在面对大量任务的同时，对用户来说，需要等待很长时间，效率很低，对内核来说，需要频繁的内核申请，创建线程，销毁线程。造成内核过度调用，降低性能与效率。

2、线程池的应用场景

需要大量的线程来完成任务，且完成任务的时间比较短。 WEB服务器完成网页请求这样的任务，使用线程池技术是非常合适的。因为单个任务小，而任务数量巨大，你可以想象一个热门网站的点击次数。但对于长时间的任务，比如一个 Telnet连接请求，线程池的优点就不明显了。因为Telnet会话时间比线程的创建时间大多了。
对性能要求苛刻的应用，比如要求服务器迅速响应客户请求。
接受突发性的大量请求，但不至于使服务器因此产生大量线程的应用。突发性大量客户请求，在没有线程池情况下，将产生大量线程，虽然理论上大部分操作系统线程数目最大值不是问题，短时间内产生大量线程可能使内存到达极限，出现错误。

例如：
1:网购商品秒杀
2:云盘文件上传和下载
3:12306网上购票系统等
总之：只要有并发的地方、任务数量大或小、每个任务执行时间长或短的都可以使用线程池

3、线程池的种类

创建固定数量线程池，循环从任务队列中获取任务对象
获取到任务对象后，执行任务对象中的任务接口

4、线程池代码实例

//thread_pool.hpp
#include 
#include 
#include 
#include 
#include 
using namespace std;

namespace ns_threadpool
{
    //最大线程数量
    const int g_num = 5;
    template <class T>
    class ThreadPool
    {
    private:
        int _num;
        queue<T> _task_queue; //该成员是一个临界资源
        pthread_mutex_t _mtx;
        pthread_cond_t _cond;

    public:
        ThreadPool(int num = g_num) : _num(num)
        {
            pthread_mutex_init(&_mtx, nullptr);
            pthread_cond_init(&_cond, nullptr);
        }

        bool IsEmpty()
        {
            return _task_queue.empty();
        }

        void Lock()
        {
            pthread_mutex_lock(&_mtx);
        }

        void UnLock()
        {
            pthread_mutex_unlock(&_mtx);
        }

        void Wait()
        {
            pthread_cond_wait(&_cond, &_mtx);
        }

        void Wakeup()
        {
            //唤醒在该条件下等待的一个线程
            pthread_cond_signal(&_cond);
        }

        //在类中要让线程执行内类成员方法，是不可行的
        //必须让线程执行静态方法
        static void *Rountine(void *args)
        {
            pthread_detach(pthread_self()); //进行线程分离
            ThreadPool<T> *tp = (ThreadPool<T> *)args;
            while (true)
            {
                tp->Lock();
                //任务队列为空,线程就把自己挂起
                while (tp->IsEmpty()) //用while反正等待失败或者伪唤醒
                {
                    tp->Wait();
                }

                //代码走到这一步，说明该任务队列中一定有任务了
                T t;
                tp->PopTask(&t); // tp->_task_queue.front();
                tp->UnLock();
                t.run();
                //sleep(1);
            }
        }

        void InitThreadPool()
        {
            pthread_t tid;
            for (int i = 0; i < _num; ++i)
            {
                //这次传this原因是static成员函数不能访问内类非静态的成员变量，需要通过this访问
                pthread_create(&tid, nullptr, Rountine, (void *)this);
            }
        }

        void PushTask(const T &in)
        {
            Lock();
            _task_queue.push(in);
            UnLock();
			//添加任务后，需要唤醒线程
            Wakeup();
        }

        void PopTask(T *out)
        {
            *out = _task_queue.front();
            _task_queue.pop();
        }

        ~ThreadPool()
        {
            pthread_mutex_destroy(&_mtx);
            pthread_cond_destroy(&_cond);
        }
    };

}


//Task.hpp
#pragma once
#include 
#include 
using namespace std;
namespace fl
{
    class Task
    {
    private:
        int _x;
        int _y;
        char _op; //+-*/%

    public:
        Task() {}
        Task(int x, int y, char op)
            : _x(x), _y(y), _op(op)
        {
        }
        ~Task() {}

        int run()
        {
            int res = 0;
            switch (_op)
            {
            case \'+\':
                res = _x + _y;
                break;
            case \'-\':
                res = _x - _y;
                break;
            case \'*\':
                res = _x * _y;
                break;
            case \'/\':
                res = _x / _y;
                break;
            case \'%\':
                res = _x % _y;
                break;
            default:
                cout << \"bug?\" << endl;
                break;
            }
            cout << \"当前任务正在被：\" << pthread_self() << \"处理:\"
                 << _x << _op << _y << \"=\" << res << endl;
            return res;
        }
    };
}



//main.cpp
#include \"thread_pool.hpp\"
#include \"Task.hpp\"
#include 
#include 
using namespace ns_threadpool;
using namespace fl;

int main()
{
    ThreadPool<Task> *tp = new ThreadPool<Task>();
    tp->InitThreadPool();
    srand((long long)time(nullptr));
    while (true)
    {
        sleep(1);
        Task t(rand() % 20 + 1, rand() % 10 + 1, \"+-*/%\"[rand() % 5]);
        tp->PushTask(t);        
    }
    return 0;
}

运行代码：
$\"多线程(线程池，读者写者，自旋锁)_第2张图片\"$
添加任务后，需要唤醒线程，那为什么不唤醒在条件变量下的所有线程，而是唤醒一个线程？
因为唤醒全部线程会产生惊群效应
惊群效应（thundering herd）是指多进程（多线程）在同时阻塞等待同一个事件的时候（休眠状态），如果等待的这个事件发生，那么他就会唤醒等待的所有进程（或者线程），但是最终却只能有一个进程（线程）获得这个时间的“控制权”，对该事件进行处理，而其他进程（线程）获取“控制权”失败，只能重新进入休眠状态，这种现象和性能浪费就叫做惊群效应

5、单例模式

单例模式是一种 \"经典的, 常用的\" 设计模式
什么是设计模式？

IT行业这么火, 涌入的人很多. 俗话说林子大了啥鸟都有. 大佬和菜鸡们两极分化的越来越严重. 为了让菜鸡们不太拖大佬的后腿, 于是大佬们针对一些经典的常见的场景, 给定了一些对应的解决方案, 这个就是设计模式

单例模式的特点

在很多服务器开发场景中, 经常需要让服务器加载很多的数据 (上百G) 到内存中. 此时往往要用一个单例的类来管理这些数据.

5.1 饿汉方式和懒汉方式

单例模式可以通过饿汉实现方式和懒汉实现方式，这两者有什么区别呢？
举个洗碗的栗子：

吃完饭, 立刻洗碗, 这种就是饿汉方式. 因为下一顿吃的时候可以立刻拿着碗就能吃饭
吃完饭, 先把碗放下, 然后下一顿饭用到这个碗了再洗碗, 就是懒汉方式

5.2 懒汉方式实现单例模式

//thread_pool.hpp
#include 
#include 
#include 
#include 
#include 
using namespace std;

namespace ns_threadpool
{
    const int g_num = 5;
    template <class T>
    class ThreadPool
    {
    private:
        int _num;
        queue<T> _task_queue; //该成员是一个临界资源
        pthread_mutex_t _mtx;
        pthread_cond_t _cond;

        static ThreadPool<T> *ins;

        //单例模式
        //构造函数必须实现，但是必须私有化
        ThreadPool(int num = g_num) : _num(num)
        {
            pthread_mutex_init(&_mtx, nullptr);
            pthread_cond_init(&_cond, nullptr);
        }

        //构造
        //ThreadPool(const ThreadPool &tp) = delete;

        //赋值
        //ThreadPool &operator(ThreadPool &tp) = delete;

    public:

        //静态成员函数可以通过类进行调用
        static ThreadPool<T> *GetInstance()
        {
            static pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;

            //双重判定减少锁的争用，提高获取单例的效率
            if (ins == nullptr)
            {
                //当前的单例对象还没有被创建
                if (ins == nullptr)
                {
                    pthread_mutex_lock(&lock);
                    ins = new ThreadPool<T>();
                    ins->InitThreadPool();
                    cout << \"首次加载对象\" << endl;
                }
                pthread_mutex_unlock(&lock);
            }
            return ins;
        }

        bool IsEmpty()
        {
            return _task_queue.empty();
        }

        void Lock()
        {
            pthread_mutex_lock(&_mtx);
        }

        void UnLock()
        {
            pthread_mutex_unlock(&_mtx);
        }

        void Wait()
        {
            pthread_cond_wait(&_cond, &_mtx);
        }

        void Wakeup()
        {
            pthread_cond_signal(&_cond);
        }

        //在类中要让线程执行内类成员方法，是不可行的
        //必须让线程执行静态方法
        static void *Rountine(void *args)
        {
            pthread_detach(pthread_self()); //进行线程分离
            ThreadPool<T> *tp = (ThreadPool<T> *)args;
            while (true)
            {
                tp->Lock();
                //任务队列为空,线程就把自己挂起
                while (tp->IsEmpty()) //用while反正等待失败或者伪唤醒
                {
                    tp->Wait();
                }

                //代码走到这一步，说明该任务队列中一定有任务了
                T t;
                tp->PopTask(&t); // tp->_task_queue.front();
                tp->UnLock();
                t.run();
                // sleep(1);
            }
        }

        void InitThreadPool()
        {
            pthread_t tid;
            for (int i = 0; i < _num; ++i)
            {
                //这次传this原因是static成员函数不能访问内类非静态的成员变量，需要通过this访问
                pthread_create(&tid, nullptr, Rountine, (void *)this);
            }
        }

        void PushTask(const T &in)
        {
            Lock();
            _task_queue.push(in);
            UnLock();

            Wakeup();
        }

        void PopTask(T *out)
        {
            *out = _task_queue.front();
            _task_queue.pop();
        }

        ~ThreadPool()
        {
            pthread_mutex_destroy(&_mtx);
            pthread_cond_destroy(&_cond);
        }
    };

    template <class T>
    ThreadPool<T> *ThreadPool<T>::ins = nullptr;
}



//Task.hpp
#pragma once
#include 
#include 
using namespace std;
namespace fl
{
    class Task
    {
    private:
        int _x;
        int _y;
        char _op; //+-*/%

    public:
        Task() {}
        Task(int x, int y, char op)
            : _x(x), _y(y), _op(op)
        {
        }
        ~Task() {}

        int run()
        {
            int res = 0;
            switch (_op)
            {
            case \'+\':
                res = _x + _y;
                break;
            case \'-\':
                res = _x - _y;
                break;
            case \'*\':
                res = _x * _y;
                break;
            case \'/\':
                res = _x / _y;
                break;
            case \'%\':
                res = _x % _y;
                break;
            default:
                cout << \"bug?\" << endl;
                break;
            }
            cout << \"当前任务正在被：\" << pthread_self() << \"处理:\"
                 << _x << _op << _y << \"=\" << res << endl;
            return res;
        }
    };
}



//main.cpp
#include \"thread_pool.hpp\"
#include \"Task.hpp\"
#include 
#include 
using namespace fl;
using namespace ns_threadpool;
using namespace std;
int main()
{
    cout << \"当前正在运行我的进程其他代码...\" << endl;
    cout << \"当前正在运行我的进程其他代码...\" << endl;
    cout << \"当前正在运行我的进程其他代码...\" << endl;
    cout << \"当前正在运行我的进程其他代码...\" << endl;
    cout << \"当前正在运行我的进程其他代码...\" << endl;

    // sleep(5);
    srand((long long)time(nullptr));
    while (true)
    {
        sleep(1);
        Task t(rand() % 20 + 1, rand() % 10 + 1, \"+-*/%\"[rand() % 5]);

        ThreadPool<Task>::GetInstance()->PushTask(t);
    }
    return 0;
}

运行代码：
$\"多线程(线程池，读者写者，自旋锁)_第3张图片\"$

6、读写者问题

在编写多线程的时候，有一种情况是十分常见的。那就是，有些公共数据修改的机会比较少。相比较改写，它们读的机会反而高的多。通常而言，在读的过程中，往往伴随着查找的操作，中间耗时很长。给这种代码段加锁，会极大地降低我们程序的效率。那么有没有一种方法，可以专门处理这种多读少写的情况呢？有，那就是读写锁。
$\"多线程(线程池，读者写者，自旋锁)_第4张图片\"$
注意：写独占，读共享，读锁优先级高

优先级：

读者优先：读者和写者同时到来的时候，我们让读者先进入访问。因为读者多写者少，可能会存在写饥饿问题。
写者优先：当读者和写者同时到来的时候，比如当前写者晚来的的时候，所有的读者，都不要进入临界区访问，等临界区中没有读者的时候，让写者先写入。

7、自旋锁

7.1 自旋锁的概念

自旋锁：当一个线程尝试去获取某一把锁的时候，如果这个锁此时已经被别人获取(占用)，那么此线程就无法获取到这把锁，该线程将会等待，间隔一段时间后会再次尝试获取

跟互斥锁一样，一个执行单元要想访问被自旋锁保护的共享资源，必须先得到锁，在访问完共享资源后，必须释放锁。如果在获取自旋锁时，没有任何执行单元保持该锁，那么将立即得到锁;如果在获取自旋锁时锁已经有保持者，那么获取锁操作将自旋在那里，直到该自旋锁的保持者释放了锁。由此我们可以看出，自旋锁是一种比较低级的保护数据结构或代码片段的原始方式，这种锁可能存在两个问题：

可能会形成死锁：试图递归地获得自旋锁必然会引起死锁:递归程序的持有实例在第二个实例循环，以试图获得相同自旋锁时，不会释放此自旋锁。在递归程序中使用自旋锁应遵守下列策略:递归程序决不能在持有自旋锁时调用它自己，也决不能在递归调用时试图获得相同的自旋锁。此外如果一个进程已经将资源锁定，那么，即使其它申请这个资源的进程不停地疯狂\"自旋\",也无法获得资源，从而进入死循环
过多占用cpu资源：如果不加限制，由于申请者一直在循环等待，因此自旋锁在锁定的时候,如果不成功,不会睡眠,会持续的尝试，因此会耗费大量CPU资源。因此，一般自旋锁实现会有一个参数限定最多持续尝试次数，超出后, 自旋锁放弃当前time slice. 等下一次机会

由此可见，自旋锁比较适用于锁使用者保持锁时间比较短的情况。因为自旋的消耗会小于线程阻塞挂起再唤醒的操作的消耗，这些操作会导致线程发生两次上下文切换，正是由于自旋锁使用者一般保持锁时间非常短，因此选择自旋而不是睡眠是非常必要的，所以在短时间内自旋锁的效率远高于互斥锁。

7.2 自旋锁的相关接口

初始化以及销毁锁

$\"在这里插入图片描述\"$

加锁