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

发布时间: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、线程池的应用场景

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

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

3、线程池的种类

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

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张图片\"
注意:写独占,读共享,读锁优先级高

优先级:

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

7、自旋锁

7.1 自旋锁的概念

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

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

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

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

7.2 自旋锁的相关接口

  • 初始化以及销毁锁

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

  • 加锁

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

  • 解锁
    \"在这里插入图片描述\"

8、其他常见的各种锁

  • 悲观锁:在每次取数据时,总是担心数据会被其他线程修改,所以会在取数据前先加锁(读锁,写锁,行锁等),当其他线程想要访问数据时,被阻塞挂起。互斥锁、条件变量和信号量都是悲观锁。
  • 乐观锁:每次取数据时候,总是乐观的认为数据不会被其他线程修改,因此不上锁。但是在更新数据前,会判断其他数据在更新前有没有对数据进行修改

ItVuer - 免责声明 - 关于我们 - 联系我们

本网站信息来源于互联网,如有侵权请联系:561261067@qq.com

桂ICP备16001015号