Linux 线程：线程同步、生产者消费者模型

时间：2024-04-07 13:55:33

目录

一、死锁

二、条件变量实现线程同步

1、为什么需要线程同步

2、条件变量、同步、竞态条件

3、条件变量函数：初始化 销毁 等待 唤醒

4、实现简单的多线程程序

不唤醒则一直等待

实现线程同步

三、生产者消费者

1、借助超市模型理解

2、优点

四、基于阻塞队列的生产者消费者模型

lockGuard.hpp

BlockQueue.hpp 

Task.hpp

ConProd.cc

一、死锁

死锁是一个计算机科学中的概念，它描述了一种在多进程系统中可能出现的僵局，其中每个进程都持有至少一项资源，并正在等待其他进程中被占用的资源才能继续执行，然而，这些进程又都不肯释放自己已占有的资源，从而导致所有进程都无法向前推进，形成了一个持续的阻塞状态。

死锁发生的四个必备条件如下：

  1. 互斥条件（Mutual Exclusion）： 指系统中的至少某部分资源是非共享的，即在任意时刻，一个资源只能被一个进程占用。当一个进程占有某种资源时，其他进程若请求该资源，则必须等待至该资源被释放为止。

  2. 请求与保持条件（Hold and Wait）： 进程在请求新的资源的同时，继续保持对已分配资源的占有。这意味着某个进程可能已经拥有一些资源，但因缺乏其他资源而暂停执行，即使它并没有释放已获取的资源。

  3. 不可剥夺条件（No Preemption）： 已经分配给一个进程的资源，在该进程自愿释放之前，系统无法强制剥夺这部分资源。这就意味着进程在等待新资源的过程中，无法将其已占有的资源转交给其他等待的进程。

  4. 循环等待条件（Circular Wait）： 存在一个进程间的资源请求关系构成了一个环状结构，即每个进程都在等待下一个进程所占用的资源，形成闭环等待链。这样，任一进程都无法得到满足，因而也无法退出等待状态。

为了避免死锁的发生，可以从破坏上述四个条件中的一个或几个出发制定策略：

  • 一致的加锁顺序：确保所有进程按照预先设定的全局顺序请求资源，以此消除循环等待条件。
  • 避免持有资源不释放：鼓励进程在完成对资源的使用之后立即释放，尤其是当进程不再需要资源或者转换执行状态时。
  • 一次性分配策略：在进程开始执行前就一次性分配其所需的所有资源，如果系统无法一次性满足所有需求，则拒绝该进程启动，从而避免了请求与保持条件。
  • 资源预分配或超时回收：通过合理的资源预分配或设置资源请求超时机制，可以在一定程度上打破死锁僵局。

二、条件变量实现线程同步

1、为什么需要线程同步

        在多线程环境下，若某个线程频繁地获取并占用临界资源，不仅可能导致其他线程因无法获取资源而处于饥饿状态，同时也会造成系统资源的极大浪费。解决这一问题的关键在于实现线程间的同步，以确保合理、有序地访问临界资源。

  • 具体而言，在访问临界资源之前，线程必须首先检查资源是否可用，而这一步骤本质上也是一种对临界资源的访问，因此也需要在加锁和解锁操作之间进行。但是，传统的通过循环检测资源是否就绪的方式会导致线程反复尝试获取资源，从而加剧系统负担。

        为了解决这个问题，我们需要一种机制能够让线程在资源未就绪时不再自我频繁检测，而在资源就绪时能够及时得到通知，并立即进行资源请求和访问。这就是条件变量的作用所在。条件变量允许线程在特定条件不满足时挂起等待，一旦条件满足，由其他线程负责唤醒相应的等待线程，进而有效避免了无谓的资源竞争与浪费，实现了线程间更为高效和协调的同步。

2、条件变量、同步、竞态条件

条件变量：
  • 当一个线程试图访问或操作一个依赖于某种条件的临界资源时，比如一个空的队列，若此时没有元素可供消费，线程就会陷入无法继续执行的状态。
  • 这时，条件变量允许线程在该条件不满足时进入等待状态，而不是持续消耗CPU资源进行无效循环检查（自旋等待）。
  • 当另一个线程修改了状态，例如向队列中添加了一个新的元素，满足了原先等待线程的需求条件，这时可以通过发送一个信号告知等待的线程，使其从等待状态恢复执行。
同步的概念： 同步是指通过特定机制确保不同线程按照预定的顺序或逻辑关系访问共享资源的过程。它的目标是在并发环境下保证数据的一致性和完整性，防止因多个线程同时访问导致的数据混乱，以及由于资源竞争带来的不公平调度和饥饿问题。
竞态条件：
  • 竞态条件（Race Condition）是指程序的结果依赖于多个线程执行的相对时机，如果这些线程都在访问和修改同一个共享数据，且没有采取适当的同步措施，那么可能会出现不可预期的行为。
  • 例如，两个线程分别从队列中取出和添加元素，如果不加锁，可能出现一个线程刚判断队列非空然后开始移除元素，但还未完成操作时，另一个线程已经把最后一个元素添加进去了，最终可能导致队列在不应该为空的情况下变为空。

因此，使用互斥锁和条件变量可以有效避免竞态条件的发生：

  • 互斥锁用于确保每次只有一个线程可以访问临界区（如队列结构），从而消除数据竞争。
  • 条件变量则在此基础上提供了额外的逻辑层，使得线程可以在特定条件成立时才执行后续操作，而不是盲目地尝试访问资源。

3、条件变量函数：初始化 销毁 等待 唤醒

在Linux多线程编程中，条件变量相关的函数主要用于线程间的同步与通信。以下是对这些函数的详细解释：

pthread_cond_init: 这个函数用于初始化一个条件变量。

