0

我想到了。我犯了一个愚蠢的错误，我实际上并没有从队列中删除元素，我只是在读取第一个元素。我修改了代码，下面的代码不起作用。谢谢大家的帮助。

我正在尝试使用 boost 来实现生产者消费者问题，这实际上是一个更大项目的一部分。我已经从互联网上的示例中实现了一个程序，甚至还有一些我在这里找到的帮助。但是目前我的代码只是挂起。基于一些好的建议，我决定使用 boost ciruclar 缓冲区在生产者和消费者之间保存我的数据。那里有很多类似的代码，我能够从中汇集想法并自己编写一些东西。但是，我似乎仍然遇到与以前相同的问题（这是我的程序只是挂起）。我以为我没有像以前那样犯同样的错误..

我的代码在下面给出，我已经取出了我之前的代码，我只是我自己的单链接列表。

缓冲区头：

#ifndef PCDBUFFER_H
#define PCDBUFFER_H

#include <pcl/io/pcd_io.h>
#include <boost/thread/mutex.hpp>
#include <boost/thread/condition.hpp>
#include <boost/circular_buffer.hpp>

class pcdBuffer
{
    public:
        pcdBuffer(int buffSize);
        void put(int data);
        int get();
        bool isFull();
        bool isEmpty();
        int getSize();
        int getCapacity();
    private:
        boost::mutex bmutex;
        boost::condition_variable buffEmpty;
        boost::condition_variable buffFull;
        boost::circular_buffer<int> buffer;
};


#endif

缓冲源（仅相关部分）：

#include "pcdBuffer.h"
#include <iostream>

//boost::mutex io_mutex;

pcdBuffer::pcdBuffer(int buffSize)
{
    buffer.set_capacity(buffSize);
}

void pcdBuffer::put(int data)
{
    {
        boost::mutex::scoped_lock buffLock(bmutex);
        while(buffer.full())
        {
            std::cout << "Buffer is full" << std::endl;
            buffFull.wait(buffLock);
        }
        buffer.push_back(data);
    }
    buffEmpty.notify_one();
}

int pcdBuffer::get()
{
    int data;
    {
        boost::mutex::scoped_lock buffLock(bmutex);
        while(buffer.empty())
        {
            std::cout << "Buffer is empty" << std::endl;
            buffEmpty.wait(buffLock);
        }
        data = buffer.front();
            buffer.pop_front();
    }
    buffFull.notify_one();
    return data;
}

代码的主要驱动程序：

#include <iostream>
#include <boost/thread/mutex.hpp>
#include <boost/thread/thread.hpp>
#include <unistd.h>
#include "pcdBuffer.h"

pcdBuffer buff(100);

void producer()
{
    int i = 10;
    while (true)
    {
        buff.put(i);
        i++;
    }
}

void consumer()
{
    int i;
    while(true)
    {
        i = buff.get();
        std::cout << "Data: " << i << std::endl;
    }
}

int main(int argc, char** argv)
{
    std::cout << "Starting main...." << std::endl;
    std::cout << "Buffer Details: " << std::endl;
    std::cout << "Capacity: " << buff.getCapacity() << ", isEmpty: " << buff.isEmpty() << ", isFull: " << buff.isFull() << std::endl;
    boost::thread cons(consumer);
    sleep(5);
    boost::thread prod(producer);
    prod.join();
    cons.join();
    return 0;
}

我的缓冲区容量已正确初始化为 100。消费者线程等待并报告“缓冲区为空”5 秒，但之后我从 put 方法和“数据：10”中获取“缓冲区已满”消费者函数在标准输出上交替。如您所见，10 是我放入的第一个元素。似乎缓冲区已满，并没有通知消费者，但我检查了我的锁，并认为它们是正确的。非常感谢您对此的任何帮助。

这是我编写此代码的参考链接：

http://www.boost.org/doc/libs/1_49_0/libs/circular_buffer/doc/circular_buffer.html#classboost_1_1circular__buffer_19ba12c0142a21a7d960877c22fa3ea00

http://www.drdobbs.com/article/print?articleId=184401518&siteSectionName=cpp

循环缓冲区的线程安全实现

4

3 回答 3

5

首先，与其编写自己的列表，不如直接换行而不是编写自己的std::list列表pcdQueue。这是正确的，这std::list不是线程安全的原样，但无论如何你都在你的类中提供了必要的同步原语。

您的程序挂起的原因：您保留锁定并填充队列直到它已满。您通过消费者的通知notify_one是无用的，因为您的消费者将再次锁定，因为互斥锁已被占用（由生产者中的锁定）。

当您最终通过等待释放锁（当队列已满）时 condition_variable，您不会唤醒您的消费者，因此您的消费者和生产者都被阻塞并且您的程序挂起。

将其更改为：

void pcdQueue::produce()
{
    int i=0;
    while(true)
    {
        {
            boost::mutex::scoped_lock lock(qmutex);
            while( ! qlen < buffSize ) {
                std::cout << "Queue is full" << std::endl;
                full.wait(lock);
            }

            enqueue(i); // or myList.push_back(i) if you switch to std::list
        }

        empty.notify_one();


    }
}

你的consume()方法也有同样的问题。将其更改为：

pcdFrame* pcdQueue::consume()
{
    pcdFrame *frame;

    {
        boost::mutex::scoped_lock lock(qmutex);
        while( qlen == 0 ) {
            std::cout << "Queue is empty" << std::endl;
            empty.wait(lock);
        }

        frame = dequeue();
    }
    full.notify_one();

    return frame;
}

通常，请注意并注意通知仅在有人等待时才有效。否则，他们就会“迷失”。此外，请注意，您不需要在调用时保持互斥锁锁定notify_one（实际上，这可能会导致额外的上下文切换开销，因为您唤醒了另一个线程，然后该线程将等待当前（仍然）的互斥锁锁定（由你）。所以首先，释放互斥锁，然后告诉另一个线程继续。

请注意，两个线程都是无限运行的，所以你的主程序仍然会在第一次挂起join()并且永远不会退出。您可以在 -loops 中包含一个停止标志，while以告诉您的线程完成。

于 2012-04-17T12:05:37.883 回答
0

Boost 现在在它的 lockfree 部分提供了一种生产者/消费者队列类型，虽然它可以在队列填满时锁定，但它基本上是无锁的。

您可以在此处找到文档：

http://www.boost.org/doc/libs/1_54_0/doc/html/lockfree.html

这是一个相当新的添加，所以我认为它还不是许多标准包的一部分。另一方面，这看起来主要是标题，所有这些都依赖于相当低级的系统东西，这些东西已经被提升了很长时间。

于 2013-08-09T03:39:34.530 回答
0

我遇到了同样的悬挂问题。多亏了 Johannes 的回答，我才能让它完全发挥作用。但是，我不得不使用 full.wait_for(lock, boost::chrono::milliseconds(100)); 在使用非常短的循环缓冲区（3-4 个元素）时，在消费者和生产者中防止挂起。

另外，我用 while (buffer->empty() 代替了 while (true)。

最后，一切都可靠且快速地运行。

于 2014-01-27T19:04:39.883 回答