C++ 多进程多线程间通信

一、进程间通信

1、管道（Pipe）

2、消息队列（Message Queue）

3、共享内存（Shared Memory）

4、信号量（Semaphore）

5、套接字（Socket）

6、信号（Signal）

9、剪贴板（Clipboard）

二、线程间通信

1、共享内存（Shared Memory）

2、消息队列（Message Queue）

3、同步对象

4、原子操作

5、Future和Promise

6、管道

一、进程间通信

1、管道（Pipe）

C++管道通信的原理基于内核中的缓存机制和文件描述符。匿名管道和命名管道是两种主要的管道类型，它们在创建、使用和通信特性上有所不同。匿名管道适用于具有共同祖先的进程间通信，而命名管道则提供了更广泛的通信能力。通过管道，进程间可以高效、安全地传递数据。

1.1 匿名管道

在Linux系统中，匿名管道是通过pipe函数创建的。该函数在内核中创建一个环形队列作为缓冲区，并返回两个文件描述符，一个用于读（fd[0]），一个用于写（fd[1]）。当一个进程向管道的写端（fd[1]）写入数据时，数据实际上是被写入到内核中的缓冲区。另一个进程可以从管道的读端（fd[0]）读取这些数据，读取操作实际上是从内核缓冲区中读取数据。匿名管道是单向的，数据只能在一个方向上流动。如果需要双向通信，必须创建两个管道。匿名管道的生命周期与进程相关。当所有使用管道的文件描述符都被关闭后，管道将被销毁。

示例代码：

#include <iostream>  
#include <unistd.h> // UNIX 标准函数定义  
#include <sys/types.h>  
#include <sys/wait.h>  
#include <string.h>  int main() {  int pipefd[2]; // 文件描述符数组，pipefd[0] 是读端，pipefd[1] 是写端  pid_t pid;  char buf[1024];  const char *message = "Hello from parent";  const char *response = "Hello back from child";  // 创建管道  if (pipe(pipefd) == -1) {  perror("pipe");  return 1;  }  // 创建子进程  pid = fork();  if (pid == -1) {  perror("fork");  return 1;  }  // 子进程  if (pid == 0) {  // 关闭管道的写端  close(pipefd[1]);  // 从管道读取数据  read(pipefd[0], buf, sizeof(buf));  std::cout << "Received from parent: " << buf << std::endl;  // 关闭管道的读端  close(pipefd[0]);  // 子进程退出_exit(0);   }  // 父进程  else {  // 关闭管道的读端  close(pipefd[0]);  // 向子进程发送数据  write(pipefd[1], message, strlen(message) + 1); // 发送字符串及其终结符  // 关闭管道的写端  close(pipefd[1]);  // 等待子进程结束  wait(NULL);  }  return 0;  
}

1.2 命名管道

在Linux系统中，可以使用mkfifo命令或mkfifo函数来创建命名管道。命名管道在文件系统中有一个对应的文件名，因此可以通过文件名来访问它。任何进程都可以通过打开命名管道对应的文件来访问它。进程可以使用标准的文件操作函数（如read、write）来读写命名管道。命名管道支持双向通信和跨网络通信。多个进程可以连接到同一个命名管道进行读写操作。

父进程示例代码：

#include <iostream>  
#include <unistd.h>  
#include <sys/stat.h>  
#include <fcntl.h>  
#include <string.h>  int main() {  const char *fifo_name = "/tmp/myfifo";  const char *message = "Hello from parent";  // 创建命名管道  mkfifo(fifo_name, 0666);  // 打开命名管道以写入数据  int fd = open(fifo_name, O_WRONLY);  if (fd == -1) {  perror("open");  return 1;  }  // 写入数据到命名管道  write(fd, message, strlen(message) + 1); // 发送字符串及其终结符  // 关闭命名管道  close(fd);  // 等待子进程结束（假设有子进程正在读取这个命名管道）  // 注意：这个示例中没有创建子进程，但在实际应用中你可能需要等待  return 0;  
}

子进程示例代码：