int pthread_cond_init(pthread_cond_t *restrict cond, const pthread_condattr_t *restrict attr);
  • 参数cond是指向条件变量结构体的指针，这个函数会将其初始化为可用状态。
  • 通常情况下，第二个参数attr设置为NULL，表示使用默认属性初始化条件变量。不过，也可以通过创建和设置pthread_condattr_t类型的属性对象来自定义条件变量的属性，例如指定条件变量的类型（是否支持广播等）。

pthread_cond_destroy: 此函数用于销毁一个已初始化的条件变量。

int pthread_cond_destroy(pthread_cond_t *cond);
  • 在不再需要条件变量或者所有使用该条件变量的线程都已完成之前，应调用此函数。只有当没有线程在条件变量上等待时，才能成功销毁。

pthread_cond_wait: 这个函数会让当前线程阻塞，直到指定的条件满足。

int pthread_cond_wait(pthread_cond_t *restrict cond, pthread_mutex_t *restrict mutex);
  • 线程首先必须持有与条件变量关联的互斥锁（通过参数mutex指定），然后调用此函数时会释放互斥锁，进入等待状态。
  • 当其他线程调用pthread_cond_signal或pthread_cond_broadcast唤醒等待在此条件变量上的线程时，等待的线程会重新获取互斥锁并返回。

pthread_cond_signal 和 pthread_cond_broadcast: 这两个函数用于唤醒正在条件变量上等待的线程。

int pthread_cond_signal(pthread_cond_t *cond);
int pthread_cond_broadcast(pthread_cond_t *cond);
  • pthread_cond_signal: 唤醒一个（任意一个）正在等待条件变量cond的线程。如果有多个线程在等待，仅选择一个线程解除阻塞。
  • pthread_cond_broadcast: 唤醒所有正在等待条件变量cond的线程。所有等待的线程都会收到信号，但具体哪个线程能立即恢复执行还取决于互斥锁的获取顺序。

4、实现简单的多线程程序

这段代码是一个简单的多线程程序，在POSIX环境下使用C++编写，利用pthread库实现线程间的同步。程序创建了一个互斥锁（mutex）和一个条件变量（condition variable），以及四个线程，每个线程分别执行func1、func2、func3和func4中的任务。

#include <iostream>
#include <pthread.h>
#include <unistd.h>

// 定义线程数量常量
#define TNUM 4

// 定义回调函数类型，该类型函数接收字符串引用、互斥锁指针和条件变量指针作为参数
typedef void (*func_t)(const std::string &name, pthread_mutex_t *pmtx, pthread_cond_t *pcond);

// 定义一个结构体，用于存储传递给线程的数据
class ThreadData
{
public:
    // 构造函数，初始化线程数据
    ThreadData(const std::string &name, func_t func, pthread_mutex_t *pmtx, pthread_cond_t *pcond)
        : name_(name), func_(func), pmtx_(pmtx), pcond_(pcond) {}

    // 线程名
    std::string name_;
    // 需要执行的函数指针
    func_t func_;
    // 互斥锁指针
    pthread_mutex_t *pmtx_;
    // 条件变量指针
    pthread_cond_t *pcond_;
};

// 函数1，线程执行体之一
void func1(const std::string &name, pthread_mutex_t *pmtx, pthread_cond_t *pcond)
{
    while (true)
    {
        // 线程在此等待条件变量的信号
        pthread_cond_wait(pcond, pmtx);
        std::cout << "thread " << name << " is running -- a" << std::endl;
    }
}

// 类似的函数定义，对应其他线程执行体
void func2(const std::string &name, pthread_mutex_t *pmtx, pthread_cond_t *pcond)
{

    while (true)
    {
        pthread_cond_wait(pcond, pmtx);
        std::cout << "thread " << name << " is running -- b" << std::endl;
    }
}

void func3(const std::string &name, pthread_mutex_t *pmtx, pthread_cond_t *pcond)
{
    while (true)
    {
        pthread_cond_wait(pcond, pmtx);
        std::cout << "thread " << name << " is running -- b" << std::endl;
    }
}

void func4(const std::string &name, pthread_mutex_t *pmtx, pthread_cond_t *pcond)
{
    while (true)
    {
        pthread_cond_wait(pcond, pmtx);
        std::cout << "thread " << name << " is running -- d" << std::endl;
    }
}

// 线程入口函数，负责调用实际的线程执行函数
void *Entry(void *args)
{
    // 将参数转换为ThreadData指针
    ThreadData *td = (ThreadData *)args;
    // 调用传递进来的函数
    td->func_(td->name_, td->pmtx_, td->pcond_);
    // 删除ThreadData对象，释放内存
    delete td;
    return nullptr;
}

