C++多线程通信

news2024/11/20 12:21:11

多线程通信

引言
区别
- 活锁
- - 什么是活锁
  - 与死锁的区别
  - 如何避免活锁
多线程通信
- 示例
- 运行结果

引言

多线程同步与多线程通信实质上是两种相互关联但又不完全相同的东西。本文注重多线程同步与多线程通信的区别，同时重点讲述多线程通信中的消息队列。

区别

多线程通信通常指的是线程之间交换信息或数据的方式。线程之间可能需要共享数据、发送消息或信号，以便协调它们的行为或实现某种协作。

多线程同步则是关于控制或协调多个线程的执行顺序，以确保它们能够正确地共享资源或执行特定的任务序列。同步机制用于防止数据竞争（data race）和其他并发问题，例如死锁（deadlock）和活锁（livelock）。

活锁

关于活锁记录一下。

什么是活锁

活锁（Livelock）是一种特殊的并发现象，它发生在多个进程或线程在尝试访问共享资源时，由于某种条件没有满足，导致它们不断重复尝试访问，但始终无法成功。这些进程或线程在活锁状态下会不断地改变状态，但它们并没有真正地“前进”，因此得名“活锁”。

活锁可以认为是一种特殊的饥饿现象，因为进程或线程虽然没有被阻塞，但它们始终无法访问所需的资源，从而导致它们无法继续执行。

与死锁的区别

与死锁（Deadlock）不同的是，活锁中的实体（进程或线程）并没有停止运行，而是不断尝试访问资源，但由于某些条件没有满足（例如，资源被其他进程占用，或者进程间的通信出现问题），它们始终无法成功访问。因此，活锁有可能自行解开，只要满足了相关条件，进程或线程就可以成功访问资源。

如何避免活锁

一种简单方法是采用先来先服务的策略，当多个事务请求封锁同一数据对象时，封锁子系统按请求封锁的先后次序对事务排队，数据对象上的锁一旦释放就批准申请队列中第一个事务获得锁。这样可以确保每个进程或线程都有机会访问资源，从而避免活锁的发生。

多线程通信

多线程通信的方式有：
共享内存、消息队列、互斥锁、条件变量、信号量、future和promise。
互斥锁、条件变量、信号量、future和promise的使用可以查阅：
多线程同步（上）
多线程同步（下）
本文主要讲解消息队列。

示例

下面是一个简单的实现示例。实现的是两个线程同时对一个消息队列操作，消息队列最大容纳5个数据，一个生产者线程向里存数据，一个消费者线程向外取数据，总共循环十次。

#include <iostream>  
#include <thread>  
#include <queue>  
#include <mutex>  
#include <condition_variable>  

using namespace std;
constexpr int Max = 5;
constexpr int g_count = 10;

// 消息队列类  
class MessageQueue {
public:
    // 入队操作  
    void Enqueue(int message) {
        std::unique_lock<std::mutex> lock(mutex_);
        cond_.wait(lock, [&] {return queue_.size() < 5; });
        queue_.push(message);
        cond_.notify_one(); // 通知等待的线程  
    }

    // 出队操作  
    bool Dequeue(int& message) {
        std::unique_lock<std::mutex> lock(mutex_);
        cond_.wait(lock, [this] { return !queue_.empty(); }); // 等待直到队列不为空  
        message = queue_.front();
        queue_.pop();
        cond_.notify_all();
        return true;
    }

private:
    std::queue<int> queue_;
    std::mutex mutex_;
    std::condition_variable cond_;
    bool m_exit = false;
};

// 生产者线程函数  
void Producer(MessageQueue& queue) {
    for (int i = 0; i < g_count; ++i) {
        std::cout << "Producing message: " << i << std::endl;
        queue.Enqueue(i);
        std::this_thread::sleep_for(std::chrono::milliseconds(100)); // 模拟耗时操作  
    }
}

// 消费者线程函数  
void Consumer(MessageQueue& queue) {
    int message;
    for (size_t i = 0; i < g_count; i++)
    {
        if (queue.Dequeue(message)) {
            std::cout << "Consuming message: " << message << std::endl;
            std::this_thread::sleep_for(std::chrono::milliseconds(100)); // 模拟耗时操作  
        }
    } 
}

int main() {
    MessageQueue queue;

    // 创建生产者线程  
    std::thread producerThread(Producer, std::ref(queue));

    // 创建消费者线程  
    std::thread consumerThread(Consumer, std::ref(queue));

    // 等待生产者线程完成  
    producerThread.join();

    // 等待消费者线程完成  
    consumerThread.join();

    return 0;
}