胧の宝藏之地
首页项目归档照片墙音乐灵境说说杂谈友链关于
封面

线程同步_01

写作时间：2026-07-24 20:51:17

1、条件变量

当⼀个线程互斥地访问某个变量时，它可能发现在其它线程改变状态之前，它什么也做不了。

例如⼀个线程访问队列时，发现队列为空，它只能等待，直到其它线程将⼀个节点添加到队列中。这种情况就需要⽤到条件变量

2、同步概念与竞态条件

同步：在保证数据安全的前提下，让线程能够按照某种特定的顺序访问临界资源，从⽽有效避免饥饿问题，叫做同步

竞态条件：因为时序问题，⽽导致程序异常，我们称之为竞态条件。

条件变量函数

初始化

int pthread_cond_init(pthread_cond_t restrict cond , const pthread_condattr_t restrict attr);

参数： cond ：要初始化的条件变量 attr ： NULL

销毁

int pthread_cond_destroy(pthread_cond_t *cond)

等待

int pthread_cond_wait(pthread_cond_t restrict cond , pthread_mutex_t restrict mutex);

参数：

cond ：要在这个条件变量上等待

mutex ：互斥量

唤醒等待线程

int pthread_cond_broadcast(pthread_cond_t cond); 广播唤醒

‍

int pthread_cond_signal(pthread_cond_t cond)

随机唤醒线程

‍

测试代码

#include <iostream>
#include <string.h>
#include <unistd.h>
#include <pthread.h>

pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

void *active(void *arg)
{
    std::string name = static_cast<const char *>(arg);
    while (true)
    {
        pthread_mutex_lock(&mutex);
        pthread_cond_wait(&cond, &mutex);
        std::cout << name << " 活动..." << std::endl;
        pthread_mutex_unlock(&mutex);
    }
}

int main(void)
{
    pthread_t t1, t2;
    pthread_create(&t1, NULL, active, (void *)"thread-1");
    pthread_create(&t2, NULL, active, (void *)"thread-2");
    sleep(3); // 确保两个线程已经在运⾏

    while (true)
    {
        // 对⽐测试

        // pthread_cond_signal(&cond); // 唤醒⼀个线程

        pthread_cond_broadcast(&cond); // 唤醒所有线程

        sleep(1);
    }

    pthread_join(t1, NULL);
    pthread_join(t2, NULL);
}

3、⽣产者消费者模型

‍基于BlockingQueue的⽣产者消费者模型

Mutex.hpp

#pragma once

#include<iostream>
#include<pthread.h>

class Mutex
{
public:
    Mutex()
    {
        pthread_mutex_init(&_lock , nullptr);
    }

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

    pthread_mutex_t *Ptr()
    {
        return &_lock;
    }

    void Unlock()
    {
        pthread_mutex_unlock(&_lock);
    }

    ~Mutex()
    {
        pthread_mutex_destroy(&_lock);
    }
private:
    pthread_mutex_t _lock;
};

class LockGuard
{
public:
    LockGuard(Mutex &lock):_lockref(lock)
    {
        _lockref.Lock();
    }

    ~LockGuard()
    {
        _lockref.Unlock();
    }
private:
    Mutex &_lockref;
};

‍

Cond.hpp

#ifndef __COND_HPP
#define _COND_HPP

#include<pthread.h>
#include"Mutex.hpp"

class Cond
{
public:
    Cond()
    {
        pthread_cond_init(&_cond , nullptr);
    }

    void Wait(Mutex &mutex)
    {
        int n = pthread_cond_wait(&_cond , mutex.Ptr());
        (void)n;
    }

    void Signal()
    {
        int n = pthread_cond_signal(&_cond);
        (void)n;
    }

    void Broadcast()
    {
        int n = pthread_cond_broadcast(&_cond);
        (void)n;
    }

    ~Cond()
    {
        pthread_cond_destroy(&_cond);
    }
private:
    pthread_cond_t _cond;
};

#endif

BlockQueue

#ifndef __BLOCK_QUEUE_H
#define __BLOCK_QUEUE_H

#include <iostream>
#include <pthread.h>
#include <queue>
#include "Mutex.hpp"
#include "Cond.hpp"

const int defaultcap = 5;

template <typename T>
class BlockQueue
{
public:
    BlockQueue(int cap = defaultcap) : _cap(cap)
    {
        sleep_consumer_num = 0;
        sleep_productor_num = 0;
    }