int main()
{
    // 初始化互斥锁和条件变量
    pthread_mutex_t mtx;
    pthread_mutex_init(&mtx, nullptr);
    pthread_cond_t cond;
    pthread_cond_init(&cond, nullptr);

    // 创建线程数组
    pthread_t tids[TNUM];

    // 函数指针数组，存放四个线程执行函数
    func_t funcs[TNUM] = {func1, func2, func3, func4};

    // 创建并启动TNUM个线程
    for (int i = 0; i < TNUM; i++)
    {
        // 创建线程名
        std::string name = "Thread";
        name += std::to_string(i + 1);

        // 创建ThreadData对象，封装线程数据
        ThreadData *td = new ThreadData(name, funcs[i], &mtx, &cond);

        // 创建并启动线程
        pthread_create(tids + i, nullptr, Entry, (void *)td);
    }

    // 主线程休眠2秒
    sleep(2);

    // 主线程进入无限循环，定期发送信号唤醒等待条件变量的线程
    while (true)
    {
        std::cout << "resume thread run code ..." << std::endl;
        pthread_cond_signal(&cond); // 发送信号唤醒一个等待条件变量的线程
        sleep(1); // 主线程休眠1秒
    }

    // （理论上，上面的循环应该有一个退出条件，这里未给出）

    // 等待所有线程完成，并输出相关信息
    for (int i = 0; i < TNUM; i++)
    {
        pthread_join(tids[i], nullptr);
        std::cout << "thread: " << tids[i] << " quit" << std::endl;
    }

    // 销毁互斥锁和条件变量
    pthread_mutex_destroy(&mtx);
    pthread_cond_destroy(&cond);

    return 0;
}

  1. 定义了一个ThreadData结构体类，用于封装传递给各个线程的数据，包括线程名称、要执行的函数指针和需要使用的互斥锁及条件变量指针。

  2. 定义了四个函数func1、func2、func3、func4，它们都具有相似的结构：在一个无限循环内调用pthread_cond_wait来阻塞线程，等待条件变量接收到信号。一旦接收到信号，线程会打印一条表示自己正在运行的消息，并再次进入等待状态。

  3. Entry函数作为线程入口点，接收ThreadData实例参数，调用传入的成员函数执行实际的任务。

  4. 在main函数中：

    • 初始化互斥锁和条件变量。
    • 创建四个线程，并将ThreadData实例传递给每个线程，对应的函数指针指向func1至func4。
    • 主线程休眠2秒后，开始进入无限循环，每次循环都会打印一条消息并调用pthread_cond_signal唤醒一个等待条件变量的线程。
    • 由于没有特殊逻辑控制哪个线程被唤醒，实际上每次pthread_cond_signal都会随机地唤醒其中一个等待的线程（尽管实际行为取决于具体实现，但通常不会保证特定的唤醒顺序）。
    • 然后主线程休眠1秒，再次循环。
    • 最终，虽然代码中没有显示退出循环的条件，但在真实场景中用户可能会手动中断程序（如示例中的^C所示，表示按Ctrl+C中断了程序）。

  5. 当所有线程都结束时，主线程会通过pthread_join等待每个子线程完成，并清理相关的互斥锁和条件变量资源。

这个多线程程序有明显的缺点：没有明确的退出机制，且在 pthread_cond_wait 调用前后没有包裹互斥锁操作，可能导致竞态条件。

[hbr@VM-16-9-centos synchronization]$ make
g++ -o mycond mycond.cc -lpthread
[hbr@VM-16-9-centos synchronization]$ ./mycond 
resume thread run code ...
thread Thread1 is running -- a
resume thread run code ...
thread Thread2 is running -- b
resume thread run code ...
thread Thread3 is running -- b
resume thread run code ...
thread Thread4 is running -- d
resume thread run code ...
thread Thread1 is running -- a
resume thread run code ...
thread Thread2 is running -- b
resume thread run code ...
thread Thread3 is running -- b
resume thread run code ...
thread Thread4 is running -- d
resume thread run code ...
thread Thread1 is running -- a
resume thread run code ...
thread Thread2 is running -- b
^C

pthread_cond_broadcast: 唤醒所有正在等待条件变量cond的线程。

    sleep(2);
    while (true)
    {
        std::cout << "resume thread run code ..." << std::endl;
        // pthread_cond_signal(&cond);
        pthread_cond_broadcast(&cond);
        sleep(1);
    }

[hbr@VM-16-9-centos synchronization]$ ./mycond 
resume thread run code ...
thread Thread3 is running -- b
thread Thread1 is running -- a
thread Thread4 is running -- d
thread Thread2 is running -- b
resume thread run code ...
thread Thread3 is running -- b
thread Thread1 is running -- a
thread Thread4 is running -- d
thread Thread2 is running -- b
resume thread run code ...
thread Thread3 is running -- b
thread Thread1 is running -- a
thread Thread4 is running -- d
thread Thread2 is running -- b
^C

不唤醒则一直等待

在 main 函数中，主线程并没有调用 pthread_cond_signal 或 pthread_cond_broadcast 来唤醒任何一个等待条件变量的线程。这意味着工作线程中的 pthread_cond_wait 函数会一直阻塞在那里，直到接收到一个条件信号。

#include <iostream>
#include <pthread.h>
#include <unistd.h>

#define TNUM 4
typedef void (*func_t)(const std::string &name, pthread_mutex_t *pmtx, pthread_cond_t *pcond);
volatile bool quit = false;

class ThreadData
{
public:
    ThreadData(const std::string &name, func_t func, pthread_mutex_t *pmtx, pthread_cond_t *pcond)
        : name_(name), func_(func), pmtx_(pmtx), pcond_(pcond) {}
    std::string name_;
    func_t func_;
    pthread_mutex_t *pmtx_;
    pthread_cond_t *pcond_;
};

