linux线程池

admin2024-05-15  1

经过前面的阻塞队列和环形阻塞队列的学习。我们现在来实现一下多线程
/threadpool.h/
/* 线程池:

  • 一种线程使用模式。线程过多会带来调度开销,进而影响缓存局部性和整体性能。而线程池维护着多个线程,等待着监督管理者分配可并发执行的任务。这避免了在处理短时间任务时创建与销毁线程的代价。线程池不仅能够保证内核的充分利用,还能防止过分调度。可用线程数量应该取决于可用的并发处理器、处理器内核、内存、网络sockets等的数量。
  • 线程池的应用场景:
    1. 需要大量的线程来完成任务,且完成任务的时间比较短。 WEB服务器完成网页请求这样的任务,使用线程池技术是非常合适的。因为单个任务小,而任务数量巨大,你可以想象一个热门网站的点击次数。 但对于长时间的任务,比如一个Telnet连接请求,线程池的优点就不明显了。因为Telnet会话时间比线程的创建时间大多了。
    1. 对性能要求苛刻的应用,比如要求服务器迅速响应客户请求。
    1. 接受突发性的大量请求,但不至于使服务器因此产生大量线程的应用。突发性大量客户请求,在没有线程池情况下,将产生大量线程,虽然理论上大部分操作系统线程数目最大值不是问题,短时间内产生大量线程可能使内存到达极限,
      出现错误.
  • 线程池的种类:
  • 线程池示例:
    1. 创建固定数量线程池,循环从任务队列中获取任务对象,
    1. 获取到任务对象后,执行任务对象中的任务接口
      */

实现步骤1,我们先自定义一个线程类

这个类的解析:
1 我们先定义一个函数类型
申明一个返回值是void* 参数是void*的一个函数

typedef std::function<void*(void*)> func_t;

2 构造函数
这个构造函数和之前写的构造函数不一样,这里构造函数只是为了初始化_name属性。

    Thread()
    {
        char buffer[num];
        snprintf(buffer, sizeof buffer,"thread-%d",threadnum++);
        _name = buffer;
    }

3 start函数,我们把对线程的初始化放到了start函数里面并且通过传递this指针的方式,将本个类的属性传过去。

    void start(func_t func,void *args=nullptr)
    {
        // Context* ctx = new Context();
        // ctx->_this=this;
        // ctx->_args =_args;
        _func =func;
        _args =args;
        int n = pthread_create(&_tid, nullptr, start_routine, this);
        assert(n == 0); //编译debug的方式发布的时候存在,release方式发布,assert就不存在了,n就是一个定义了,但是没有被使用的变量
        // 在有些编译器下会有warning
        (void)n;
    }

4 start_routine创建线程执行的函数

    static void* start_routine(void*args)
    {
        Thread* _this = static_cast<Thread*>(args);
        // Context *ctx = static_cast<Context *>(args);
        // void *ret = ctx->_this->_func();
        // delete ctx;
        // return ret;
        return _this->callback();
    }

完成的代码

#pragma once
#include<iostream>
#include<string>
#include<pthread.h>
#include <stdio.h>
#include<unistd.h>
#include <functional>
#include<cassert>

// 首先我们需要什么 , 名字 ,id ,需要的执行的函数, 还有线程属性可以包括线程的堆栈大小、调度策略、分离状态等。

// class Thread;

//上下文,当成一个大号的结构体
// class Context
// {
//     typedef std::function<void*(void*)> func_t;
// public:
//     Thread *_this;
//     void *_args;
//     func_t func;
// public:
//     Context():_this(nullptr), _args(nullptr),_func(nullptr)
//     {}
//     ~Context()
//     {}
// };


class Thread
{
private:
    static void* start_routine(void*args)
    {
        Thread* _this = static_cast<Thread*>(args);
        // Context *ctx = static_cast<Context *>(args);
        // void *ret = ctx->_this->_func();
        // delete ctx;
        // return ret;
        return _this->callback();
    }
public:
    // 这里就是申明了一种函数类型
    typedef std::function<void*(void*)> func_t;
    const int num = 1024;
public:
    Thread()
    {
        char buffer[num];
        snprintf(buffer, sizeof buffer,"thread-%d",threadnum++);
        _name = buffer;
    }