    void Enqueue(T in)
    {
        {
            LockGuard lockguard(_mutex);

            while (_bp.size() == _cap)
            {
                sleep_productor_num++;
                _productor_cond.Wait(_mutex);
                sleep_productor_num--;
            }

            _bp.push(in);

            if (sleep_consumer_num > 0)
                _consumer_cond.Signal();
        }
    }

    void Pop(T *out)
    {
        {
            LockGuard lockguard(_mutex);

            while (_bp.empty())
            {
                sleep_consumer_num++;
                _consumer_cond.Wait(_mutex);
                sleep_consumer_num--;
            }

            *out = _bp.front();
            _bp.pop();

            if (sleep_productor_num > 0)
                _productor_cond.Signal();
        }
    }

    ~BlockQueue()
    {

    }

private:
    std::queue<T> _bp;
    int _cap;

    Mutex _mutex;
    Cond _consumer_cond;
    Cond _productor_cond;

    int sleep_productor_num;
    int sleep_consumer_num;
};

#endif

‍

Task.hpp

#ifndef __TASK_HPP
#define __TASK_HPP

#include<iostream>
#include<string>
#include<functional>


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


void Print()
{
    std::cout<<"待处理任务" <<std::endl;
}

// class Task
// {
// public:
//     Task(){}

//     Task(int x , int y):_x(x) , _y(y)
//     {}

//     void Execute()
//     {
//         _result = _x + _y;
//     }

//     void operator()()
//     {
//         Execute();
//     }

//     std::string getResult()
//     {
//         return std::to_string(_x) + " + " + std::to_string(_y) + " = " +std::to_string(_result);         
//     }

//     std::string Question()
//     {
//         return std::to_string(_x) + " + " + std::to_string(_y) + " =?";
//     }

//     ~Task()
//     {}

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





#endif

‍

Thread.hpp

#pragma once

#include <iostream>
#include <string>
#include <functional>
#include <pthread.h>

namespace ThreadModule
{
    static int gnumber = 1;
    using callback_t = std::function<void()>;

    enum class TSTATUS
    {
        THREAD_NEW,
        THREAD_RUNNING,
        THREAD_STOP
    };

    std::string Status2String(TSTATUS s)
    {
        switch (s)
        {
        case TSTATUS::THREAD_NEW:
            return "THREAD_NEW";
        case TSTATUS::THREAD_RUNNING:
            return "THREAD_RUNNING";
        case TSTATUS::THREAD_STOP:
            return "THREAD_STOP";
        default:
            return "UNKNOWN";
        }
    }

    std::string IsJoined(bool joinable)
    {
        return joinable ? "true" : "false";
    }

    class Thread
    {
    private:
        void ToRunning()
        {
            _status = TSTATUS::THREAD_RUNNING;
        }
        void ToStop()
        {
            _status = TSTATUS::THREAD_STOP;
        }
        static void *ThreadRoutine(void *args)
        {
            Thread *self = static_cast<Thread *>(args);
            pthread_setname_np(self->_tid, self->_name.c_str());
            self->_cb();
            self->ToStop();
            return nullptr;
        }

    public:
        Thread(callback_t cb)
            : _tid(-1), _status(TSTATUS::THREAD_NEW), _joinable(true), _cb(cb), _result(nullptr)
        {
            _name = "New-Thread-" + std::to_string(gnumber++);
        }
        bool Start()
        {
            int n = pthread_create(&_tid, nullptr, ThreadRoutine, this);
            if (n != 0)
                return false;

            ToRunning();
            return true;
        }
        void Join()
        {
            if (_joinable)
            {
                int n = pthread_join(_tid, &_result);
                if (n != 0)
                {
                    std::cerr << "join error: " << n << std::endl;
                    return;
                }
                (void)_result;
                _status = TSTATUS::THREAD_STOP;
            }
            else
            {
                std::cerr << "error, thread join status: " << IsJoined(_joinable) << std::endl;
            }
        }

        void Die()
        {
            if (_status == TSTATUS::THREAD_RUNNING)
            {
                pthread_cancel(_tid);
                _status = TSTATUS::THREAD_STOP;
            }
        }
        void Detach()
        {
            if (_status == TSTATUS::THREAD_RUNNING && _joinable)
            {
                pthread_detach(_tid);
                _joinable = false;
            }
            else
            {
                std::cerr << "detach " << _name << " failed" << std::endl;
            }
        }
        void PrintInfo()
        {
            std::cout << "thread name : " << _name << std::endl;
            std::cout << "thread _tid : " << _tid << std::endl;
            std::cout << "thread _status : " << Status2String(_status) << std::endl;
            std::cout << "thread _joinable : " << IsJoined(_joinable) << std::endl;
        }