void func1(const std::string &name, pthread_mutex_t *pmtx, pthread_cond_t *pcond)
{

    while (!quit)
    {
        pthread_cond_wait(pcond, pmtx);
        std::cout << "thread " << name << " is running -- a" << std::endl;
    }
}
void func2(const std::string &name, pthread_mutex_t *pmtx, pthread_cond_t *pcond)
{

    while (!quit)
    {
        pthread_cond_wait(pcond, pmtx);
        std::cout << "thread " << name << " is running -- b" << std::endl;
    }
}
void func3(const std::string &name, pthread_mutex_t *pmtx, pthread_cond_t *pcond)
{
    while (!quit)
    {
        pthread_cond_wait(pcond, pmtx);
        std::cout << "thread " << name << " is running -- b" << std::endl;
    }
}
void func4(const std::string &name, pthread_mutex_t *pmtx, pthread_cond_t *pcond)
{
    while (!quit)
    {
        pthread_cond_wait(pcond, pmtx);
        std::cout << "thread " << name << " is running -- d" << std::endl;
    }
}
void *Entry(void *args)
{
    ThreadData *td = (ThreadData *)args;
    td->func_(td->name_, td->pmtx_, td->pcond_);
    delete td;
    return nullptr;
}
int main()
{
    pthread_mutex_t mtx;
    pthread_cond_t cond;
    pthread_mutex_init(&mtx, nullptr);
    pthread_cond_init(&cond, nullptr);

    pthread_t tids[TNUM];
    func_t funcs[TNUM] = {func1, func2, func3, func4};
    for (int i = 0; i < TNUM; i++)
    {
        std::string name = "Thread";
        name += std::to_string(i + 1);
        ThreadData *td = new ThreadData(name, funcs[i], &mtx, &cond);
        pthread_create(tids + i, nullptr, Entry, (void *)td);
    }

    sleep(2);
    int n = 10;
    while (n)
    {
        std::cout << "resume thread run code ..." << n-- << std::endl;
        // pthread_cond_signal(&cond);
        //pthread_cond_broadcast(&cond);
        sleep(1);
    }

    quit = true;

    for (int i = 0; i < TNUM; i++)
    {
        pthread_join(tids[i], nullptr);
        std::cout << "thread: " << tids[i] << "quit" << std::endl;
    }

    pthread_mutex_destroy(&mtx);
    pthread_cond_destroy(&cond);
    return 0;
}
[hbr@VM-16-9-centos synchronization]$ ./mycond 
resume thread run code ...10
resume thread run code ...9
resume thread run code ...8
resume thread run code ...7
resume thread run code ...6
resume thread run code ...5
resume thread run code ...4
resume thread run code ...3
resume thread run code ...2
resume thread run code ...1

这是因为 pthread_cond_wait 函数的作用是在满足特定条件时让线程等待，如果不主动唤醒，它会一直阻塞在那里。当一个线程调用 pthread_cond_wait 函数时，会发生以下两个关键操作：

  1. 线程会自动释放它已经持有的互斥锁（在这里是 pmtx）。
  2. 线程进入休眠状态，直到接收到 pthread_cond_signal 或 pthread_cond_broadcast 发送的信号，或者其他线程对条件变量进行广播（wake up all waiting threads）。

如果没有其他线程调用 pthread_cond_signal 或 pthread_cond_broadcast，那么调用了 pthread_cond_wait 的线程就不会被唤醒，也就无法重新获取互斥锁并继续执行其后续的代码。因此，在没有唤醒信号的情况下，线程会一直停留在等待状态，无法执行后面的任务。

实现线程同步

这次我们改进了第二段代码的不足之处，不仅设置了退出条件（quit），而且在工作线程调用 pthread_cond_wait 之前和之后增加了对互斥锁的操作，确保了临界区的正确同步。

#include <iostream>
#include <pthread.h>
#include <unistd.h>

#define TNUM 4
typedef void (*func_t)(const std::string &name, pthread_mutex_t *pmtx, pthread_cond_t *pcond);
volatile bool quit = false;

class ThreadData
{
public:
    ThreadData(const std::string &name, func_t func, pthread_mutex_t *pmtx, pthread_cond_t *pcond)
        : name_(name), func_(func), pmtx_(pmtx), pcond_(pcond) {}
    std::string name_;
    func_t func_;
    pthread_mutex_t *pmtx_;
    pthread_cond_t *pcond_;
};

void func1(const std::string &name, pthread_mutex_t *pmtx, pthread_cond_t *pcond)
{
    while (!quit)
    {
        pthread_mutex_lock(pmtx);
        pthread_cond_wait(pcond, pmtx);
        std::cout << "thread " << name << " is running -- a" << std::endl;
        pthread_mutex_unlock(pmtx);
    }
}
void func2(const std::string &name, pthread_mutex_t *pmtx, pthread_cond_t *pcond)
{
    while (!quit)
    {
        pthread_mutex_lock(pmtx);
        pthread_cond_wait(pcond, pmtx);
        std::cout << "thread " << name << " is running -- b" << std::endl;
        pthread_mutex_unlock(pmtx);
    }
}
void func3(const std::string &name, pthread_mutex_t *pmtx, pthread_cond_t *pcond)
{
    while (!quit)
    {
        pthread_mutex_lock(pmtx);
        pthread_cond_wait(pcond, pmtx);
        std::cout << "thread " << name << " is running -- b" << std::endl;
        pthread_mutex_unlock(pmtx);
    }
}
void func4(const std::string &name, pthread_mutex_t *pmtx, pthread_cond_t *pcond)
{
    while (!quit)
    {
        pthread_mutex_lock(pmtx);
        pthread_cond_wait(pcond, pmtx);
        std::cout << "thread " << name << " is running -- d" << std::endl;
        pthread_mutex_unlock(pmtx);
    }
}
void *Entry(void *args)
{
    ThreadData *td = (ThreadData *)args;
    td->func_(td->name_, td->pmtx_, td->pcond_);
    delete td;
    return nullptr;
}
int main()
{
    pthread_mutex_t mtx;
    pthread_cond_t cond;
    pthread_mutex_init(&mtx, nullptr);
    pthread_cond_init(&cond, nullptr);

    pthread_t tids[TNUM];
    func_t funcs[TNUM] = {func1, func2, func3, func4};
    for (int i = 0; i < TNUM; i++)
    {
        std::string name = "Thread";
        name += std::to_string(i + 1);
        ThreadData *td = new ThreadData(name, funcs[i], &mtx, &cond);
        pthread_create(tids + i, nullptr, Entry, (void *)td);
    }

    sleep(2);
    int n = 5;
    while (n)
    {
        std::cout << "resume thread run code ..." << n-- << std::endl;
        // pthread_cond_signal(&cond);
        pthread_cond_broadcast(&cond);
        sleep(1);
    }
    std::cout << "ctrl done" << std::endl;
    quit = true;
    pthread_cond_broadcast(&cond);
    for (int i = 0; i < TNUM; i++)
    {
        pthread_join(tids[i], nullptr);
        std::cout << "thread: " << tids[i] << "quit" << std::endl;
    }

    pthread_mutex_destroy(&mtx);
    pthread_cond_destroy(&cond);
    return 0;
}