#include <iostream>  
#include <unistd.h>  
#include <fcntl.h>  
#include <string.h>  int main() {  const char *fifo_name = "/tmp/myfifo";  char buf[1024];  // 打开命名管道以读取数据  int fd = open(fifo_name, O_RDONLY);  if (fd == -1) {  perror("open");  return 1;  }  // 从命名管道读取数据  ssize_t num_bytes = read(fd, buf, sizeof(buf) - 1);  if (num_bytes == -1) {  perror("read");  return 1;  }  buf[num_bytes] = '\0'; // 确保字符串正确终结  // 输出接收到的消息  std::cout << "Received from parent: " << buf << std::endl;  // 关闭命名管道  close(fd);  return 0;  
}

2、消息队列（Message Queue）

定义：消息队列独立于进程而存在，可以用作进程间传递数据的媒介。在大多数操作系统中，消息队列的实现依赖于内核的支持。当进程向消息队列发送消息时，这些消息会被存储在内核空间中，直到其他进程从队列中读取它们。

特点：它允许进程间异步通信，克服了管道缺点（管道只能承载无格式字节流，且缓冲区大小受限）。

以下是一个使用 POSIX 消息队列的 C++ 示例：

发送者（sender.cpp）

#include <mqueue.h>  
#include <sys/stat.h>  
#include <fcntl.h>  
#include <string.h>  
#include <iostream>  int main() {  mqd_t mqd;  const char* queueName = "/myqueue";  // 创建或打开消息队列  mqd = mq_open(queueName, O_CREAT | O_WRONLY, 0644, NULL);  if (mqd == (mqd_t)-1) {  perror("mq_open");  return 1;  }  // 发送消息  const char* message = "Hello, Message Queue!";  unsigned int priority = 0;  if (mq_send(mqd, message, strlen(message) + 1, priority) == -1) {  perror("mq_send");  mq_close(mqd);  mq_unlink(queueName);  return 1;  }  std::cout << "Message sent\n";  mq_close(mqd);  mq_unlink(queueName); // 可选：如果不再需要队列，则删除它  return 0;  
}

接收者（receiver.cpp）

#include <mqueue.h>  
#include <sys/stat.h>  
#include <fcntl.h>  
#include <iostream>  int main() {  mqd_t mqd;  const char* queueName = "/myqueue";  char buffer[256];  unsigned int priority;  // 打开消息队列  mqd = mq_open(queueName, O_CREAT | O_RDONLY, 0644, NULL);  if (mqd == (mqd_t)-1) {  perror("mq_open");  return 1;  }  // 接收消息  ssize_t bytesRead = mq_receive(mqd, buffer, sizeof(buffer), &priority);  if (bytesRead == -1) {  perror("mq_receive");  mq_close(mqd);  return 1;  }  buffer[bytesRead] = '\0'; // 确保字符串以 null 结尾  std::cout << "Received message: " << buffer << std::endl;  mq_close(mqd);  // 注意：不要在这里调用 mq_unlink，除非你想删除队列  return 0;  
}

注意事项：
确保在发送者和接收者之间正确同步消息队列的创建和删除。
消息队列的大小和消息的最大长度在创建队列时可以指定。
权限（如 0644）需要根据你的需求进行设置。

3、共享内存（Shared Memory）

定义：允许多个进程访问同一块内存区域，从而实现进程间数据共享；

特点：它是最快的进程间通信方式，避免了数据的拷贝，但共享内存需要解决并发访问和同步问题，常用的同步机制包括互斥锁（Mutexes）、信号量（Semaphores）和事件（Events）等。

在Windows系统中，使用共享内存通常涉及到CreateFileMapping和MapViewOfFile等Win32 API函数。为了在多个进程之间安全地读写共享内存，我们需要在这些进程之间实现某种形式的同步机制。

创建共享内存的生产者（写入者）

