AI 助理
备案控制台
开发者社区 开发与运维 文章 正文

4.7 C++ Boost 多线程并发库

2023-08-21 187
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： C++语言并没有对多线程与网络的良好支持，虽然新的C++标准加入了基本的`thread`库，但是对于并发编程的支持仍然很基础，Boost库提供了数个用于实现高并发与网络相关的开发库这让我们在开发跨平台并发网络应用时能够像Java等语言一样高效开发。thread库为C++增加了多线程处理能力，其主要提供了清晰的，互斥量，线程，条件变量等，可以很容易的实现多线程应用开发，而且该库是可跨平台的，并且支持`POSIX`和`Windows`线程。

Boost 库是一个由C/C++语言的开发者创建并更新维护的开源类库，其提供了许多功能强大的程序库和工具，用于开发高质量、可移植、高效的C应用程序。Boost库可以作为标准C库的后备，通常被称为准标准库，是C标准化进程的重要开发引擎之一。使用Boost库可以加速C应用程序的开发过程，提高代码质量和性能，并且可以适用于多种不同的系统平台和编译器。Boost库已被广泛应用于许多不同领域的C++应用程序开发中，如网络应用程序、图像处理、数值计算、多线程应用程序和文件系统处理等。

C++语言并没有对多线程与网络的良好支持，虽然新的C++标准加入了基本的thread库，但是对于并发编程的支持仍然很基础，Boost库提供了数个用于实现高并发与网络相关的开发库这让我们在开发跨平台并发网络应用时能够像Java等语言一样高效开发。

thread库为C++增加了多线程处理能力，其主要提供了清晰的，互斥量，线程，条件变量等，可以很容易的实现多线程应用开发，而且该库是可跨平台的，并且支持POSIX和Windows线程。

7.1 互斥锁

互斥锁通过在访问共享资源的线程之间进行通信来避免并发问题。互斥锁仅允许一个线程在任何给定时间点上访问共享资源。如果已经有一个线程锁定了互斥锁，则任何其他线程都必须等待锁被释放。一旦锁被释放，等待队列中的一个线程将被允许继续其工作。

Boost库中的 boost::mutex 类型表示一个互斥锁。它提供了两个主要函数来控制互斥锁：lock() 和 unlock()。当一个线程想要访问一个共享资源时，它会调用互斥锁的 lock() 函数来获取锁，如果无法获得，线程将最多等待直到锁被释放。在线程访问完共享资源后，它需要调用 unlock() 函数来释放锁，以便其他线程可以获得锁并访问共享资源。

互斥体是用于线程同步的一种手段，其主要用于在多线程环境下，防止多个线程同时操作共享资源，当某线程被锁，其他线程则需要等待它解锁后才能继续访问共享资源。

  • thread提供了6种互斥类型,但常用的只有3种:
    • mutex 独占互斥锁
    • recursive_mutex 递归互斥锁
    • shared_mutex 读写锁

通常我们会使用Mutex来保护共享资源，防止在多线程环境中数据的不一致性，当一个资源被锁定，其他线程只能阻塞等待释放后才可继续操作。

#define BOOST_THREAD_VERSION 5
#include <iostream>
#include <boost/thread/thread_guard.hpp>

using namespace std;
using namespace boost;

// 最基本的互斥锁
void MutexA()
{
   
  boost::mutex mutex;
  try
  {
   
    mutex.lock();
    // 执行变量存取操作
    mutex.unlock();
  }
  catch (...)
  {
   
    mutex.unlock();
  }
}

// 智能互斥锁(无需加解锁)
void MutexB()
{
   
  boost::mutex mutex;
  boost::lock_guard<boost::mutex> global_mutex(mutex);
  // 只需要定义以上代码即可
}

在Boost中创建多线程非常简单，只需要定义一个MyThread线程函数，并在主函数中开启线程即可实现。

#include <iostream>
#include <boost/thread/thread.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/bind.hpp>

using namespace std;

boost::mutex io_mutex;