在每个工作线程调用 pthread_cond_wait 函数之前和之后添加了对互斥锁 pmtx 的锁定和解锁操作。这种改动增强了代码的同步机制，具有以下几个重要作用：

  1. 互斥访问共享资源： 通过在调用 pthread_cond_wait 前调用 pthread_mutex_lock，确保了在等待条件变量之前，线程已获得对共享资源的独占访问权。这样可以防止其他线程在等待条件变量的线程被唤醒之前修改共享资源的状态，保证了数据一致性。

  2. 正确同步线程： 使用互斥锁配合条件变量，可以精确地控制线程何时等待（pthread_cond_wait）和何时被唤醒继续执行。当线程被唤醒时，它必须重新获取互斥锁才能继续执行临界区代码，这样可以避免多个线程同时唤醒后竞态执行。

  3. 优雅地终止线程： 在 main 函数中，当 quit 变量被设置为 true 时，线程会检查该变量并退出循环。由于每个线程在等待条件变量时都持有着互斥锁，所以在退出循环前解锁互斥锁是很重要的，这样其他线程才有机会更新共享状态并决定是否继续等待。

[hbr@VM-16-9-centos synchronization]$ ./mycond 
resume thread run code ...5
thread Thread3 is running -- b
thread Thread2 is running -- b
thread Thread4 is running -- d
thread Thread1 is running -- a
resume thread run code ...4
thread Thread3 is running -- b
thread Thread2 is running -- b
thread Thread4 is running -- d
thread Thread1 is running -- a
resume thread run code ...3
thread Thread3 is running -- b
thread Thread2 is running -- b
thread Thread4 is running -- d
thread Thread1 is running -- a
resume thread run code ...2
thread Thread3 is running -- b
thread Thread2 is running -- b
thread Thread4 is running -- d
thread Thread1 is running -- a
resume thread run code ...1
thread Thread3 is running -- b
thread Thread2 is running -- b
thread Thread4 is running -- d
thread Thread1 is running -- a
ctrl done
thread Thread3 is running -- b
thread Thread2 is running -- b
thread Thread4 is running -- d
thread Thread1 is running -- a
thread: 139858591500032quit
thread: 139858583107328quit
thread: 139858574714624quit
thread: 139858566321920quit

三、生产者消费者

1、借助超市模型理解

利用“超市”这一生动的类比，我们可以深入理解生产者-消费者问题这一经典的并发编程模型。在此模型框架内，“超市”象征着一种关键的共享资源——一个具有固定大小的缓冲区或队列，它是连接生产者与消费者的核心介质。

生产者：

  • 在这个经济系统隐喻中，生产者代表着那些负责创建和产出数据或实体的线程或进程。举例来说，生产者可能是一个不断生成日志记录的后台线程，或者是持续接收硬件传感器数据流的设备驱动程序。
  • 生产者的职责在于源源不断地制造“商品”（即数据元素），并将它们妥善地存入“超市货架”（即缓冲区）。当超市库存饱和，也就是缓冲区满载时，生产者无法继续存放商品，这就引申出了生产者间对于缓冲区存储空间的竞争性“锁定”，形成了一种“竞争关系”。

消费者：

  • 相反，消费者则是在此模型中消化这些数据或实体的角色，可以设想成是一个分析日志记录的后台处理作业，或是实时响应并处理传感器数据的主应用程序。
  • 消费者从“超市”中提取商品进行后续操作。若缓冲区耗尽无货，消费者们不得不暂时处于等待状态直至有新数据补充进来，这也意味着在缓冲区空置时，消费者群体内部为了抢占首个可用的空余位置也可能发生“竞争关系”。

超市（缓冲区）：

  • “超市”作为缓冲区的化身，扮演了至关重要的中介桥梁，它提供了一个安全可靠的临时存储和传输数据的机制。这种设计让生产者和消费者无需同步运行，极大地降低了它们之间的耦合度，增强了系统整体的灵活性和可扩展性。
  • 缓冲区的有效管理和调度策略极其重要，设计者必须确保机制能够有效规避潜在的死锁和饥饿问题，确保所有参与者都能公平且高效地利用有限的缓冲区空间，从而最大限度地提高整个系统的吞吐量和响应速度。

互斥/同步关系：

  • 生产者和消费者之间的关系：

    • 对应关系：竞争、互斥关系及互斥/同步
    • 生产者和消费者之间存在着竞争关系，这是因为缓冲区（超市）的容量有限，当缓冲区满时，生产者不能再添加商品（数据），此时生产者会与其他想要放入商品的生产者竞争缓冲区的空间。同样，当缓冲区空时，多个消费者会竞争缓冲区中的商品。
    • 生产者和消费者间也存在互斥关系，意味着在同一时刻只有一个生产者可以向缓冲区添加商品，同时只有一个消费者可以从缓冲区取出商品，这是为了避免数据的不一致性，即通过互斥锁等机制确保对缓冲区的操作是原子的。
    • 互斥/同步是指生产者和消费者之间的协作模式，生产者在添加商品后可能需要发送一个信号告诉消费者商品已经准备好，消费者在消费商品后也可能要通知生产者可以继续生产。这涉及到同步机制，比如条件变量或信号量，来保证两者间的协调运作。

  • 消费者和消费者之间的关系：

    • 对应关系：竞争关系
    • 当缓冲区中的商品数量不足以满足所有消费者的需求时，消费者之间会发生竞争关系，因为它们都想尽快从缓冲区获取商品进行处理。

  • 生产者和生产者之间的关系：

    • 对应关系：竞争关系
    • 在缓冲区容量有限的情况下，不同的生产者线程可能会彼此竞争，争相将商品放入缓冲区。当缓冲区满时，新的生产者必须等待其他生产者移除商品，释放出空间后才能继续添加。

  • 在生产者-消费者问题中，“超市”作为一个共享的公共资源，它的管理决定了生产者和消费者如何正确、安全且高效地进行数据交换，避免冲突和死锁现象的发生。通过合适的同步和互斥机制，生态系统中的每个角色都能在不影响其他角色的前提下完成自己的任务。

