如何在线程之间同步数据

Question

这是我打印数据的课

class PrintData
{
    int data[20];

public:
    void setData(int dataValue[])
    {
        for( int i = 0 ; i < 20; i++)
        data[i] = dataValue[i];
    }
    void Print()
    {
        for (int i = 0; i < 20; i++)
            std::cout << data[i];

        std::cout << std::endl;
    }
};

这是主要功能

int number[20] ;
void updateNumber()
{
    for (int i = 0; i < 1000; i++) {
    //  std::this_thread::sleep_for(std::chrono::milliseconds(1000));
        for (int k = 0; k < 20; k++)
            number[k] = k;
    
        // after one iteration it should wait and after the print.Print() is executed than it should again update the data
    }
}

int main()
{
    PrintData print;
    std::thread t(&updateNumber);
    while (true)
    {
        // if upDateNumber has updated all the numbers than only than only set the number
        print.setData(number);
        
        print.Print();

    }
    return 0;
}

在线程中完成迭代后，它应该等待 print.setData(number) 函数执行，一旦该函数执行完毕，它应该再次更新数据。

如果 print.setData(number) 被调用并且线程仍未完成对数组的更新，那么 print.setData(number) 不应更新数据。

Answer 1

涉及conditional variables 的producer consumer 问题的简单示例如下所示：

#include <thread>
#include <mutex>
#include <iostream>
#include <condition_variable>
#include <vector>
#include <unistd.h>
#define MAX_SIZE 2

struct Task
{
    std::condition_variable m_cond;
    std::mutex m_lock;
    Task(){}
};

std::vector<int> m_data;

Task m_producer;
Task m_consumers[MAX_SIZE];
std::mutex m_lock;

static bool s_Busy = false;

static void producer(void)
{

    for(;;)
    {
        size_t input=0;

        std::unique_lock<std::mutex> lock{m_lock};//{m_producer.m_lock};
        if (!s_Busy) {
            std::cout << "Enter a number: ";
            std::cin >> input;
            std::cout << "Producer waiting..." << std::this_thread::get_id() << "\r\n";
            m_producer.m_cond.wait(lock);
        }
        s_Busy = true;
        if (m_data.size() < input) {
            for (size_t i=0; i < input; ++i){
                m_data.push_back(i);
            }
        }
        for (int i=0; i < MAX_SIZE; ++i) {
            m_consumers[i].m_cond.notify_one();
        }
        lock.unlock();
    }

}


static void consumers(void)
{
    for(;;)
    {
        std::unique_lock<std::mutex> lock{m_lock};
        if (!s_Busy) {
            std::cout <<"Consumers waiting....!" << std::this_thread::get_id() << "\r\n";
            for (int i=0; i < MAX_SIZE; ++i) {
                m_consumers[i].m_cond.notify_all();
            }
        }
        if (!m_data.empty()) {
            std::cout << "Remove: " << m_data.at(0) << std::endl;
            m_data.erase(m_data.begin());
            usleep(1);
        }
        s_Busy = false;
        m_producer.m_cond.notify_one();
        lock.unlock();
    }
}


int main()
{

    std::vector<std::thread> cnsmrs;
    std::thread usr{producer};
    for (int i=0; i < MAX_SIZE; ++i)
        cnsmrs.push_back(std::thread{consumers});

    usr.join();
    for(int i=0 ; i < MAX_SIZE; ++i)
        cnsmrs.at(i).join();
    return 0;
}

你可以玩不同的逻辑和实现。

Answer 2

希望对你有所帮助：（semaphore 是 Qt 的 QSemaphore 的自我实现）

#include <thread>
#include <mutex>
#include <condition_variable>
#include <iostream>


class semaphore
{
public:
    semaphore(int n = 0) : m_n(n) 
    {
    }

public:
    void acquire(int n = 1)
    {
        std::unique_lock <std::mutex> lk(m_buf_mut);

        while (m_n < n) {
            m_cv.wait(lk);
        }

        m_n -= n;

    }

    void release(int n = 1)
    {
        {
            std::unique_lock <std::mutex> lk(m_buf_mut);
            m_n += n;
        }

        m_cv.notify_all();
    }

    bool tryAcquire(int n = 1)
    {
        std::unique_lock <std::mutex> lk(m_buf_mut);

        if (m_n >= n) {
            m_n -= n;
            return true;
        }

        return false;
    }


private:
    std::mutex m_buf_mut;
    int m_n;
    std::condition_variable m_cv;
};

class PrintData
{
    int data[20];

public:
    void setData(int dataValue[])
    {
        for( int i = 0 ; i < 20; i++)
        data[i] = dataValue[i];
    }
    void Print()
    {
        for (int i = 0; i < 20; i++)
            std::cout << data[i];

        std::cout << std::endl;
    }
};


int number[20] ;
void updateNumber(semaphore *freeSem, semaphore *usedSem)
{
    for (int i = 0; i < 1000; i++) {
    //  std::this_thread::sleep_for(std::chrono::milliseconds(1000));
    //
    
        freeSem->acquire();

        for (int k = 0; k < 20; k++)
            number[k] = k;

        usedSem->release();
    
        // after one iteration it should wait and after the print.Print() is executed than it should again update the data
    }
}

int main()
{
    PrintData print;

    semaphore freeSem(1);
    semaphore usedSem(0);

    std::thread t(&updateNumber, &freeSem, &usedSem);
    while (true)
    {
        // if upDateNumber has updated all the numbers than only than only set the number
        usedSem.acquire();
        print.setData(number);
        
        print.Print();
        freeSem.release();

    }
    return 0;
}