void MyThread(int id)
{
   
  for (int i = 1; i < 10; ++i)
  {
   
    // 定义智能互斥锁,防止出现输出乱序
    boost::mutex::scoped_lock lock(io_mutex);
    std::cout << id << ": " << i << std::endl;
  }
}

int main(int argc, char *argv[])
{
   
  boost::thread thrd1(boost::bind(&MyThread, 1));
  boost::thread thrd2(boost::bind(&MyThread, 2));

  //中断线程
  thrd1.interrupt();

  //获取线程ID
  cout << "线程ID:" << thrd1.get_id() << endl;

  // 等待线程
  thrd1.join();
  thrd2.join();

  //超过3s结束线程
  thrd1.timed_join(boost::posix_time::seconds(3));
  thrd2.timed_join(boost::posix_time::seconds(3));

  std::system("pause");
  return 0;
}

7.2 线程局部存储

Boost库中提供了线程局部存储（Thread Local Storage，简称TLS）的支持，可以让程序中的每个线程都拥有独立的数据空间，互相之间不会受到干扰。这对于一些线程之间需要共享数据，但需要保证数据安全的场景非常有用，例如线程池等。

有时候函数使用了局部静态变量或全局变量，导致无法用于多线程环境，因为无法保证变量在多线程环境下重入的正确操作。

#include <iostream>
#include <boost/thread/thread.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/thread/tss.hpp>

using namespace std;

// 定义一个全局互斥体
boost::mutex io_mutex;

// 线程本地存储一个整数,声明
boost::thread_specific_ptr<int> ptr;

struct MyThread
{
   
  MyThread(int id) :id(id){
   }

  void operator()()
  {
   
    // 如果ptr内部为0则说明没有,我们就初始化为0
    if (ptr.get() == 0)
      ptr.reset(new int(0));

    for (int x = 0; x <10; ++x)
    {
   
      // 往自己的线程上加
      (*ptr)++;
      boost::mutex::scoped_lock lock(io_mutex);
      std::cout << "当前ID: " << id << " 本地存储数值: " << *ptr << std::endl;
    }
  }

public:
  int id;
};

int main(int argc, char *argv[])
{
   
  boost::thread thrd1(MyThread(1));
  boost::thread thrd2(MyThread(2));

  thrd1.join();
  thrd2.join();

  std::system("pause");
  return 0;
}

如果本地存储的类型是一个结构体，如下定义了MyStruct本地结构体，来实现本地数据累加。

#include <iostream>
#include <string>
#include <boost/thread/thread.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/thread/tss.hpp>

using namespace std;

// 定义一个全局互斥体
boost::mutex io_mutex;

// 定义本地存储结构体
typedef struct MyStruct
{
   
  int uid;
  std::string uname;

  MyStruct(int x, std::string y)
  {
   
    uid = x;
    uname = y;
  }
}MyStruct;

// 线程本地存储一个整数,声明
boost::thread_specific_ptr<MyStruct> ptr;

struct MyThread
{
   
  MyThread(int id) :id(id){
   }
  void operator()()
  {
   
    // 如果ptr内部为0则说明没有,我们就初始化为0
    if (ptr.get() == 0)
      ptr.reset(new MyStruct(0,"lyshark"));

    for (int x = 0; x <10; ++x)
    {
   
      // 往自己的线程上加
      (*ptr).uid = (*ptr).uid + 1;
      (*ptr).uname = "lyshark";
      boost::mutex::scoped_lock lock(io_mutex);
      std::cout << "当前ID: " << id << " 本地存储数值: " << (*ptr).uid << "本地存储名字: " << (*ptr).uname << std::endl;
    }
  }
public:
  int id;
};

int main(int argc, char *argv[])
{
   
  boost::thread thrd1(MyThread(1));
  boost::thread thrd2(MyThread(2));

  thrd1.join();
  thrd2.join();

  std::system("pause");
  return 0;
}

7.3 使用线程组

线程组thread_group用于管理一组线程，就像线程池一样，其内部使用了std::list<thread*>来容纳每个线程对象。

当需要创建新线程时，使用create_thread()工厂函数，并通过bind绑定传递参数即可实现创建，如下是最简单的线程组创建。

