4

我目前有两个线程，一个生产者和一个消费者。生产者是一种静态方法，将数据插入到 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 始终处于“停止”状态，而我相信它应该在“运行”和“停止”之间切换。加上很慢。关于我可能没有考虑过并且可能做错了什么的任何建议？

4

1 回答 1

5

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);

你的第三点中的一个问题是你建议我应该隔离整个队列，而我添加到队列的方法是静态的，并且消费者（队列阅读器）永远在一个单独的线程中运行。你能告诉我为什么这是我设计中的一个缺陷吗？

“永远运行”或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，并且代码更清晰。

此外，当您以这种方式解耦组件时 - 您可以单独测试它们。此外，它们变得更加可重用。

于 2013-04-09T22:42:35.597 回答