多线程--生产者/消费者线程模型

Posted 2022-01-15 dianrain

//程序演进1
//thread loop，忙等Busy wait
//不断的检查是不是该做什么事情了；为了减少CPU占用，sleep睡眠一会
//while (1)
//
//    do_something(); 
//    sleep(time);
//

//程序演进2
//while (1)
//
//    read_form_intput();
//    do_something();
//

//程序员演进3，消息模型
//while (waitForMsg)
//
//    if (isQuitMsg() == false)
//    
//        handleMsg();
//    
//

//线程模型--生产者/消费者
//Thread A（Producer）往消息队列（Message Quenue）加消息
//Thread B（Consumer）从消息队列取消息

//演进：
//        消息队列变事件队列
//        消息队列变缓存队列
//        一读多写（一个消费者，多个生产者）；多读多写（多个消费者和多个生产者）
//        流量控制（生产速度和消费速度的控制）

//演进：线程池
//任务队列（消息队列）
//多个发布任务者（生产者），多个处理任务者（消费者）



//程序演进4，I/O处理
//while (1)
//
//    slect / epoll();
//    handleIORrequest();
//


//I/O模型
//accept默认为阻塞式--需要解决忙等待问题
//解决思路：创建线程来read处理，演进到线程池

#include "pch.h"
#include <iostream>
#include "CProdcons.h"
#include <pthread.h>

struct prodcons

    // 缓冲区相关数据结构
    int buffer[BUFFER_SIZE]; /* 实际数据存放的数组*/
    pthread_mutex_t lock; /* 互斥体lock 用于对缓冲区的互斥操作 */
    int readpos, writepos; /* 读写指针*/
    pthread_cond_t notempty; /* 缓冲区非空的条件变量 */
    pthread_cond_t notfull; /* 缓冲区未满的条件变量 */
;


void init(struct prodcons *b)

    pthread_mutex_init(&b->lock, NULL);
    pthread_cond_init(&b->notempty, NULL);
    pthread_cond_init(&b->notfull, NULL);
    b->readpos = 0;
    b->writepos = 0;

/* 将产品放入缓冲区,这里是存入一个整数*/
void put(struct prodcons *b, int data)

    pthread_mutex_lock(&b->lock);
    /* 等待缓冲区未满*/
    if ((b->writepos + 1) % BUFFER_SIZE == b->readpos)
    
        pthread_cond_wait(&b->notfull, &b->lock);
    
    /* 写数据,并移动指针 */
    b->buffer[b->writepos] = data;
    b->writepos++;
    if (b->writepos >= BUFFER_SIZE)
        b->writepos = 0;
    /* 设置缓冲区非空的条件变量*/
    pthread_cond_signal(&b->notempty);
    pthread_mutex_unlock(&b->lock);

/* 从缓冲区中取出整数*/
int get(struct prodcons *b)

    int data;
    pthread_mutex_lock(&b->lock);
    /* 等待缓冲区非空*/
    if (b->writepos == b->readpos)
    
        pthread_cond_wait(&b->notempty, &b->lock);
    
    /* 读数据,移动读指针*/
    data = b->buffer[b->readpos];
    b->readpos++;
    if (b->readpos >= BUFFER_SIZE)
        b->readpos = 0;
    /* 设置缓冲区未满的条件变量*/
    pthread_cond_signal(&b->notfull);
    pthread_mutex_unlock(&b->lock);
    return data;



#define OVER ( - 1)
struct prodcons buffer;
void *producer(void *data)

    int n;
    for (n = 0; n < 10000; n++)
    
        printf("%d --->\n", n);
        put(&buffer, n);
     put(&buffer, OVER);
    return NULL;


void *consumer(void *data)

    int d;
    while (1)
    
        d = get(&buffer);
        if (d == OVER)
            break;
        printf("--->%d \n", d);
    
    return NULL;




int main()

    pthread_t th_a, th_b;
    void *retval;
    init(&buffer);
    /* 创建生产者和消费者线程*/
    pthread_create(&th_a, NULL, producer, 0);
    pthread_create(&th_b, NULL, consumer, 0);
    /* 等待两个线程结束*/
    pthread_join(th_a, &retval);
    pthread_join(th_b, &retval);
    return 0;