#include <windows.h>  
#include <iostream>  int main() {  // 定义共享内存的名称和大小  const char* shmName = "Global\\MySharedMemory";  const size_t shmSize = 256;  const char* mutexName = "Global\\MySharedMemoryMutex";  // 创建互斥锁  HANDLE hMutex = CreateMutex(  NULL,               // 默认安全属性  FALSE,              // 初始不拥有互斥锁  mutexName);         // 互斥锁名称  if (hMutex == NULL) {  std::cerr << "Could not create mutex (" << GetLastError() << ").\n";  return 1;  }  // 创建或打开一个文件映射对象  HANDLE hMapFile = CreateFileMapping(  INVALID_HANDLE_VALUE,    // 使用分页文件  NULL,                    // 默认安全属性  PAGE_READWRITE,          // 读写访问  0,                       // 高32位文件大小  shmSize,                 // 低32位文件大小  shmName);                // 对象名  if (hMapFile == NULL) {  std::cerr << "Could not create file mapping object (" << GetLastError() << ").\n";  return 1;  }  // 将文件映射对象映射到视图  void* pBuf = MapViewOfFile(  hMapFile,                // 文件映射对象  FILE_MAP_ALL_ACCESS,     // 读写访问  0,                       // 高32位偏移量  0,                       // 低32位偏移量  0);                      // 映射整个文件  if (pBuf == NULL) {  std::cerr << "Could not map view of file (" << GetLastError() << ").\n";  CloseHandle(hMapFile);  return 1;  } // 等待互斥锁  WaitForSingleObject(hMutex, INFINITE);  // 写入数据  std::strcpy_s(static_cast<char*>(pBuf), shmSize, "Hello, Shared Memory!");  // 释放互斥锁  ReleaseMutex(hMutex);  // 取消映射视图  UnmapViewOfFile(pBuf);   // 关闭文件映射对象句柄  CloseHandle(hMapFile);// 关闭互斥锁句柄（注意：通常在程序结束时自动关闭，但显式关闭是个好习惯）  CloseHandle(hMutex);  std::cout << "Shared memory written successfully.\n";  return 0;  
}

访问共享内存的消费者（读取者）

#include <windows.h>  
#include <iostream>  int main() {  // 定义共享内存和互斥锁的名称  const char* shmName = "Global\\MySharedMemory";  const char* mutexName = "Global\\MySharedMemoryMutex";  // 打开互斥锁  HANDLE hMutex = OpenMutex(  SYNCHRONIZE,           // 访问权限  FALSE,                 // 不更改现有所有者的所有权  mutexName);            // 互斥锁名称  if (hMutex == NULL) {  std::cerr << "Could not open mutex (" << GetLastError() << ").\n";  return 1;  }  // 打开一个现有的文件映射对象  HANDLE hMapFile = OpenFileMapping(  FILE_MAP_READ,  // 读取访问  FALSE,          // 不继承句柄  shmName);       // 对象名  if (hMapFile == NULL) {  std::cerr << "Could not open file mapping object (" << GetLastError() << ").\n";  return 1;  }  // 将文件映射对象映射到视图  void* pBuf = MapViewOfFile(  hMapFile,                // 文件映射对象  FILE_MAP_READ,           // 读取访问  0,                       // 高32位偏移量  0,                       // 低32位偏移量  0);                      // 映射整个文件  if (pBuf == NULL) {  std::cerr << "Could not map view of file (" << GetLastError() << ").\n";  CloseHandle(hMapFile);  return 1;  } // 等待互斥锁  WaitForSingleObject(hMutex, INFINITE);  // 读取数据  std::cout << "Shared memory content: " << static_cast<char*>(pBuf) << std::endl;  // 释放互斥锁  ReleaseMutex(hMutex);  // 取消映射视图  UnmapViewOfFile(pBuf);  // 关闭文件映射对象句柄  CloseHandle(hMapFile);  // 关闭互斥锁句柄  CloseHandle(hMutex);  return 0;  
}

请注意，这里的互斥锁是在全局命名空间中创建的（通过前缀"Global\\"），这意味着它可以在系统范围内的任何进程中访问。这是必需的，因为我们的目标是让多个不同的进程能够识别并访问同一个互斥锁。
此外，我们使用了WaitForSingleObject函数来等待互斥锁变得可用，并使用ReleaseMutex来释放互斥锁。这两个函数一起工作，以确保在任何给定时间只有一个进程可以访问共享内存区域。
最后，请确保在实际应用中适当地处理所有可能的错误情况，并且在不再需要时关闭所有句柄。这有助于避免资源泄漏和其他潜在问题。

4、信号量（Semaphore）

定义：它是用于多个进程对共享资源访问的同步机制；

特点：信号量的操作包括P操作和V操作，分别用于申请资源和释放资源。

以下是一个使用C++和Windows API编写的示例，该示例中创建了两个进程，一个生产者进程和一个消费者进程。生产者生产一定数量的“产品”，并通过信号量来通知消费者这些产品已经准备好被消费。

生产者示例代码

