什么是消息队列更好？互斥体和cond或互斥体和信号量？

Question

我正在实现一个基于std ::队列的C++消息队列。

由于我需要popers等待一个空队列，我正考虑使用互斥锁进行互斥，并且cond将空闲线程挂起，就像glib与gasyncqueue一样。

但是在我看来，一个互斥信号量将会完成这项工作，我认为它包含一个整数，而且这个数字看起来像是一个非常高的数字，可以处理待处理的消息。

信号量的优点在于，每次从等待中返回时都不需要手动检查条件，因为您现在确定某人插入了某物（当某人插入了2个物品，并且您是第二个线程到达时） 。

你会选择哪一个？

编辑： 改变了问题响应@格雷格罗杰斯

Answer 1

单个信号不会做的工作 - 你需要比较（互斥+信号）和（互斥+条件变量）。

这是很容易通过努力实现它，看看这个：

void push(T t) 
{ 
    queue.push(t); 
    sem.post(); 
} 

T pop() 
{ 
    sem.wait(); 
    T t = queue.top(); 
    queue.pop(); 
    return t; 
} 

正如你可以看到有当你实际读取/写入队列，即使信号不会相互排斥，（从信号量）在那里。多个线程可以同时调用push并打断队列，或者多个线程可以同时调用pop并将其中断。或者，一个线程可能会调用pop并删除队列的第一个元素，而另一个线程则调用push。

你应该使用你认为更容易实现的任何一个，我怀疑如果有的话性能会有很大的变化（尽管这可能很有趣）。

Answer 2

如果你想允许多个用户同时在同一时间使用你的队列，你应该使用信号。

sema(10) // ten threads/process have the concurrent access. 

sema_lock(&sema_obj) 
queue 
sema_unlock(&sema_obj) 

互斥将一次“授权”一个用户。

pthread_mutex_lock(&mutex_obj) 
global_data; 
pthread_mutex_unlock(&mutex_obj) 

这是主要区别，您应该决定哪种解决方案适合您的要求。 但我会选择互斥方法，因为您不需要指定有多少用户可以获取资源。

Answer 3

个人而言，我使用互斥锁来序列化访问列表，并通过在套接字上发送一个字节（由socketpair（）生成）来唤醒用户。这可能不如信号量或条件变量有效，但它的优点是允许用户在select（）/ poll（）中阻塞。这样，除了数据队列之外，消费者还可以等待其他事情，如果需要的话。它还允许您在几乎所有操作系统上使用完全相同的排队代码，因为几乎每个操作系统都支持BSD套接字API。

伪代码如下：

// Called by the producer. Adds a data item to the queue, and sends a byte 
// on the socket to notify the consumer, if necessary 
void PushToQueue(const DataItem & di) 
{ 
    mutex.Lock(); 
    bool sendSignal = (queue.size() == 0); 
    queue.push_back(di); 
    mutex.Unlock(); 
    if (sendSignal) producerSocket.SendAByteNonBlocking(); 
} 

// Called by consumer after consumerSocket selects as ready-for-read 
// Returns true if (di) was written to, or false if there wasn't anything to read after all 
// Consumer should call this in a loop until it returns false, and then 
// go back to sleep inside select() to wait for further data from the producer 
bool PopFromQueue(DataItem & di) 
{ 
    consumerSocket.ReadAsManyBytesAsPossibleWithoutBlockingAndThrowThemAway(); 
    mutex.Lock(); 
    bool ret = (queue.size() > 0); 
    if (ret) queue.pop_front(di); 
    mutex.Unlock(); 
    return ret; 
}