Linux - 多线程实现生产者与消费者模式

一、生产者-消费者模式的简介

在实际的软件开发过程中，我们将产生数据的模块称为生产者，处理数据的模块成为消费者。但仅有这两者还不够成为一个生产者-消费者模式，还需要有一个缓冲区（一段内存区域）作为中介，生产者产生的数据放入缓冲区，消费者从缓冲区读取数据并处理。（注：上述所说的模块是广义的，可以是类，函数，线程，进程等）

我们可以将这二者之间的关系图表示出来：

总结：我们用3-2-1的方法来简单描述一个生产者-消费者模式，3代表有三种关系：生产者-生产者，消费者-消费者，生产者-消费者；2代表两个角色：生产者，消费者；1代表一个交易场所：缓冲区

二、生产者-消费者模式之间的关系

模拟实现生产者-消费者模式之前，我们需要先捋清除这之间的关系：

生产者-生产者：很明显，这两者之间必定是一种竞争关系，也就是说一个生产者往缓冲区放数据时另一个生产者就不能去访问这块空间

消费者-消费者：同样，两个消费者之间也是竞争的关系，这就好比两个人同时看中一件商品时，他们之间就是一种竞争的关系

生产者-消费者：生产者与消费者之间其实是一种同步与互斥的关系，假设只有一个生产者一个消费者时，只有生产者放入数据后消费者才能读取，消费者拿到数据后生产者才去生产，这就是一种同步；但当生产者生产数据的时候消费者就不能从缓冲区拿数据，或者消费者读数据的时候生产者就不能往缓冲区里写数据，否则很可能会导致两者都存/取数据失败，产生二义性问题。

三、如何实现单生产者-单消费者模型

对于单生产者-单消费者模式来说，我们可以用单链表来模拟一个交易场所，生产数据就相当于往链表里面插入结点，消费者读取数据就相当于删除链表结点（这里我采用的是头插头删的方式，当然也可以用尾插、尾删的方式实现）

既然已经确定了生产场所，其次我们要考虑到如何才能让这二者时间实现同步与互斥。多线程编程中提到了“条件变量（Condition Variable）”，在此，我们先给出几个函数：

初始化和销毁：

int pthread_cond_destroy(pthread_cond_t *cond);
int pthread_cond_init(pthread_cond_t *restrict cond,const pthread_condattr_t *restrict attr);
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
//attr参数为NULL，表示缺省属性。如果Condition Variable是静态分配的,也可以用宏定义PTHEAD_COND_INITIALIZER初始化,相当于用pthread_cond_init函数初始化并且attr参数为NULL。

操作函数：

int pthread_cond_broadcast(pthread_cond_t *cond);
//一次唤醒在某个Condition Variable上等待的所有线程

int pthread_cond_signal(pthread_cond_t *cond);
//一次唤醒在某个Condition Variable上等待的另一个线程

int pthread_cond_timedwait(pthread_cond_t *restrict cond,pthread_mutex_t *restrict mutex,
const struct timespec *restrict abstime);
//pthread_cond_timedwait函数还有一个额外的参数来设定等待超时，如果到达了abstime所指定的时刻仍然没有别的线程来唤醒当前线程，就返回ETIMEDOUT。

int pthread_cond_wait(pthread_cond_t *restrict cond,pthread_mutex_t *restrict mutex);
//调用pthread_cond_wait在一个Condition Variable上阻塞等待，这个函数做以下三步操作：
1.释放Mutex
2.阻塞等待
3.当被唤醒时,重新获得Mutex并返回

我们可以利用条件变量中的相关函数来模拟实现，当生产者线程生产数据时消费者线程就wait，当被唤醒后，此时链表已经有数据（消费者等待成功），拿走结点；消费者读取数据的过程中生产者在wait，等到消费者读取数据完之后被唤醒，然后继续往缓冲区写入数据（循环………），等待与被唤醒的过程就可以利用上述的pthread_cond_wait和pthread_cond_signal函数实现。

具体代码如下：

pthread_mutex_t mylock = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t mycond = PTHREAD_COND_INITIALIZER;

typedef struct _node
{
int data;
struct _node* next;
}node,*pnode,**ppnode;

pnode alloc_node(int data)
{
pnode _n = (pnode)malloc(sizeof(node));
if( _n == NULL )
{
perror("malloc");
return NULL;
}
_n->data = data;
_n->next = NULL;
return _n;
}

void InitList(ppnode _h)
{
*_h = alloc_node(0);
}

void PushFront(pnode _h,int data)
{
assert(_h);
pnode ret = alloc_node(data);
ret->next = _h->next;
_h->next = ret;
}
void DelNode(pnode node)
{
assert(node);
free(node);
}
void PopFront(pnode _h,int* data)
{
assert(_h);
assert(data);
if(_h->next == NULL)
{
printf("list is empty!\n");
return ;
}

pnode del = _h->next;
_h->next = del->next;
*data = del->data;
DelNode(del);
}

void destory(pnode _h)
{
assert(_h);
pnode start = _h;
//  pnode del = _h;
int data = 0;
while(start->next)
{
PopFront(start,&data);
//      del = start;
//      start = del->next;
//      free(del);
//      del = NULL;
}
DelNode(_h);
}

void show(pnode _h)
{
assert(_h);
pnode start = _h->next;
while(start)
{
printf("%d ",start->data);
start = start->next;
}
printf("\n");
}

void* thread_product(void* arg)
{
pnode node = (pnode)arg;
while(1)
{
usleep(123123);
int data = rand()%10000;
pthread_mutex_lock(&mylock);
PushFront(node,data);
pthread_mutex_unlock(&mylock);
pthread_cond_signal(&mycond);
printf("product:%d \n",data);
}
}

void* thread_consumer(void* arg)
{
pnode node = (pnode)arg;
int data = 0;
while(1)
{
usleep(1000);
pthread_mutex_lock(&mylock);
while(node->next == NULL)
{
pthread_cond_wait(&mycond,&mylock);
}
PopFront(node,&data);
pthread_mutex_unlock(&mylock);
printf("consumer:%d \n",data);
}
}

void test_list(pnode head)
{
int i = 0;
for(; i<10; i++)
{
PushFront(head,i);
show(head);
}
int data = 0;
for(; i>=0; i--)
{
PopFront(head,&data);
show(head);
}
destory(head);
show(head);
}

int main()
{
pnode head = NULL;
InitList(&head);
//  test_list(head);

pthread_t tid1;
pthread_t tid2;

pthread_create(&tid1,NULL,thread_product,(void*)head);
pthread_create(&tid2,NULL,thread_consumer,(void*)head);

pthread_join(tid1,NULL);
pthread_join(tid2,NULL);

pthread_mutex_destroy(&mylock);
pthread_cond_destroy(&mycond);
return 0;
}