// Producer.cpp  
#include <windows.h>  
#include <iostream>  // 假设信号量的名称是固定的  
#define SEMAPHORE_NAME TEXT("Global\\MySemaphore")  int main() {  HANDLE hSemaphore = CreateSemaphore(  NULL,        // 默认安全属性  0,           // 初始计数为0，表示没有产品可用  10,          // 最大计数为10，表示最多可以有10个产品  SEMAPHORE_NAME); // 信号量名称  if (hSemaphore == NULL) {  std::cerr << "Failed to create semaphore." << std::endl;  return 1;  }  for (int i = 0; i < 5; ++i) {  // 模拟生产产品  Sleep(1000); // 假设生产一个产品需要1秒  std::cout << "Produced item " << i + 1 << std::endl;  // 增加信号量的计数，表示有一个产品已经生产好了  if (!ReleaseSemaphore(hSemaphore, 1, NULL)) {  std::cerr << "Failed to release semaphore." << std::endl;  break;  }  }  // 关闭句柄（注意：在Windows中，当进程结束时，所有句柄都会自动关闭）  // 但显式关闭是一个好习惯，特别是在大型或长期运行的程序中  // CloseHandle(hSemaphore); // 在这个例子中，可以省略，因为进程即将结束  return 0;  
}

消费者示例代码

// Consumer.cpp  
#include <windows.h>  
#include <iostream>  // 假设信号量的名称与生产者中相同  
#define SEMAPHORE_NAME TEXT("Global\\MySemaphore")  int main() {  HANDLE hSemaphore = OpenSemaphore(  SEMAPHORE_ALL_ACCESS, // 请求完全访问权限  FALSE,               // 不希望句柄被子进程继承  SEMAPHORE_NAME);     // 信号量名称  if (hSemaphore == NULL) {  std::cerr << "Failed to open semaphore." << std::endl;  return 1;  }  for (int i = 0; i < 5; ++i) {  // 等待信号量变为非零（即等待有产品可用）  if (WaitForSingleObject(hSemaphore, INFINITE) != WAIT_OBJECT_0) {  std::cerr << "Failed to wait for semaphore." << std::endl;  break;  }  // 模拟消费产品  Sleep(500); // 假设消费一个产品需要0.5秒  std::cout << "Consumed item " << i + 1 << std::endl;  }  // 关闭句柄  CloseHandle(hSemaphore);  return 0;  
}

请注意，我们使用了全局命名的信号量（通过前缀Global\），这意味着信号量在整个系统中都是可见的，可以被任何进程访问。如果你只想在同一用户会话中的进程间共享信号量，可以使用Local\前缀。
确保在生产者进程开始生产之前，消费者进程不会尝试等待信号量，否则消费者可能会立即进入等待状态，直到生产者开始生产。
在实际的应用程序中，你可能需要更复杂的错误处理和同步机制来确保程序的健壮性。
这个例子假设生产者和消费者都知道要生产/消费多少个产品。在更复杂的场景中，你可能需要其他机制来通知生产者何时停止生产或消费者何时停止消费。
在Windows中，通常不需要显式关闭句柄，因为当进程结束时，所有句柄都会被自动关闭。但是，显式关闭句柄是一个好习惯，特别是在大型或长期运行的程序中。在这个简单的例子中，我们可以省略CloseHandle调用，因为进程很快就会结束。然而，在更复杂的应用程序或库中，你应该始终关闭不再需要的。

5、套接字（Socket）

定义：它是网络通信的接口，提供了端到端的通信服务；

特点：它支持TCP/UDP等多种协议，可以根据需要选择合适的协议进行通信。

在Windows系统中，通过Winsock库实现进程间通信。首先需要包含Winsock头文件，并链接相应的库。

#include <iostream>  
#include <winsock2.h>  
#include <ws2tcpip.h>  #pragma comment(lib, "ws2_32.lib")

然后在程序开始时初始化Winsock库。

WSADATA wsaData;  
int iResult = WSAStartup(MAKEWORD(2, 2), &wsaData);  
if (iResult != 0) {  std::cerr << "WSAStartup failed with error: " << iResult << std::endl;  return 1;  
}

最后创建套接字，一个进程作为服务器（监听套接字），另一个进程作为客户端（连接套接字）。

服务器代码：