#include <iostream>
#include <boost/thread.hpp>
#include <boost/function.hpp>
#include <boost/bind.hpp>

using namespace std;

boost::mutex io_mutex;

void MyThread(int x, string str)
{
   
  try
  {
   
    // 延时2秒
    boost::this_thread::sleep(boost::posix_time::seconds(2));
    for (int i = 0; i < x; i++)
    {
   
      boost::mutex::scoped_lock lock(io_mutex);
      cout << "输出字符串: " << str << " 计次: " << i << endl;
    }
  }
  catch (boost::thread_interrupted&)
  {
   
    cout << "thread is interrupt" << endl;
  }
}

int main(int argc,char *argv[])
{
   
  // 定义线程组
  boost::thread_group group;

  for (int x = 0; x < 10; x++)
  {
   
    // 创建新线程
    group.create_thread(boost::bind(MyThread, x, "hello lyshark"));
  }

  cout << "当前线程数量: " << group.size() << endl;
  group.join_all();

  std::system("pause");
  return 0;
}

我们还可以通过add_thread和remove_thread将特定的线程对象放入到不同的线程组中，来实现对线程的批量操作。

#include <iostream>
#include <string>
#include <boost/thread.hpp>
#include <boost/function.hpp>
#include <boost/bind.hpp>

using namespace std;

typedef struct MyStruct
{
   
  int uuid;
  std::string uname;
}MyStruct;

boost::mutex io_mutex;

void MyThread(MyStruct ptr)
{
   
  try
  {
   
    for (int i = 0; i < 5; i++)
    {
   
      boost::mutex::scoped_lock lock(io_mutex);
      cout << "UUID: " << ptr.uuid << " UName: " << ptr.uname << endl;
    }
  }
  catch (boost::thread_interrupted&)
  {
   
    cout << "thread is interrupt" << endl;
  }
}

int main(int argc,char *argv[])
{
   
  MyStruct my_struct;
  boost::thread_group group;

  // 创建线程并赋值
  my_struct.uuid = 1001;
  my_struct.uname = "lyshark";
  boost::thread thrd1(&MyThread,my_struct);

  my_struct.uuid = 1002;
  my_struct.uname = "admin";
  boost::thread thrd2(&MyThread, my_struct);

  // 将线程加入线程组
  group.add_thread(&thrd1);
  group.add_thread(&thrd2);

  // 中断所有线程
  // group.interrupt_all();

  // 判断thrd1是否在组内
  bool is_in = group.is_thread_in(&thrd1);
  std::cout << "是否在组内: " << is_in << std::endl;

  // 移除线程组
  group.remove_thread(&thrd1);
  group.remove_thread(&thrd2);

  // 等待线程组执行结束
  group.join_all();
  boost::this_thread::sleep(boost::posix_time::seconds(2));

  std::system("pause");
  return 0;
}

7.4 获取线程返回值

获取线程返回值，需要使用异步的方式得到，Boost中提供了ASIO库来实现异步操作，该库采用了前摄器设计模式，实现了可移植的异步IO操作。

首先来简单的看一下，如何使用异步的方式实现创建线程的。

#define BOOST_THREAD_VERSION 5
#include <iostream>
#include <boost/asio.hpp>
#include <boost/bind.hpp>
#include <boost/thread.hpp>
#include <boost/function.hpp>

using namespace std;

void MyThread(int x)
{
   
  for (int i = 0; i < x; i++)
    std::cout << i << std::endl;
}

int main(int argc, char *argv[])
{
   
  // 第一种使用方式
  auto x = async(&MyThread, 10);
  x.wait();

  // 直接通过bind绑定参数
  async(boost::bind(MyThread, 20));

  // 直接使用lambda表达式
  auto y = boost::async([]
  {
   
    cout << "hello lyshark" << endl; 
  });
  y.wait();

  std::system("pause");
  return 0;
}

当我们需要获取单个线程的返回值时，可以使用valid()方法或使用get()将返回值从线程里拉取出来。

#define BOOST_THREAD_VERSION 5
#include <iostream>
#include <boost/bind.hpp>
#include <boost/thread.hpp>
#include <boost/function.hpp>