综上所述，生产者-消费者问题的核心是通过共享资源（缓冲区）协调两个或多个并发活动，确保它们既能独立工

2、优点

生产者消费者模型的优势在于：

  1. 深度解耦：生产者消费者模型实现了生产者和消费者之间的逻辑分离，使得生产和消费两个过程变得相对独立。生产者专注于生成数据，无需关心数据的具体处理细节，而消费者专心于处理数据，无需了解数据的来源和生成过程。这样显著减少了模块之间的耦合度，提高了代码的可维护性和可重用性。

  2. 并发支持：模型天然支持多线程或多进程环境下的并发执行。生产者可以在独立的线程或进程中持续生成数据，同时，消费者也可以在不同的线程或进程中并行地处理这些数据。这种并发执行提升了系统整体的效率和吞吐量。

  3. 负载均衡与动态调整 ：当生产者和消费者的工作负载不均匀时，该模型能够适应并优化这种差异。如果消费者处理速度快于生产者生成速度，那么消费者可以在没有数据可处理时等待；相反，如果生产者速度较快，缓冲区可以起到临时存储的作用，使消费者有充足的时间来逐步处理积累的数据，从而实现了对系统内部忙闲不均情况的良好应对。这种特性有助于避免资源浪费，确保系统稳定运行并保持高效率。

四、基于阻塞队列的生产者消费者模型

1、思路：

在多线程编程中，阻塞队列（Blocking Queue）是一种特殊的线程安全队列设计，它支持线程间的高效同步和数据交换。与传统的非阻塞队列相比，阻塞队列增加了额外的线程调度逻辑，使得队列在特定条件下能够暂停线程执行，直到满足特定条件为止。

特点：

  1. 队列为空时的消费行为： 当一个或多个消费者线程尝试从空的阻塞队列中获取元素时，这些线程将不会立即返回一个默认值或者抛出异常，而是进入阻塞状态，这意味着这些线程会暂时停止执行，等待其他线程将元素放入队列。一旦有生产者线程将元素成功放入队列，阻塞的消费者线程将被唤醒并继续执行，从队列中取出新放入的元素。

  2. 队列满时的生产行为： 反之，当生产者线程尝试向已满的阻塞队列中添加元素时，如果队列容量有限且已达到上限，该线程也不会立刻失败或抛出异常，而是同样进入阻塞状态，等待有消费者线程从队列中取出元素，腾出空间。一旦有元素被消费，阻塞的生产者线程得以解除阻塞，继续完成元素的入队操作。

通过这样的机制，阻塞队列自然而然地支持了“生产者-消费者”模型的实现，其中生产者线程负责生成数据并将其放入队列，而消费者线程则负责从队列中取出数据并进行处理。这种设计极大地简化了多线程间复杂同步问题的解决，实现了线程之间的无锁化协同工作，提高了系统的稳定性和效率。

2、lockGuard.hpp管理互斥

lockGuard.hpp 是实现“资源获取即初始化”（RAII, Resource Acquisition Is Initialization）设计模式的一个实用工具类，尤其适用于管理互斥锁（mutex）。在多线程编程中，互斥锁是一种常用的同步机制，用于保护共享资源不被多个线程同时访问。

#pragma once

#include <iostream>
#include <pthread.h>

class Mutex
{
public:
    Mutex(pthread_mutex_t *mtx):pmtx_(mtx)
    {}
    void lock() 
    {
        std::cout << "要进行加锁" << std::endl;
        pthread_mutex_lock(pmtx_);
    }
    void unlock()
    {
        std::cout << "要进行解锁" << std::endl;
        pthread_mutex_unlock(pmtx_);
    }
    ~Mutex()
    {}
private:
    pthread_mutex_t *pmtx_;
};

// RAII风格的加锁方式
class lockGuard
{
public:
    lockGuard(pthread_mutex_t *mtx):mtx_(mtx)
    {
        mtx_.lock();
    }
    ~lockGuard()
    {
        mtx_.unlock();
    }
private:
    Mutex mtx_;
};

lockGuard 类的设计思路如下：

  1. 类声明：

    • lockGuard 类接收一个指向 pthread_mutex_t 类型的指针作为构造函数参数，这个指针通常指向需要锁定的互斥锁。

  2. 构造函数：

    • 在 lockGuard 类实例化时（即创建对象时），会立即调用 lock() 成员函数，这会导致对应的互斥锁被锁定，从而确保在构造期间获得互斥锁的所有权。

  3. 析构函数：

    • 当 lockGuard 类的对象生命周期结束（例如，离开作用域或显式删除对象时），析构函数会被自动调用。在析构函数中，unlock() 成员函数被调用，这会释放之前在构造时获取的互斥锁。

  4. 效果：

    • 通过这种方式，lockGuard 类保证了互斥锁始终会在适当的时间被正确地锁定和解锁。这种“自动”管理锁的行为消除了手动管理锁时可能出现的忘记解锁或者提前解锁的风险，增强了代码的健壮性和安全性。

  5. 使用示例：

    Mutex mutex; // 假设有一个全局或局部的互斥锁实例
{
    lockGuard lock(&mutex); // 创建一个 lockGuard 对象，此时互斥锁被锁定
    // ... 这里是受保护的代码区域 ...
} // 当离开此作用域时，lockGuard 对象被销毁，互斥锁自动解锁