// 假设我们使用TCP套接字  
SOCKET ListenSocket = INVALID_SOCKET;  
sockaddr_in service;  // 创建套接字  
ListenSocket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);  
if (ListenSocket == INVALID_SOCKET) {  std::cerr << "socket failed with error: " << WSAGetLastError() << std::endl;  WSACleanup();  return 1;  
}  // 设置套接字地址结构  
service.sin_family = AF_INET;  
service.sin_addr.s_addr = inet_addr("127.0.0.1");  
service.sin_port = htons(12345);  // 绑定套接字  
if (bind(ListenSocket, (SOCKADDR *)&service, sizeof(service)) == SOCKET_ERROR) {  std::cerr << "bind failed with error: " << WSAGetLastError() << std::endl;  closesocket(ListenSocket);  WSACleanup();  return 1;  
}  // 监听套接字  
if (listen(ListenSocket, SOMAXCONN) == SOCKET_ERROR) {  std::cerr << "listen failed with error: " << WSAGetLastError() << std::endl;  closesocket(ListenSocket);  WSACleanup();  return 1;  
}  // 等待客户端连接（这里应该是一个循环，但为了简化，我们只接受一个连接）  
SOCKET ClientSocket = accept(ListenSocket, NULL, NULL);  
if (ClientSocket == INVALID_SOCKET) {  std::cerr << "accept failed with error: " << WSAGetLastError() << std::endl;  closesocket(ListenSocket);  WSACleanup();  return 1;  
}  // 现在你可以通过ClientSocket与客户端通信了  
// ...（发送和接收数据）  // 关闭套接字  
closesocket(ClientSocket);  
closesocket(ListenSocket);

客户端代码：

// 创建套接字  
SOCKET ConnectSocket = INVALID_SOCKET;  
sockaddr_in target;  ConnectSocket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);  
if (ConnectSocket == INVALID_SOCKET) {  std::cerr << "socket failed with error: " << WSAGetLastError() << std::endl;  WSACleanup();  return 1;  
}  target.sin_family = AF_INET;  
target.sin_addr.s_addr = inet_addr("127.0.0.1");  
target.sin_port = htons(12345);  // 连接到服务器  
if (connect(ConnectSocket, (SOCKADDR *)&target, sizeof(target)) == SOCKET_ERROR) {  std::cerr << "connect failed with error: " << WSAGetLastError() << std::endl;  closesocket(ConnectSocket);  WSACleanup();  return 1;  
}  // 现在你可以通过ConnectSocket与服务器通信了  
// ...（发送和接收数据）  // 关闭套接字  
closesocket(ConnectSocket);

6、信号（Signal）

定义：信号是Unix/Linux系统中进程间通信的一种简单方式，允许一个进程向另一个进程发送信号；信号可以由多种原因产生，包括用户操作（如Ctrl+C产生SIGINT信号）、硬件异常（如非法内存访问产生SIGSEGV信号）以及程序显式请求（如使用kill函数发送信号）。

特点：它是一种异步通信方式。

信号的发送：可以通过kill函数或raise函数发送信号。kill函数允许一个进程向另一个进程发送信号，而raise函数则允许进程向自己发送信号。

信号的接收：当信号被发送到进程时，操作系统会中断该进程的正常流程，并调用相应的信号处理函数（如果已设置）。如果没有设置信号处理函数，则进程会按照信号的默认行为执行（如终止进程、忽略信号或暂停进程等）。

信号处理函数：可以使用signal函数或更可靠的sigaction函数来设置信号处理函数。信号处理函数必须遵循特定的原型，并且当信号到达时会被调用。

#include <iostream>  
#include <csignal>  
#include <unistd.h> // 对于sleep()函数  // 信号处理函数  
void signalHandler(int signum) {  std::cout << "捕获到信号 " << signum << std::endl;  // 清理并关闭  // 注意：在实际的应用程序中，这里可能需要更复杂的清理代码  // 退出程序  exit(signum);  
}  int main () {  // 注册信号SIGINT和信号处理程序  signal(SIGINT, signalHandler);    while(1) {  std::cout << "等待信号..." << std::endl;  sleep(1); // 暂停一秒  }  return 0;  
}

在这个例子中，程序进入一个无限循环，每秒钟打印一条消息，并等待用户发送SIGINT信号（通常通过Ctrl+C）。当信号被捕获时，signalHandler函数被调用，程序随后退出。
请注意，使用signal函数有几个限制，包括它不能保证信号处理函数的原子性（即，在信号处理函数执行期间，其他信号可能会被阻塞或丢失）。因此，在需要可靠信号处理的场景下，通常推荐使用sigaction函数。不过，上述示例足以展示信号的基本用法。