    void join()
    {
        int n = pthread_join(_tid, nullptr);
        assert(n == 0);
        (void)n;
    }

    void start(func_t func,void *args=nullptr)
    {
        // Context* ctx = new Context();
        // ctx->_this=this;
        // ctx->_args =_args;
        _func =func;
        _args =args;
        int n = pthread_create(&_tid, nullptr, start_routine, this);
        assert(n == 0); //编译debug的方式发布的时候存在,release方式发布,assert就不存在了,n就是一个定义了,但是没有被使用的变量
        // 在有些编译器下会有warning
        (void)n;
    }

    std::string threadname()
    {
        return _name;
    }
     void *callback() 
    { 
        return _func(_args);
    }

    ~Thread()
    {
        // do nothing
    }
private:
    std::string _name; // 线程名字
    pthread_t _tid;   // 线程id
    void* _args; // 这个其实是pthread_create 函数的const pthread_attr_t attr属性 。             // 如果传入 NULL,线程将使用默认属性创建。线程属性可以包括线程的堆栈大小、调度策略、分离状态等。
    func_t _func; // 执行的任务函数
    static int threadnum;
};
 int Thread::threadnum = 1;

任务类Task.hpp,定义一种任务并且有一个函数对应

#pragma once
#include <functional>
#include <stdio.h>
#include <string>
#include <iostream>

class Task
{
public:
    // 申明一种函数类型
    using func_t = std::function<int(int, int, char)>;

    // 构造方法
    Task() {}
    Task(int x, int y, char op, func_t func)
        : _x(x), _y(y), _op(op), _func(func)
    {
    }

    std::string operator()()
    {
        int result = _func(_x, _y, _op);
        char buffer[1024];
        snprintf(buffer, sizeof buffer, "%d %c %d=%d", _x, _op, _y, result);
        return buffer;
    }
    std::string toTaskString()
    {
        char buffer[1024];
        snprintf(buffer, sizeof buffer, "%d %c %d = ?", _x, _op, _y);
        return buffer;
    }

private:
    int _x;
    int _y;
    char _op;
    func_t _func;
};

int mymath(int x, int y, char op)
{
    int result = 0;
    switch (op)
    {
    case '+':
        result = x + y;
        break;
    case '-':
        result = x - y;
        break;
    case '*':
        result = x * y;
        break;
    case '/':
        if (y == 0)
        {
            std::cout << "除0错误" << std::endl;
        }
        result = x / y;
        break;
    case '%':
        if (y == 0)
        {
            std::cerr << "mod zero error!" << std::endl;
            result = -1;
        }
        else
            result = x % y;
            break;
    default:
        break;
    }
    return result;
}

线程池类。重点

#include "Thread.hpp"
#include "Mutex.hpp"
#include <vector>
#include <queue>
#include <mutex>
#include <pthread.h>
const int gnum = 3;

template <class T>
class ThreadPool;

template <class T>
class ThreadData
{
public:
    ThreadPool<T> *threadPool;
    std::string name;

public:
    ThreadData(ThreadPool<T> *tp, const std::string &n)
        : threadPool(tp), name(n)
    {
    }
};