总结起来，lockGuard 类是一个轻量级的封装工具，通过 RAII 特性简化了对互斥锁的管理和使用，有助于编写更加简洁且不易出错的多线程代码。在 BlockQueue 类中，正是使用了 lockGuard 类来确保对队列操作的安全性，避免不同线程在访问队列时产生竞态条件。

3、BlockQueue.hpp 阻塞队列

BlockQueue.hpp 文件定义了一个名为 BlockQueue 的 C++ 模板类，它是一个固定容量的阻塞队列，主要用于多线程编程环境中的生产者-消费者场景。这个队列利用了 POSIX 线程同步原语 pthread_mutex_t（互斥锁）和 pthread_cond_t（条件变量）来实现在队列满或空时对线程的阻塞与唤醒。

#pragma once

#include <iostream>
#include <queue>
#include <mutex>
#include <pthread.h>
#include "lockGuard.hpp"

const int gDefaultCap = 5;

template <class T>
class BlockQueue
{
private:
    bool isQueueEmpty()
    {
        return bq_.size() == 0;
    }
    bool isQueueFull()
    {
        return bq_.size() == capacity_;
    }

public:
    BlockQueue(int capacity = gDefaultCap) : capacity_(capacity)
    {
        pthread_mutex_init(&mtx_, nullptr);
        pthread_cond_init(&Empty_, nullptr);
        pthread_cond_init(&Full_, nullptr);
    }
    void push(const T &in) // 生产者
    {
        lockGuard lockgrard(&mtx_); // 自动调用构造函数
        while (isQueueFull())
            pthread_cond_wait(&Full_, &mtx_);
        bq_.push(in);
        pthread_cond_signal(&Empty_);
    } // 自动调用lockgrard 析构函数
    void pop(T *out)
    {
        lockGuard lockguard(&mtx_);
        // pthread_mutex_lock(&mtx_);
        while (isQueueEmpty())
            pthread_cond_wait(&Empty_, &mtx_);
        *out = bq_.front();
        bq_.pop();

        pthread_cond_signal(&Full_);

        // pthread_mutex_unlock(&mtx_);
    }
    ~BlockQueue()
    {
        pthread_mutex_destroy(&mtx_);
        pthread_cond_destroy(&Empty_);
        pthread_cond_destroy(&Full_);
    }

private:
    std::queue<T> bq_;     // 阻塞队列
    int capacity_;         // 容量上限
    pthread_mutex_t mtx_;  // 通过互斥锁保证队列安全
    pthread_cond_t Empty_; // 用它来表示bq 是否空的条件
    pthread_cond_t Full_;  //  用它来表示bq 是否满的条件
};

以下是 BlockQueue 类的主要特点和功能：

  1. 模板类：BlockQueue 是一个模板类，允许用户指定队列中元素的数据类型 T，这意味着它可以用来存储任何类型的对象。

  2. 私有成员：

    • std::queue<T> bq_：用于存储实际的队列元素，是一个标准库中的队列容器。
    • int capacity_：队列的最大容量，初始化时可由用户指定，默认值为 gDefaultCap（5）。
    • pthread_mutex_t mtx_：互斥锁，用于控制对队列的并发访问，确保线程安全。
    • pthread_cond_t Empty_ 和 pthread_cond_t Full_：两个条件变量，分别表示队列为空和队列已满的信号。

  3. 公共成员函数：

    • 构造函数 BlockQueue(int capacity = gDefaultCap)：初始化队列及其相关的同步原语。
    • push(const T &in)：向队列中添加元素的方法，当队列满时，调用 pthread_cond_wait 让生产者线程进入等待状态，直到队列有空位时再唤醒。
    • pop(T *out)：从队列中移除并返回一个元素的方法，当队列空时，消费者线程也会通过 pthread_cond_wait 进入等待状态，直到队列中有元素可用时再唤醒。
    • 析构函数 ~BlockQueue()：销毁队列时，同时清理关联的互斥锁和条件变量。

  4. 辅助工具：

    • 在 push 和 pop 方法中都使用了 lockGuard 类，这是一种遵循 RAII（Resource Acquisition Is Initialization）原则的智能指针形式的类，负责在构造时自动锁定互斥锁并在析构时自动解锁，确保了即使出现异常也能正确释放资源。

通过以上设计，BlockQueue 类提供了一种机制，使得生产者线程在无法立即添加元素到队列时会暂停运行，而消费者线程在队列为空时也会停止工作。这样既解决了线程间同步的问题，也避免了不必要的 CPU 资源浪费。

4、Task.hpp派发任务

Task.hpp 文件定义了一个名为 Task 的 C++ 类，该类代表了一个简单的任务实体，其中包含两个整数成员变量 x_ 和 y_，以及一个 std::function<int(int, int)> 类型的成员变量 func_。

#pragma once

#include <iostream>
#include <functional>

typedef std::function<int(int, int)> func_t;

class Task
{

public:
    Task(){}
    Task(int x, int y, func_t func):x_(x), y_(y), func_(func)
    {}
    int operator ()()
    {
        return func_(x_, y_);
    }
public:
    int x_;
    int y_;
    // int type;
    func_t func_;
};