9、剪贴板（Clipboard）

在Windows等图形界面中，剪贴板也可以作为一种进程间通信方式，它允许在不同进程之间复制和粘贴数据。

二、线程间通信

C++多线程通信是指在一个进程中运行的不同线程之间交换数据或控制信息，以协调它们的执行。在C++中，线程间通信可以通过多种方式实现，以下是几种主要的通信方式：

1、共享内存（Shared Memory）

共享内存是C++多线程通信中最直接的方式。多个线程可以访问同一块内存区域，从而实现数据的交换和共享。然而，由于多个线程可能同时访问同一块内存，因此需要使用同步机制（如互斥锁、读写锁等）来保证线程安全，避免竞态条件和数据不一致的问题。

实现方式：
使用全局变量、成员变量（对于多线程类）或通过指针/引用传递的数据结构，使得多个线程能够访问同一份数据。
使用C++标准库中的std::mutex、std::lock_guard、std::unique_lock等同步机制来保护对共享内存的访问。

以下是一个简单的C++示例代码，展示了如何使用std::thread、std::mutex和共享内存来实现线程间通信。在这个例子中，我们将创建两个线程：一个生产者线程和一个消费者线程。生产者线程将向共享内存区域写入数据，而消费者线程将从该区域读取数据。为了同步访问，我们将使用一个互斥锁来保护共享内存。