        ~Thread()
        {
        }

    private:
        std::string _name;
        pthread_t _tid;
        TSTATUS _status;
        bool _joinable;
        // 线程的任务，回调函数
        callback_t _cb;

        // 线程退出信息
        void *_result;
    };
}

Main.hpp

#include "BlockQueue.hpp"
#include "Thread.hpp"
#include "Task.hpp"
#include <memory>
#include <unistd.h>
#include <ctime>
#include <cstdlib>

int num = 1;
pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;

int GetNumber()
{
    pthread_mutex_lock(&lock);
    int number = num++;
    pthread_mutex_unlock(&lock);
    return number;
}

// void *ConsumerRoutine(void *args)
// {
//     int number = GetNumber();
//     std::string name = "Consumer-" + std::to_string(number);
//     pthread_setname_np(pthread_self(), name.c_str());
//     BlockQueue<int> *bp = static_cast<BlockQueue<int> *>(args);
//     while (true)
//     {
//         sleep(1);
//         int data = 0;
//         bp->Pop(&data);
//         std::cout << name << "消费: " << data << std::endl;
//     }
// }

// void *ProdoutorRoutine(void *args)
// {
//     int number = GetNumber();
//     std::string name = "Prodoutor-" + std::to_string(number);
//     pthread_setname_np(pthread_self(), name.c_str());
//     BlockQueue<int> *bp = static_cast<BlockQueue<int> *>(args);
//     int data = 10;
//     while (true)
//     {
//         sleep(1);
//         bp->Enqueue(data);
//         std::cout << name << "生产: " << data++ << std::endl;
//     }
// }

using namespace ThreadModule;

int main()
{
    srand(time(nullptr) ^ getpid());

    std::unique_ptr<BlockQueue<task_t>> bp = std::make_unique<BlockQueue<task_t>>();

    Thread consumer([&bp]()
                    {
        // int number = GetNumber();
        // std::string name = "Consumer-" + std::to_string(number);
        // pthread_setname_np(pthread_self() , name.c_str());

        while(true)
        {
            sleep(1);
            task_t t;
            bp->Pop(&t);
            // t.Execute();
            t();
            // std::cout<<"消费: " <<t.getResult() <<std::endl;
        } });

    Thread productor([&bp]()
                     {
        // int number = GetNumber();
        // std::string name = "Prodoutor-" + std::to_string(number);
        // pthread_setname_np(pthread_self(), name.c_str());

        while (true)
        {
            int x = rand() % 12 + 1;
            usleep(rand()%1000);
            int y = rand() % 6 + 1;

            // Task t(x , y);
            bp->Enqueue(Print);
            // std::cout<< "生产: " <<t.Question() << std::endl;
        } });

    consumer.Start();
    productor.Start();

    consumer.Join();
    productor.Join();

    // BlockQueue<int> *bp = new BlockQueue<int>();
    // pthread_t c , p;
    // pthread_create(&p , nullptr , ProdoutorRoutine , bp);
    // pthread_create(&c , nullptr , ConsumerRoutine , bp);
    // pthread_join(c , nullptr);

    // pthread_t c[3] , p[2];
    // pthread_create(c , nullptr , ConsumerRoutine , bp);
    // pthread_create(c+1 , nullptr , ConsumerRoutine , bp);
    // pthread_create(c+2 , nullptr , ConsumerRoutine , bp);
    // pthread_create(p , nullptr , ProdoutorRoutine , bp);
    // pthread_create(p+1 , nullptr , ProdoutorRoutine , bp);

    // pthread_join(c[0] , nullptr);
    // pthread_join(c[1] , nullptr);
    // pthread_join(c[2] , nullptr);
    // pthread_join(p[0] , nullptr);
    // pthread_join(p[1] , nullptr);

    return 0;
}

makefile

cp_bq:Main.cc
    g++ -o $@ $^ -std=c++14
.PHONY:clean
clean:
    rm -f cp_bq

‍

avatar

胧

RECOMMENDED

线程控制

2026-07-21 01:47:21

线程池——单例模式

2026-07-27 22:36:45

进程控制

2026-07-11 20:27:31

Table of Contents