func_ 保存的是一个可调用对象，它可以是任意接受两个整数参数并返回一个整数结果的函数或者 lambda 表达式。Task 类重载了 () 操作符，使得可以直接通过调用 Task 对象的方式执行 func_ 函数，将 x_ 和 y_ 作为参数传递给 func_。

5、ConProd.cc主函数

在 ConProd.cc 使用 Task 类以及之前提到的 BlockQueue 类实现一个多线程生产者-消费者模型。

#include "BlockQueue.hpp"
#include "Task.hpp"

#include <pthread.h>
#include <unistd.h>
#include <ctime>

int myAdd(int x, int y)
{
    return x + y;
}

void* consumer(void *args)
{
    BlockQueue<Task> *bqueue = (BlockQueue<Task> *)args;
    while(true)
    {
        // 获取任务
        Task t;
        bqueue->pop(&t);
        // 完成任务
        std::cout << pthread_self() <<" consumer: "<< t.x_ << "+" << t.y_ << "=" << t() << std::endl;
        sleep(1);
    }

    return nullptr;
}

void* productor(void *args)
{
    BlockQueue<Task> *bqueue = (BlockQueue<Task> *)args;
    // int
    // int a = 1;
    while(true)
    {
        // 制作任务 -- 不一定是从生产者来的
        int x = rand()%10 + 1;
        usleep(rand()%1000);
        int y = rand()%5 + 1;
        Task t(x, y, myAdd);
        // 生产任务
        bqueue->push(t);
        // 输出消息
        std::cout <<pthread_self() <<" productor: "<< t.x_ << "+" << t.y_ << "=?" << std::endl;
        sleep(1);
    }
    return nullptr;
}

int main()
{
    srand((uint64_t)time(nullptr) ^ getpid() ^ 0x32457);
    BlockQueue<Task> *bqueue = new BlockQueue<Task>();

    pthread_t c[2],p[2];
    pthread_create(c, nullptr, consumer, bqueue);
    //pthread_create(c + 1, nullptr, consumer, bqueue);
    pthread_create(p, nullptr, productor, bqueue);
    //pthread_create(p + 1, nullptr, productor, bqueue);

    pthread_join(c[0], nullptr);
    //pthread_join(c[1], nullptr);
    pthread_join(p[0], nullptr);
    //pthread_join(p[1], nullptr);

    delete bqueue;

    return 0;
}

  • consumer 函数：这是消费者线程的主体函数，它接收一个指向 BlockQueue<Task> 的指针作为参数。消费者线程不断从队列中弹出 Task 对象，并调用 Task 的 () 运算符完成任务，即执行与 Task 关联的函数，并输出运算结果。

  • productor 函数：这是生产者线程的主体函数，同样接收一个指向 BlockQueue<Task> 的指针。生产者线程不断地生成新的 Task 对象，这里的 Task 使用 myAdd 函数作为其计算逻辑，然后将新创建的 Task 推入队列中。

在 main 函数中，程序首先初始化了一个 BlockQueue<Task> 实例，并创建了两个消费者线程和两个生产者线程。每个线程在运行时都会引用同一个队列对象。生产者线程生成随机数并创建 Task，将其推入队列；消费者线程从队列中取出 Task 执行并输出结果。所有线程在完成各自的任务后，主程序通过 pthread_join 等待所有线程结束，并最终释放队列资源。

int main()
{
    srand((uint64_t)time(nullptr) ^ getpid() ^ 0x32457);
    BlockQueue<Task> *bqueue = new BlockQueue<Task>();

    // pthread_t c,p;
    // pthread_create(&c, nullptr, consumer, bqueue);
    // pthread_create(&p, nullptr, productor, bqueue);

    // pthread_join(c, nullptr);
    // pthread_join(p, nullptr);

    pthread_t c[2],p[2];
    pthread_create(c + 1, nullptr, consumer, bqueue);
    pthread_create(p + 1, nullptr, productor, bqueue);

    pthread_join(c[0], nullptr);
    pthread_join(c[1], nullptr);
    pthread_join(p[0], nullptr);
    pthread_join(p[1], nullptr);

    delete bqueue;

    return 0;
}
[hbr@VM-16-9-centos Producer-Consumer]$ ./cp 
要进行加锁
要进行加锁
要进行解锁
140143505307392 productor: 3+5=?
要进行解锁
140143513700096 consumer: 3+5=8
要进行加锁
要进行加锁
要进行解锁
140143505307392 productor: 9+5=?
要进行解锁
140143513700096 consumer: 9+5=14
要进行加锁
要进行加锁
要进行解锁
140143505307392 productor: 6+5=?
要进行解锁
140143513700096 consumer: 6+5=11
要进行加锁
要进行加锁
要进行解锁
140143505307392 productor: 4+3=?
要进行解锁
140143513700096 consumer: 4+3=7
^C
[hbr@VM-16-9-centos Producer-Consumer]$
  • 每次生产者线程向队列中添加任务时，都会显示"要进行加锁"，这是因为push方法内部使用了lockGuard，从而自动锁定互斥锁。当队列不满时，生产者线程会添加任务并显示待解决的加法问题。
  • 随后，消费者线程从队列中取出任务时同样显示"要进行加锁"，然后执行任务并打印出结果。执行完任务后，消费者线程会释放互斥锁，显示"要进行解锁"。

整个流程体现了生产者-消费者模型的典型行为：生产者线程不断地生成任务并放入队列，当队列满时阻塞；消费者线程从队列中取出并执行任务，当队列空时阻塞。通过这种方式，生产者和消费者线程协同工作，保持了线程间的同步和资源的有效利用。由于只有一个消费者和一个生产者线程在运行，所以可以看到它们交替进行任务生产和消费的动作。

6、Makefile

cp:ConProd.cc
	g++ -o $@ $^ -lpthread
.PHONY:clean
clean:
	rm -f cp

相关文章