通过公共数据结构的两个线程进行通信.设计问题

Question

我目前有两个线程是生产者和消费者.生成器是一种静态方法,它在Deque类型的静态容器中插入数据,并通过boost::condition_variable在deque对象中插入对象来通知使用者.然后,使用者从Deque类型中读取数据并将其从容器中删除.两个线程使用进行通信boost::condition_variable

这是正在发生的事情的摘要.这是消费者和生产者的代码

    //Static Method : This is the producer. Different classes add data to the container using this method
    void C::Add_Data(obj a)
    {
        try
        {       
            int a = MyContainer.size();
            UpdateTextBoxA("Current Size is " + a);
            UpdateTextBoxB("Running"); 
            MyContainer.push_back(a);
            condition_consumer.notify_one(); //This condition is static member
            UpdateTextBoxB("Stopped");                 
        }
        catch (std::exception& e)
        {
            std::string err = e.what();
        }
    }//end method


    //Consumer Method - Runs in a separate independent thread
    void C::Read_Data()
    {
        while(true)
        {
            boost::mutex::scoped_lock lock(mutex_c);
            while(MyContainer.size()!=0)
            {
                try
                {
                    obj a = MyContainer.front();
                    ....
                    ....
                    ....
                    MyContainer.pop_front();
                }
                catch (std::exception& e)
                {
                    std::string err = e.what();
                }
            }
            condition_consumer.wait(lock);
        }

    }//end method

现在插入到Deque类型对象中的对象非常快,每秒约500个对象.虽然运行它,但我注意到TextBoxB始终处于"已停止"状态,而我认为它应该在"正在运行"和"停止"之间切换.加上很慢.关于我可能没有考虑过什么可能做错的任何建议？

Answer 1

1)您应该使用MyContainer.push_back(a);互斥锁 - 否则您将获得数据竞争,这是未定义的行为(+您可能还需要MyContainer.size();通过互斥锁保护,具体取决于它的类型和您使用的C++ ISO /编译器版本).

2)void C::Read_Data()应该是:

void C::Read_Data()
{
    scoped_lock slock(mutex_c);
    while(true) // you may also need some exit condition/mechanism
    {
        condition_consumer.wait(slock,[&]{return !MyContainer.empty();});
        // at this line MyContainer.empty()==false and slock is locked
        // so you may pop value from deque
    }
}

3)您将并发队列的逻辑与生成/消费逻辑混合在一起.相反,您可以将并发队列部分隔离到独立实体:

现场演示

// C++98
template<typename T>
class concurrent_queue
{
    queue<T> q;
    mutable mutex m;
    mutable condition_variable c;
public:
    void push(const T &t)
    {
        (lock_guard<mutex>(m)),
            q.push(t),
            c.notify_one();
    }
    void pop(T &result)
    {
        unique_lock<mutex> u(m);
        while(q.empty())
            c.wait(u);
        result = q.front();
        q.pop();
    }
};

感谢您的回复.你能解释条件等待语句中的第二个参数吗？[&]{return !MyContainer.empty();}

condition_variable :: wait的第二个版本将谓词作为第二个参数.它基本上等于谓词是假的,有助于"忽略" 虚假的唤醒.

[&]{return !MyContainer.empty();}- 这是lambda函数.它是C++ 11的新功能 - 它允许"就地"定义功能.如果您没有C++ 11,那么只需创建独立谓词或使用wait带有手动while循环的单参数版本:

while(MyContainer.empty()) condition_consumer.wait(lock);

在第3点中你提出我应该隔离整个队列的一个问题,而我添加到队列方法是静态的,而消费者(队列读取器)在一个单独的线程中永远运行.你能告诉我为什么这是我设计中的缺陷吗？

"永远跑"或者没有问题static.static concurrent_queue<T> member如果您的设计需要,您甚至可以制作.

缺点是多线程同步与其他类型的工作相结合.但是当你有concurrent_queue时 - 所有同步都被隔离在该原语中,并且生成/使用数据的代码不会受到锁和等待的污染:

concurrent_queue<int> c;
thread producer([&]
{
    for(int i=0;i!=100;++i)
        c.push(i);
});
thread consumer([&]
{
    int x;
    do{
        c.pop(x);
        std::cout << x << std::endl;
    }while(x!=11);
});
producer.join();
consumer.join();

如您所见,没有"手动"同步push/pop,代码更清晰.

此外,当您以这种方式分离组件时 - 您可以单独测试它们.而且,它们变得更加可重用.