C++ Boost Lockfree超详细讲解使用方法

来源：jb51　　时间：2022/11/23 18:50:13　　对本文有异议

一、说明

Boost.Lockfree 提供线程安全和无锁容器。可以从多个线程访问此库中的容器，而无需同步访问。

在 1.56.0 版本中，Boost.Lockfree 只提供了两个容器：boost::lockfree::queue 类型的队列和 boost::lockfree::stack 类型的栈。对于队列，可以使用第二个实现：boost::lockfree::spsc_queue。此类针对只有一个线程写入队列和只有一个线程从队列读取的用例进行了优化。类名中的缩写 spsc 代表单一生产者/单一消费者。

二、示例和代码

示例 46.1。使用 boost::lockfree::spsc_queue

#include <boost/lockfree/spsc_queue.hpp>
#include <thread>
#include <iostream>
boost::lockfree::spsc_queue<int> q{100};
int sum = 0;
void produce()
{
  for (int i = 1; i <= 100; ++i)
    q.push(i);
}
void consume()
{
  int i;
  while (q.pop(i))
    sum += i;
}
int main()
{
  std::thread t1{produce};
  std::thread t2{consume};
  t1.join();
  t2.join();
  consume();
  std::cout << sum << '\n';
}

Example46.1

示例 46.1 使用容器 boost::lockfree::spsc_queue。第一个执行函数 produce() 的线程将数字 1 到 100 添加到容器中。第二个线程执行 consume()，从容器中读取数字并将它们相加。因为容器 boost::lockfree::spsc_queue 明确支持来自两个线程的并发访问，所以不需要同步线程。

请注意，函数 consume() 会在线程终止后被第二次调用。这需要计算所有 100 个数字的总数，即 5050。因为 consume() 在循环中访问队列，它读取数字的速度可能比 produce() 插入数字的速度快。如果队列为空，pop() 返回 false。因此，执行 consume() 的线程可能会终止，因为另一个线程中的 produce() 无法足够快地填充队列。如果执行 produce() 的线程终止，那么很明显所有数字都已添加到队列中。第二次调用 consume() 确保将可能尚未读取的数字添加到 sum 中。

队列的大小被传递给构造函数。因为 boost::lockfree::spsc_queue 是用循环缓冲区实现的，所以示例 46.1 中的队列容量为 100 个元素。如果由于队列已满而无法添加值，则 push() 返回 false。该示例不检查 push() 的返回值，因为恰好有 100 个数字被添加到队列中。因此，100 个元素就足够了。

示例 46.2。 boost::lockfree::spsc_queue 和 boost::lockfree::capacity

#include <boost/lockfree/spsc_queue.hpp>
#include <boost/lockfree/policies.hpp>
#include <thread>
#include <iostream>
using namespace boost::lockfree;
spsc_queue<int, capacity<100>> q;
int sum = 0;
void produce()
{
  for (int i = 1; i <= 100; ++i)
    q.push(i);
}
void consume()
{
  while (q.consume_one([](int i){ sum += i; }))
    ;
}
int main()
{
  std::thread t1{produce};
  std::thread t2{consume};
  t1.join();
  t2.join();
  q.consume_all([](int i){ sum += i; });
  std::cout << sum << '\n';
}

Example46.2

示例 46.2 与前面的示例类似，但这次循环缓冲区的大小是在编译时设置的。这是通过模板 boost::lockfree::capacity 完成的，它需要容量作为模板参数。 q 是用默认构造函数实例化的——容量不能在运行时设置。

函数 consume() 已更改为使用 consume_one() 而不是 pop() 来读取数字。 lambda 函数作为参数传递给 consume_one()。 consume_one() 就像 pop() 一样读取一个数字，但该数字不是通过对调用者的引用返回的。它作为唯一参数传递给 lambda 函数。

当线程终止时，main() 调用成员函数 consume_all()，而不是 consume()。 consume_all() 的工作方式与 consume_one() 类似，但要确保队列在调用后为空。只要队列中有元素，consume_all() 就会调用 lambda 函数。

示例 46.2 再次将 5050 写入标准输出。

示例 46.3。具有可变容器大小的 boost::lockfree::queue

#include <boost/lockfree/queue.hpp>
#include <thread>
#include <atomic>
#include <iostream>
boost::lockfree::queue<int> q{100};
std::atomic<int> sum{0};
void produce()
{
  for (int i = 1; i <= 10000; ++i)
    q.push(i);
}
void consume()
{
  int i;
  while (q.pop(i))
    sum += i;
}
int main()
{
  std::thread t1{produce};
  std::thread t2{consume};
  std::thread t3{consume};
  t1.join();
  t2.join();
  t3.join();
  consume();
  std::cout << sum << '\n';
}

Example46.3

示例 46.3 在两个线程中执行 consume()。因为有多个线程从队列中读取，所以不得使用类 boost::lockfree::spsc_queue。此示例改为使用 boost::lockfree::queue。

多亏了 std::atomic，对变量 sum 的访问现在也是线程安全的。

队列的大小设置为 100——这是传递给构造函数的参数。但是，这只是初始大小。默认情况下，boost::lockfree::queue 不使用循环缓冲区实现。如果添加到队列中的项目多于设置的容量，则会自动增加。如果初始大小不够，boost::lockfree::queue 会动态分配额外的内存。

这意味着 boost::lockfree::queue 不一定是无锁的。 boost::lockfree::queue 默认使用的分配器是 boost::lockfree::allocator，它基于 std::allocator。因此，此分配器确定 boost::lockfree::queue 是否是无约束的无锁。

#include <boost/lockfree/queue.hpp>
#include <thread>
#include <atomic>
#include <iostream>
using namespace boost::lockfree;
queue<int, fixed_sized<true>> q{10000};
std::atomic<int> sum{0};
void produce()
{
  for (int i = 1; i <= 10000; ++i)
    q.push(i);
}
void consume()
{
  int i;
  while (q.pop(i))
    sum += i;
}
int main()
{
  std::thread t1{produce};
  std::thread t2{consume};
  std::thread t3{consume};
  t1.join();
  t2.join();
  t3.join();
  consume();
  std::cout << sum << '\n';
}