信号量和线程池

2024-11-12 08:25:36

1.信号量

POSIX信号量，用与同步操作，达到无冲突的访问共享资源目的，POSIX信号量可以用于线程间同步

初始化信号量

#include <semaphore.h>
int sem_init(sem_t *sem, int pshared, unsigned int value);

  • sem：指向sem_t类型信号量结构的指针，该结构将被初始化
  • pshared：指示信号量是否被多个进程共享，如果pshared为0，则信号量只被统一进程的线程共享，如果 pshared不为0，则信号量可以被多个进程共享
  • value：信号量的初始值
  • 成功时返回 0，失败时返回 -1，并设置 errno 以指示错误类型。

销毁信号量

int sem_destroy(sem_t *sem)；

等待信号量

int sem_wait(sem_t *sem); //P()

等待信号量，会将信号量的值-1

发布信号量

int sem_post(sem_t *sem);//V()

发布信号量，表示资源使用完毕，可以归还资源了，将信号量值+1

2.基于信号量的环形队列

在环形队列中，

  • 队列为空，让生产者先访问
  • 队列为满，让消费者先访问
  • 队列不为空&&队列不为满，生产和消费同时进行

关于消费者，消费的是数据资源，生产的是空间资源

关于生产者，生产的是数据资源，消费的是空间资源

#pragma once 

#include<iostream>
#include<vector>
#include<string>
#include<pthread.h>
#include<semaphore.h>


template<class T>
class RingQueue
{
private:

    void P(sem_t& s)
    {
        sem_wait(&s);
    }
    void V(sem_t& s)
    {
        sem_post(&s);
    }
public:
    RingQueue(int max_cap):
        _ringqueue(max_cap),_max_cap(max_cap),_c_step(0),_p_step(0)
    {
        sem_init(&_data_sem,0,0);
        sem_init(&_space_sem,0,_max_cap);

        pthread_mutex_init(&_c_mutex,nullptr);
        pthread_mutex_init(&_p_mutex,nullptr);
    }

    void Push(const T& in)//生产者
    {
        // 信号量：是一个计数器，是资源的预订机制。
        //预订在外部，可以不判断资源是否满足，就可以知道内部资源的情况！
        P(_space_sem);
        pthread_mutex_lock(&_p_mutex);
        
        _ringqueue[_p_step]=in;
        _p_step++;
        _p_step%=_max_cap;

        pthread_mutex_unlock(&_p_mutex);
        V(_data_sem);

    }

    void Pop(T* out)//消费者
    {
        P(_data_sem);
        pthread_mutex_lock(&_c_mutex);

        *out=_ringqueue[_c_step];
        _c_step++;
        _c_step%=_max_cap;

        pthread_mutex_unlock(&_c_mutex);
        V(_space_sem);


    }
    ~RingQueue()
    {
        sem_destroy(&_data_sem);
        sem_destroy(&_space_sem);

        pthread_mutex_destroy(&_c_mutex);
        pthread_mutex_destroy(&_p_mutex);
        
    }
private:
    std::vector<T> _ringqueue;
    int _max_cap;
    int _c_step;
    int _p_step;
    sem_t _data_sem;
    sem_t _space_sem;
    pthread_mutex_t _c_mutex;
    pthread_mutex_t _p_mutex;

};


#pragma once

#include<iostream>
#include<functional>

// typedef std::function<void()> task_t;
// using task_t = std::function<void()>;

// void Download()
// {
//     std::cout << "我是一个下载的任务" << std::endl;
// }


// 要做加法
class Task
{
public:
    Task()
    {
    }
    Task(int x, int y) : _x(x), _y(y)
    {
    }
    void Excute()
    {
        _result = _x + _y;
    }
    void operator ()()
    {
        Excute();
    }
    std::string debug()
    {
        std::string msg = std::to_string(_x) + "+" + std::to_string(_y) + "=?";
        return msg;
    }
    std::string result()
    {
        std::string msg = std::to_string(_x) + "+" + std::to_string(_y) + "=" + std::to_string(_result);
        return msg;
    }

private:
    int _x;
    int _y;
    int _result;
};


#include "RingQueue.hpp"
#include "Task.hpp"
#include <iostream>
#include <pthread.h>
#include <unistd.h>
#include <ctime>

