最快实现单线程提供数据，多线程消费数据

Posted 2023-02-16

【中文标题】最快实现单线程提供数据，多线程消费数据

【英文标题】：Fastest implementation of one thread providing data, many threads consuming data

【发布时间】：2009-10-13 22:53:39

【问题描述】：

我有很多数据想要分发给许多不同的线程。此数据来自单个线程。消费线程可以同时安全地访问容器。

数据需要每隔 delta 秒（50ms

我正在使用 linux（平台特定的解决方案非常好/预期），我关心每一毫秒。我应该使用哪种锁定机制，或者是否有更好的模型来解决这个问题？

【问题讨论】：

你应该更清楚地指定事情。所有线程都获得相同的数据，还是先到先得？如果是后者，你想如何定义顺序？让调度器做它的事，还是你想要更多的控制权？最后，非阻塞是强制性的吗？如果是这样，你就有点被自旋锁和类似的结构所困。

在这里使用自旋锁可能是个坏主意。如果锁是高度竞争的（听起来确实如此），自旋锁会遭受严重的性能下降，通常执行比读/写锁差几个数量级。基本上，竞争线程会占用 CPU 旋转并用锁使线程饿死。使用带有退避机制的自旋锁有时可以帮助解决这个问题。

所有线程访问相同的数据，那里的优先级没有问题。读取线程响应某些事件而唤醒，需要立即响应。如果他们有最新的数据，他们会做出最好的决定，但是更新容器对象可能需要 10 毫秒，在此过程中，使用陈旧数据进行响应比等待更好。所以我想他们试图获得一个读锁，并在成功时更新他们的状态，在失败时仍然响应，只是使用陈旧的数据。唯一可以阻塞的线程是等待更新容器时的写入线程。

【参考方案1】：

如果只有一个数据生产者线程并且不考虑内存，您可能需要考虑使用合并和交换算法。

在其中，写入者线程创建数据结构的副本，而读取者继续使用原始数据结构，合并新的更改，然后在互斥锁或临界区（或读取器/写入器锁）中执行两个结构的交换.如果您的Unix platform supports interlocked exchange as an atomic operation，您可以执行无锁交换，通过它们实现最大化读取吞吐量。

【参考方案2】：

看来您需要使用 pthread 读/写锁。它们允许您限制对一个作者或多个读者的访问。看pthread_rwlock_init初始化锁，pthread_rwlock_rdlock获取读数据锁，pthread_rwlock_wrlock获取写数据锁。

【参考方案3】：

听起来像 pthread 读写锁以及一些线程安全队列的一个很好的用途。生产者线程将项目插入队列。工作池将从队列中拉出项目并处理数据。我不确定输出将如何工作，但您可能也想在这里使用线程安全队列......如果有意义的话，也许可以使用优先级队列来自动合并数据。

锁定的队列结构只不过是一个用于排他锁定的互斥体，一个用于数据存储的std::queue，以及一个用于唤醒等待队列的线程的条件变量。 enqueue 方法获取锁，插入队列，释放锁，并发出条件信号。 dequeue 方法获取互斥体，使用互斥体作为守卫等待条件，并在唤醒时将那里的所有数据出列。这是一个非常标准的生产者-消费者样式队列。

在推出自己的解决方案之前，您可能需要查看Boost.MPI 和Boost.Thread。它们都在底层操作系统实现上提供了更好的 C++ 接口。我经常使用Boost.Thread，但它没有提供好的消息传递接口，但它确实比 pthread 有所改进。

如果您真的很喜欢多处理，您可能需要认真考虑Boost.MPI 或Apache Qpid。我计划在未来的项目中研究Qpid 和AMPQ，因为它们都提供了基于消息的良好界面。