using namespace std;

int MyThread(int x, int y)
{
   
  Sleep(3000);
  return x + y;
}

typedef struct
{
   
  int x;
  int y;
}MyStruct;

MyStruct MyThreadStruct(int x, int y)
{
   
  MyStruct ref;
  ref.x = x + 100;
  ref.y = y + 100;

  return ref;
}

int main(int argc, char *argv[])
{
   
  // 返回数值直接使用get得到
  auto f = boost::async(boost::bind(MyThread, 10, 20));
  // auto f = boost::async(boost::launch::async,bind(MyThread, 10, 20));

  if (f.valid())
  {
   
    cout << "获取计算结果: " << f.get() << endl;
  }
  f.wait();

  // 返回参数是结构体
  auto t = boost::async(boost::bind(MyThreadStruct, 100, 200));
  MyStruct tmp = t.get();
  cout << "获取结构体参数A: " << tmp.x << " 参数B: " << tmp.y << endl;
  t.wait();

  std::system("pause");
  return 0;
}

有时候我们会一次性创建多个线程共同执行，此时想要获取到每个线程中的返回值，那么就需要使用多个future对象，代码如下。

#define BOOST_THREAD_VERSION 5
#include <iostream>
#include <boost/bind.hpp>
#include <boost/thread.hpp>
#include <boost/function.hpp>

using namespace std;

int MyThread(int x, int y)
{
   
  Sleep(3000);
  return x + y;
}

int main(int argc, char *argv[])
{
   
  std::vector<boost::future<int>> vect;

  // 启动异步线程,并放入vector容器中
  for (int i = 0; i < 100; ++i)
  {
   
    vect.push_back(boost::async(bind(MyThread, i, i * 10)));
  }

  // 等待所有线程计算结束
  boost::wait_for_all(vect.begin(), vect.end());
  for (auto &x : vect)
  {
   
    // 获取到返回值
    if (x.valid())
    {
   
      cout << "线程计算结果: " << x.get() << endl;
    }
  }

  std::system("pause");
  return 0;
}

返回数值类型如果不够存储的话，那么我们可以定义一个MyStruct结构体，通过结构体传递参数，并将计算结果返回为结构体类型。

#define BOOST_THREAD_VERSION 5
#include <iostream>
#include <boost/bind.hpp>
#include <boost/thread.hpp>
#include <boost/function.hpp>

using namespace std;

typedef struct
{
   
  int x;
  int y;
}MyStruct;

// 定义一个返回结构体的函数
MyStruct MyThreadStruct(int x, int y)
{
   
  MyStruct ref;
  ref.x = x + 100;
  ref.y = y + 100;

  return ref;
}

int main(int argc, char *argv[])
{
   
  std::vector<boost::future<MyStruct>> vec;
  for (int x = 0; x < 100; x++)
  {
   
    vec.push_back(boost::async(bind(MyThreadStruct, x, x + 10)));
  }

  boost::wait_for_all(vec.begin(), vec.end());
  for (auto &x : vec)
  {
   
    MyStruct tmp = x.get();
    cout << "获取计算结果A: " << tmp.x << " 获取结果B: " << tmp.y << endl;
  }

  std::system("pause");
  return 0;
}

由于future只能get获取一次数据，使得它不能被多线程并发访问，所以就出现了shared_future，它是future的增强，可以线程安全的多次调用get()获取到计算结果，修改很简单只需要将声明改一下，其他的不用动。

void Async()
{
   
  std::vector<boost::shared_future<MyStruct>> vec;
  for (int x = 0; x < 100; x++)
  {
   
    vec.push_back(boost::async(bind(MyThreadStruct, x, x + 10)).share());
  }

  boost::wait_for_all(vec.begin(), vec.end());
  for (auto &x : vec)
  {
   
    MyStruct tmp = x.get();
    cout << "获取计算结果: " << tmp.x << endl;
  }
}

7.5 共享锁

shared_mutex（共享互斥锁）是 C++11 标准库中引入的一种线程同步机制，可以实现同时有多个线程同时读取共享资源，但只能有一个线程写入共享资源的机制。与常见的互斥锁不同，shared_mutex 具有更加细致的控制对共享资源的访问权限。