void* Consumer(void* args)
{
    RingQueue<Task>* rq=static_cast<RingQueue<Task>*>(args);
    while(true)
    {
        Task t;
        rq->Pop(&t);
        t();
        std::cout<<"Consumer-> "<<t.result()<<std::endl;

    }
}

void* Productor(void* args)
{
    RingQueue<Task>* rq=static_cast<RingQueue<Task>*>(args);
    while(true)
    {
        sleep(1);
        int x=rand()%10+1;
        usleep(x*1000);
        int y=rand()%10+1;

        Task t(x,y);
        rq->Push(t);

        std::cout<<"Productor-> "<<t.debug()<<std::endl;
    }

}
int main()
{
    srand(time(nullptr)^getpid());
    RingQueue<Task>*rq=new RingQueue<Task>(5);


    pthread_t c1, c2, p1, p2, p3;
    pthread_create(&c1, nullptr, Consumer, rq);
    pthread_create(&c2, nullptr, Consumer, rq);
    pthread_create(&p1, nullptr, Productor, rq);
    pthread_create(&p2, nullptr, Productor, rq);
    pthread_create(&p3, nullptr, Productor, rq);


    pthread_join(c1, nullptr);
    pthread_join(c2, nullptr);
    pthread_join(p1, nullptr);
    pthread_join(p2, nullptr);
    pthread_join(p3, nullptr);

    return 0;
}

3.线程池

线程池的概念

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

线程池的应用场景：

1. 需要大量的线程来完成任务，且完成任务的时间比较短。 WEB服务器完成网页请求这样的任务，使用线程池技

术是非常合适的。因为单个任务小，而任务数量巨大，你可以想象一个热门网站的点击次数。 但对于长时间的任务，比如一个Telnet连接请求，线程池的优点就不明显了。因为Telnet会话时间比线程的创建时间大多了。

2. 对性能要求苛刻的应用，比如要求服务器迅速响应客户请求。

3. 接受突发性的大量请求，但不至于使服务器因此产生大量线程的应用。突发性大量客户请求，在没有线程池情

况下，将产生大量线程，虽然理论上大部分操作系统线程数目最大值不是问题，短时间内产生大量线程可能使内存到达极限，出现错误.

线程池示例：

1. 创建固定数量线程池，循环从任务队列中获取任务对象，

2. 获取到任务对象后，执行任务对象中的任务接口

线程池代码示例

4.可重入VS线程安全

概念

  • 线程安全：多个线程并发执行同一段代码时，不会出现不同的结果，常见对全局变量或者静态变量进行操作，并且没有锁的保护下，会出现该问题
  • 重入：同一个函数被不同执行流调用时，当前一个线程还没有执行完，就有其他执行流再次进入，我们称之为重入。一个函数在重入情况下，运行结果不会出现问题，则该函数称为可重入函数，否则就不是可重入函数

可重入和线程安全联系

  • 函数是可重入的，那线程就是安全的
  • 函数是不可重入的，那就不能由多个线程使用，有可能引发线程安全问题
  • 如果一个函数中有全局变量，那么这个函数既不是线程安全也不是可重入的

可重入和线程安全的区别

  • 可重入函数是线程安全函数的一种
  • 线程安全不一定是可重入，而可重入函数则一定是线程安全的
  • 如果对临界资源的访问加上锁，则这个函数是线程安全的

5.死锁

死锁的概念

死锁是指在一组进程中的各个进程均占有不会释放的资源，但因互相申请被其他进程所占有的不会释放的资源而处于永久等待的一种状态

死锁的四个必要条件

  • 互斥条件：一个资源每次只能被一个执行流使用
  • 请求与保持条件：一个执行流因请求资源而阻塞时，对已获得的资源保持不放
  • 不剥夺条件：一个执行流已获得的资源，在末使用完之前，不能强行剥夺
  • 循环等待条件：若干执行流之间形成一种头尾相接的循环等待资源的关系

避免死锁

  • 破坏死锁的四个必要条件
  • 加锁顺序一致
  • 避免锁未释放的场景
  • 资源一次性分配

上一篇：Q：警告无法解释导入PIL Pylance(reportMisssingIMports)-解决方法：


下一篇：Vue前端开发：gsap动画库-2. 实现代码