#include <iostream>  
#include <thread>  
#include <mutex>  
#include <condition_variable>  
#include <chrono>  // 共享数据结构和互斥锁  
std::mutex mtx;  
int shared_data = 0;  
bool data_ready = false;  // 条件变量，用于通知消费者数据已准备好  
std::condition_variable cv;  // 生产者线程函数  
void producer() {  for (int i = 0; i < 5; ++i) {  std::this_thread::sleep_for(std::chrono::seconds(1)); // 模拟耗时操作  std::lock_guard<std::mutex> lck(mtx);  shared_data = i * 10; // 生产数据  data_ready = true; // 标记数据已准备好  std::cout << "Produced: " << shared_data << std::endl;  cv.notify_one(); // 通知一个等待的线程  }  
}  // 消费者线程函数  
void consumer() {  while (true) {  std::unique_lock<std::mutex> lck(mtx);  // 等待数据准备好  cv.wait(lck, []{ return data_ready; });  // 读取数据  std::cout << "Consumed: " << shared_data << std::endl;  // 重置数据准备状态  data_ready = false;  // 释放锁，以便生产者可以继续生产  lck.unlock();  // 在这里可以添加更多的处理逻辑  // 注意：这个简单的例子没有优雅地退出消费者线程。在实际应用中，你可能需要添加一些逻辑来安全地退出循环。  }  
}  int main() {  std::thread producer_thread(producer);  std::thread consumer_thread(consumer);  producer_thread.join(); // 等待生产者线程完成  consumer_thread.join(); // 注意：在这个例子中，消费者线程永远不会自己退出，所以这里会导致死锁  // 在实际应用中，你可能需要一种方法来优雅地停止消费者线程，比如使用原子变量作为退出标志。  return 0;  
}  // 注意：上面的代码示例中，消费者线程使用了无限循环，并且没有优雅地退出循环的机制。  
// 在实际应用中，你可能需要添加一个原子变量作为退出标志，并在适当的时候设置它，以便消费者线程可以安全地退出循环。

2、消息队列（Message Queue）

消息队列是另一种常见的线程间通信方式。线程之间可以通过消息队列来传递数据，一个线程将数据放入队列中，另一个线程从队列中取出数据。这种方式可以实现线程间的解耦合，使得线程之间不需要直接访问对方的内存空间。

实现方式：
使用C++标准库中的std::queue或其他容器类来实现消息队列。配合互斥锁等同步机制来保护对队列的访问，确保线程安全。以下是一个使用std::queue、std::mutex和std::condition_variable实现的简单线程间通信示例，其中包含一个生产者线程和一个消费者线程：

#include <iostream>  
#include <queue>  
#include <thread>  
#include <mutex>  
#include <condition_variable>  
#include <stdexcept>  // 消息队列和同步机制  
std::queue<int> messages;  
std::mutex mtx;  
std::condition_variable cv;  
bool done = false; // 用于优雅地停止消费者线程  // 生产者线程函数  
void producer(int id) {  for (int i = 0; i < 5; ++i) {  std::this_thread::sleep_for(std::chrono::seconds(1)); // 模拟耗时操作  std::lock_guard<std::mutex> lck(mtx);  messages.push(id * 10 + i); // 生产消息  std::cout << "Producer " << id << " produced " << messages.back() << std::endl;  cv.notify_one(); // 通知消费者线程  }  // 所有消息生产完毕后，通知消费者线程可以退出了（可选）  // 注意：这通常不是停止消费者线程的唯一方式，因为消费者可能还在等待新消息  // 在这个例子中，我们简单地设置一个标志并再次通知消费者  {  std::lock_guard<std::mutex> lck(mtx);  done = true;  cv.notify_one();  }  
}  // 消费者线程函数  
void consumer() {  while (true) {  std::unique_lock<std::mutex> lck(mtx);  cv.wait(lck, []{ return !messages.empty() || done; }); // 等待消息或完成信号  if (done && messages.empty()) {  break; // 优雅地退出循环  }  int msg = messages.front();  messages.pop();  std::cout << "Consumer consumed " << msg << std::endl;  lck.unlock(); // 在处理消息之前释放锁（如果不需要在锁内处理）  // 处理消息（在这个例子中只是打印）  }  
}  int main() {  std::thread producer_thread1(producer, 1);  std::thread producer_thread2(producer, 2); // 可以添加多个生产者  std::thread consumer_thread(consumer);  producer_thread1.join();  producer_thread2.join(); // 等待所有生产者线程完成  // 通知消费者线程所有生产者都已完成（尽管在这个例子中，消费者线程可能会自己检测到这一点）  {  std::lock_guard<std::mutex> lck(mtx);  cv.notify_one(); // 可选，但在这个例子中，消费者线程可能已经在等待了  }  consumer_thread.join(); // 等待消费者线程完成  return 0;  
}

在这个示例中，我们创建了一个全局的std::queue<int>作为消息队列，以及一个互斥锁std::mutex和一个条件变量std::condition_variable来同步对消息队列的访问。生产者线程向队列中添加消息，并通过条件变量通知消费者线程。消费者线程则等待消息到来，处理消息，并在接收到所有生产者已完成的信号（或队列为空且done标志被设置）时退出循环。

3、同步对象

同步对象如信号量、条件变量等，可以用于协调多个线程的动作，实现线程间的同步与通信。

实现方式：
信号量：C++20引入了std::counting_semaphore，它是一种计数型的同步原语，可用于限制同时访问共享资源的线程数量，或作为事件计数器。
条件变量：C++标准库提供了std::condition_variable类，它允许一个线程等待特定条件满足时才继续执行，同时允许另一个线程改变该条件并通知等待线程。条件变量通常与互斥锁一起使用。

4、原子操作

原子操作是指不可分割的操作，即这些操作在执行过程中不能被其他线程中断。C++11引入了std::atomic模板类来支持原子操作，它提供了对变量的原子读写操作，避免了竞态条件问题。

5、Future和Promise

std::future和std::promise提供了一种机制，允许一个线程向另一个线程传递异步计算的结果。std::promise用于设置一个可由std::future检索的结果，而std::future则提供了一种阻塞或非阻塞的方式来获取这个结果。

#include <iostream>  
#include <future>  
#include <thread>  
#include <chrono>  // 生产者函数，计算并设置promise的值  
void producer(std::promise<int> promise) {  // 模拟耗时的计算  std::this_thread::sleep_for(std::chrono::seconds(1));  // 计算结果  int result = 42; // 假设这是某种复杂计算的结果  // 将结果设置到promise中  promise.set_value(result);  
}  // 消费者函数，从future中获取值  
void consumer(std::future<int> future) {  // 等待生产者设置值  int value = future.get(); // 这会阻塞，直到值被设置  // 使用值  std::cout << "The value is: " << value << std::endl;  
}  int main() {  // 创建一个promise<int>  std::promise<int> promise;  // 从promise获取future  std::future<int> future = promise.get_future();  // 启动生产者线程  std::thread producerThread(producer, std::move(promise));  // 在主线程中作为消费者  consumer(std::move(future));  // 等待生产者线程完成  producerThread.join();  return 0;  
}