该锁允许线程获取多个共享所有权和一个专享所有权，实现了读写锁机制，即多个读线程一个写线程。

#define BOOST_THREAD_VERSION 5
#include <iostream>
#include <boost/bind.hpp>
#include <boost/thread.hpp>
#include <boost/function.hpp>

using namespace std;
using namespace boost;

class MyClass
{
   
private:
  // 读写的数据
  int m_x;

  // 定义共享互斥量
  boost::shared_mutex rw_mutex;

public:
  MyClass()
  {
   
    m_x = 0;
  }

  // 写数据
  void write()
  {
   
    //写锁定
    boost::unique_lock<boost::shared_mutex> g(rw_mutex);
    ++m_x;
  }

  // 读数据
  void read(int *x)
  {
   
    // 读锁定
    boost::shared_lock<boost::shared_mutex> g(rw_mutex);
    *x = m_x;
  }
};

// 定义写函数,每次调用都会写入十次
void writer(MyClass &ptr)
{
   
  for (int x = 0; x < 10; x++)
  {
   
    ptr.write();
  }
}

// 定义读函数,每次调用读取十次
void reader(MyClass &ptr)
{
   
  int item;
  for (int x = 0; x < 10; x++)
  {
   
    ptr.read(&item);
    std::cout << "读取数据: " << item << std::endl;
  }
}

int main(int argc, char *argv[])
{
   
  MyClass ptr;
  thread_group pool;

  // 定义2个读
  pool.create_thread(boost::bind(reader, boost::ref(ptr)));
  pool.create_thread(boost::bind(reader, boost::ref(ptr)));

  // 定义1个写
  pool.create_thread(boost::bind(writer, boost::ref(ptr)));

  pool.join_all();

  std::system("pause");
  return 0;
}

7.6 获取线程ID号

实现线程池，每次将一个线程service_io存入到栈中，需要时从栈中弹出并调用内部相应的函数。

#include <stack>
#include <iostream>
#include <boost/asio.hpp>
#include <boost/bind.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/thread/thread.hpp>

using namespace std;
using namespace boost;

class ThreadPool
{
   
  static int count;
  int NoOfThread;
  int counter;
  thread_group group;
  boost::mutex mutex_;
  asio::io_service io_service;
  stack<boost::thread*> thread_stack;

public:
  ThreadPool(int num)
  {
   
    NoOfThread = num;
    counter = 0;
    boost::mutex::scoped_lock lock(mutex_);

    if (count == 0)
      count++;
    else
      return;

    // 开辟线程并放入栈中存储
    for (int i = 0; i<num; ++i)
    {
   
      thread_stack.push(group.create_thread(boost::bind(&asio::io_service::run, &io_service)));
    }
  }
  ~ThreadPool()
  {
   
    io_service.stop();
    group.join_all();
  }

  // 从栈中弹出元素
  boost::thread* get_thread()
  {
   
    // 判断成立说明不存在线程
    if (counter > NoOfThread)
      return NULL;

    // 否则从栈中弹出
    counter++;
    boost::thread* ptr = thread_stack.top();
    thread_stack.pop();
    return ptr;
  }
  // 获取到元素计数器
  int get_number()
  {
   
    return counter;
  }
};

int ThreadPool::count = 0;

int main(int argc, char * argv[])
{
   
  // 定义启动10个线程
  ThreadPool pool(10);

  // 循环线程池
  for (int x = 0; x < 10; x++)
  {
   
    boost::thread* ptr = pool.get_thread();
    cout << "线程ID: " << ptr->get_id() << endl;

    int num = pool.get_number();
    std::cout << "计数器: " << num << std::endl;
  }

  std::system("pause");
  return 0;
}

本文作者： 王瑞
本文链接： https://www.lyshark.com/post/e72e3ad3.html
版权声明： 本博客所有文章除特别声明外，均采用 BY-NC-SA 许可协议。转载请注明出处！