template <class T>
class ThreadPool
{
private:
    static void *handlerTask(void *args)
    {
        ThreadData<T> *td = static_cast<ThreadData<T> *>(args);
        while (true)
        {
            T t;
            {
                LockGuard LockGuard(td->threadPool->getMutex());
                // std::cout << td->threadPool->isQueueEmpty() << std::endl;
                while (td->threadPool->isQueueEmpty())
                {
                    td->threadPool->threadWait();
                }
                t = td->threadPool->pop();
            }

            std::cout << td->name << " 获取了一个任务: " << t.toTaskString() << " 并处理完成,结果是:" << t() << std::endl;
        }
        delete td;
        return nullptr;
    }
    

public:
	ThreadPool(int num = gnum)
        : _num(num)
    {
        pthread_mutex_init(&_mutex, nullptr);
        pthread_cond_init(&_cond, nullptr);

        for (int i = 0; i < _num; i++)
        {
            _thread.push_back(new Thread());
        }
    }
    void push(const T &t)
    {
        LockGuard lockguard(&_mutex);
        _queue.push(t);
        pthread_cond_signal(&_cond);
    }
    // 判断任务队列是否为空
    bool isQueueEmpty()
    {
        return _queue.empty();
    }
    void lockQueue()
    {
        pthread_mutex_lock(&_mutex);
    }

    void unlockQueue()
    {
        pthread_mutex_unlock(&_mutex);
    }

    void threadWait()
    {
        pthread_cond_wait(&_cond, &_mutex);
    }

    T pop()
    {
        T t = _queue.front();
        _queue.pop();
        return t;
    }

    void run()
    {
        for (const auto &t : _thread)
        {
            ThreadData<T> *td = new ThreadData<T>(this, t->threadname());
            t->start(handlerTask, td);
            std::cout << t->threadname() << " start ..." << std::endl;
        }
    }

    pthread_mutex_t *getMutex()
    {
        return &_mutex;
    }
    ~ThreadPool()
    {
        pthread_mutex_destroy(&_mutex);
        pthread_cond_destroy(&_cond);
        // 验证一下这里的类型
        for (const auto &t : _thread)
        {
            delete t;
        }
    }


private:
    int _num;                      // 你创建线程的数量
    std::queue<T> _queue;          // 这里我查了,stl容器基于链表实现的 // 充当任务队列
    std::vector<Thread *> _thread; // 线程池的线程,用vector管理
    pthread_mutex_t _mutex;        // 定义一个锁对象
    pthread_cond_t _cond;          // 定义条件变量

};


构造函数

对成员属性初始化

    ThreadPool(int num = gnum)
        : _num(num)
    {
        pthread_mutex_init(&_mutex, nullptr);
        pthread_cond_init(&_cond, nullptr);

        for (int i = 0; i < _num; i++)
        {
            _thread.push_back(new Thread());
        }
    }

Run函数

这里回去调用线程类的start方法。

    void run()
    {
        for (const auto &t : _thread)
        {
            ThreadData<T> *td = new ThreadData<T>(this, t->threadname());
            t->start(handlerTask, td);
            std::cout << t->threadname() << " start ..." << std::endl;
        }
    }

handlerTask方法

static void *handlerTask(void *args)
    {
        ThreadData<T> *td = static_cast<ThreadData<T> *>(args);
        while (true)
        {
            T t;
            {
                LockGuard LockGuard(td->threadPool->getMutex());
                // std::cout << td->threadPool->isQueueEmpty() << std::endl;
                while (td->threadPool->isQueueEmpty())
                {
                    td->threadPool->threadWait();
                }
                t = td->threadPool->pop();
            }

            std::cout << td->name << " 获取了一个任务: " << t.toTaskString() << " 并处理完成,结果是:" << t() << std::endl;
        }
        delete td;
        return nullptr;
    }

linux线程池,在这里插入图片描述,第1张

test.cc

#include"ThreadPool.hpp"
#include "Task.hpp"
#include <memory>
#include <unistd.h>

int main()
{
    ///一大堆的代码...

    // std::unique_ptr<ThreadPool<Task> > tp(new ThreadPool<Task>());
    ThreadPool<Task>::getInstance()->run();
    // tp->run();

    int x, y;
    char op;
    while(1)
    {
        std::cout << "请输入数据1# ";
        std::cin >> x;
        std::cout << "请输入数据2# ";
        std::cin >> y;
        std::cout << "请输入你要进行的运算#";
        std::cin >> op;
        Task t(x, y, op, mymath);
        // std::cout << "你刚刚录入了一个任务: " << t.toTaskString() << ", 确认提交吗?[y/n]# ";
        // char confirm;
        // std::cin >> confirm;
        // if(confirm == 'y') tp->push(t);
        ThreadPool<Task>::getInstance()->push(t);
        sleep(1);
    }
}


线程池其实是一个单例,将线程池的设计变为单例模式

#include "Thread.hpp"
#include "Mutex.hpp"
#include <vector>
#include <queue>
#include <mutex>
#include <pthread.h>
const int gnum = 3;

template <class T>
class ThreadPool;

template <class T>
class ThreadData
{
public:
    ThreadPool<T> *threadPool;
    std::string name;

public:
    ThreadData(ThreadPool<T> *tp, const std::string &n)
        : threadPool(tp), name(n)
    {
    }
};

template <class T>
class ThreadPool
{
private:
    static void *handlerTask(void *args)
    {
        ThreadData<T> *td = static_cast<ThreadData<T> *>(args);
        while (true)
        {
            T t;
            {
                LockGuard LockGuard(td->threadPool->getMutex());
                // std::cout << td->threadPool->isQueueEmpty() << std::endl;
                while (td->threadPool->isQueueEmpty())
                {
                    td->threadPool->threadWait();
                }
                t = td->threadPool->pop();
            }

            std::cout << td->name << " 获取了一个任务: " << t.toTaskString() << " 并处理完成,结果是:" << t() << std::endl;
        }
        delete td;
        return nullptr;
    }
    ThreadPool(int num = gnum)
        : _num(num)
    {
        pthread_mutex_init(&_mutex, nullptr);
        pthread_cond_init(&_cond, nullptr);

        for (int i = 0; i < _num; i++)
        {
            _thread.push_back(new Thread());
        }
    }

public:
    void push(const T &t)
    {
        LockGuard lockguard(&_mutex);
        _queue.push(t);
        pthread_cond_signal(&_cond);
    }
    // 判断任务队列是否为空
    bool isQueueEmpty()
    {
        return _queue.empty();
    }
    void lockQueue()
    {
        pthread_mutex_lock(&_mutex);
    }

    void unlockQueue()
    {
        pthread_mutex_unlock(&_mutex);
    }

    void threadWait()
    {
        pthread_cond_wait(&_cond, &_mutex);
    }

    T pop()
    {
        T t = _queue.front();
        _queue.pop();
        return t;
    }

    void run()
    {
        for (const auto &t : _thread)
        {
            ThreadData<T> *td = new ThreadData<T>(this, t->threadname());
            t->start(handlerTask, td);
            std::cout << t->threadname() << " start ..." << std::endl;
        }
    }

    pthread_mutex_t *getMutex()
    {
        return &_mutex;
    }
    ~ThreadPool()
    {
        pthread_mutex_destroy(&_mutex);
        pthread_cond_destroy(&_cond);
        // 验证一下这里的类型
        for (const auto &t : _thread)
        {
            delete t;
        }
    }
    static ThreadPool<T> *getInstance()
    {
        if (nullptr == tp)
        {
            _singlock.lock();
            if (nullptr == tp)
            {
                tp = new ThreadPool<T>();
            }
            _singlock.unlock();
        }
        return tp;
    }

private:
    int _num;                      // 你创建线程的数量
    std::queue<T> _queue;          // 这里我查了,stl容器基于链表实现的 // 充当任务队列
    std::vector<Thread *> _thread; // 线程池的线程,用vector管理
    pthread_mutex_t _mutex;        // 定义一个锁对象
    pthread_cond_t _cond;          // 定义条件变量
    static std::mutex _singlock;
    static ThreadPool<T> *tp;
};

template <class T>
ThreadPool<T> *ThreadPool<T>::tp = nullptr;
template <class T>
std::mutex ThreadPool<T>::_singlock; //这是std里面的mutex ,跟linux里面的不一样
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明原文出处。如若内容造成侵权/违法违规/事实不符,请联系SD编程学习网:675289112@qq.com进行投诉反馈,一经查实,立即删除!