文章标签：
C++
Java
存储
Windows
设计模式
关键词：
线程并发
C++线程
C++多线程
C++库
C++ boost
aliyun_11084
目录
相关文章
小白学大数据
|
22天前
|
数据采集 Java API
Jsoup库能处理多线程下载吗？
Jsoup库能处理多线程下载吗？
小白学大数据
25 0
光明顶阳顶天
|
1月前
|
安全
List并发线程安全问题
【10月更文挑战第21天】`List` 并发线程安全问题是多线程编程中一个非常重要的问题，需要我们认真对待和处理。只有通过不断地学习和实践，我们才能更好地掌握多线程编程的技巧和方法，提高程序的性能和稳定性。
光明顶阳顶天
198 59
最好zzz
|
1月前
|
安全 Java
线程安全的艺术：确保并发程序的正确性
在多线程环境中，确保线程安全是编程中的一个核心挑战。线程安全问题可能导致数据不一致、程序崩溃甚至安全漏洞。本文将分享如何确保线程安全，探讨不同的技术策略和最佳实践。
最好zzz
41 6
Java开发者
|
1月前
|
安全 Java 开发者
Java 多线程并发控制：深入理解与实战应用
《Java多线程并发控制：深入理解与实战应用》一书详细解析了Java多线程编程的核心概念、并发控制技术及其实战技巧，适合Java开发者深入学习和实践参考。
Java开发者
58 6
游客762btuqu5wybw666
|
1月前
|
存储 安全 Java
Java多线程编程中的并发容器：深入解析与实战应用####
在本文中，我们将探讨Java多线程编程中的一个核心话题——并发容器。不同于传统单一线程环境下的数据结构，并发容器专为多线程场景设计，确保数据访问的线程安全性和高效性。我们将从基础概念出发，逐步深入到`java.util.concurrent`包下的核心并发容器实现，如`ConcurrentHashMap`、`CopyOnWriteArrayList`以及`BlockingQueue`等，通过实例代码演示其使用方法，并分析它们背后的设计原理与适用场景。无论你是Java并发编程的初学者还是希望深化理解的开发者，本文都将为你提供有价值的见解与实践指导。 --- ####
游客762btuqu5wybw666
47 3
叫个什么名字
|
1月前
|
存储 设计模式 分布式计算
Java中的多线程编程：并发与并行的深度解析####
在当今软件开发领域，多线程编程已成为提升应用性能、响应速度及资源利用率的关键手段之一。本文将深入探讨Java平台上的多线程机制，从基础概念到高级应用，全面解析并发与并行编程的核心理念、实现方式及其在实际项目中的应用策略。不同于常规摘要的简洁概述，本文旨在通过详尽的技术剖析，为读者构建一个系统化的多线程知识框架，辅以生动实例，让抽象概念具体化，复杂问题简单化。 ####
叫个什么名字
119 4
最好zzz
|
1月前
|
缓存 安全 C++
C++无锁队列：解锁多线程编程新境界
【10月更文挑战第27天】
最好zzz
63 7
潇洒洒
|
1月前
|
消息中间件 存储 安全
深入探索C++无锁队列：多线程编程的高效利器
【10月更文挑战第19天】
潇洒洒
149 6
小王老师呀
|
2月前
|
存储 并行计算 安全
C++多线程应用
【10月更文挑战第29天】C++ 中的多线程应用广泛，常见场景包括并行计算、网络编程中的并发服务器和图形用户界面（GUI）应用。通过多线程可以显著提升计算速度和响应能力。示例代码展示了如何使用 `pthread` 库创建和管理线程。注意事项包括数据同步与互斥、线程间通信和线程安全的类设计，以确保程序的正确性和稳定性。
小王老师呀
45 5
栈江湖
|
3天前
|
NoSQL Redis
单线程传奇Redis，为何引入多线程？
Redis 4.0 引入多线程支持，主要用于后台对象删除、处理阻塞命令和网络 I/O 等操作，以提高并发性和性能。尽管如此，Redis 仍保留单线程执行模型处理客户端请求，确保高效性和简单性。多线程仅用于优化后台任务，如异步删除过期对象和分担读写操作，从而提升整体性能。
栈江湖
14 1