C语言线程间通信
C11 标准为线程间通信提供了条件变量(condition variable)。线程可以使用条件变量,以等待来自另一个线程的通知,通知告知了指定的条件已被满足。例如,这类通知可能代表某些数据已经准备好进行处理。
条件变量由类型为 cnd_t 的对象表示,并配合互斥一起使用。一般过程如下:线程获得互斥,然后测试条件。如果条件不满足,则线程继续等待条件变量(释放互斥),直到另一个线程再次唤醒它,然后该线程再次获得互斥,并再次测试条件,重复上述过程,直到条件满足。
头文件 threads.h 定义了使用条件变量的函数,它们如下所示:
除 cnd_destroy()以外的所有条件变量函数,如果它们引发错误,则返回值 thrd_error,否则返回值 thrd_success。当时间达到限定值时,函数 cnd_timedwait()也会返回值 thrd_timedout。
例 1 与例 2 中的程序展示了在常见的“生产者-消费者”模型中使用条件变量。程序为每个生产者和消费者开启一个新线程。生产者将一个新产品(在我们的示例中,新产品为一个 int 变量)放入一个环形缓冲区中,假设这个缓冲区没有满,然后通知等待的消费者:产品已经准备好。每个消费者从该缓冲区中取出产品,然后将实际情况通知给正在等待的生产者。
在任一特定时间,只有一个线程可以修改环形缓冲器。因此,在函数 bufPut()和 bufGet()间将存在线程同步问题,函数 bufPut()将一个元素插入到缓冲区,函数 buf-Get()将一个元素从缓冲区移除。
有两个条件变量:生产者等待其中一个条件变量,以判断缓冲器是否满了;消费者等待另一个条件变量,以判断缓冲器是否空了。缓冲区的所有必需元素都包括在结构 Buffer 中。函数 bufInit()初始化具有指定大小的 Buffer 对象,而函数 bufDestroy()销毁 Buffer 对象。
【例1】用于“生产者-消费者”模型的环形缓冲区
例 2 中的 main()函数创建了一个缓冲区,并启动了若干个生产者和消费者线程,给予每个线程一个识别号码和一个指向缓冲区的指针。每个生产者线程创建一定数量的“产品”,然后用一个 return 语句退出。一个消费者线程如果在给定延时期间无法获得产品以进行消费,则直接返回。
【例2】启动生产者和消费者线程
条件变量由类型为 cnd_t 的对象表示,并配合互斥一起使用。一般过程如下:线程获得互斥,然后测试条件。如果条件不满足,则线程继续等待条件变量(释放互斥),直到另一个线程再次唤醒它,然后该线程再次获得互斥,并再次测试条件,重复上述过程,直到条件满足。
头文件 threads.h 定义了使用条件变量的函数,它们如下所示:
int cnd_init(cnd_t*cond);初始化 cond 引用的条件变量。
void cnd_destroy(cnd_t*cond);释放指定条件变量使用的所有资源。
int cnd_signal(cnd_t*cond);在等待指定条件变量的任意数量的线程中,唤醒其中一个线程。
int cnd_broadcast(cnd_t*cond);唤醒所有等待指定条件变量的线程。
int cnd_wait(cnd_t*cond,mtx_t*mtx);阻塞正在调用的线程,并释放指定的互斥。在调用 cnd_wait()之前,线程必须持有互斥。如果另一线程通过发送一个信号解除当前线程的阻塞(也就是说,通过指定同样的条件变量作为参数调用 cond_signal()或 cnd_broadcast()),那么调用 cnd_wait()的线程在 cnd_wait()返回之前会再次获得互斥。
int cnd_timedwait(cnd_t*restrict cond,mtx_t*restrict mtx,const struct timespec*restrict ts);与 cnd_wait()类似,cnd_timedwait()阻塞调用它们的线程,但仅维持由参数 ts 指定的时间。可以通过调用函数 timespec_get()获得一个 struct timespec 对象,它表示当前时间。
除 cnd_destroy()以外的所有条件变量函数,如果它们引发错误,则返回值 thrd_error,否则返回值 thrd_success。当时间达到限定值时,函数 cnd_timedwait()也会返回值 thrd_timedout。
例 1 与例 2 中的程序展示了在常见的“生产者-消费者”模型中使用条件变量。程序为每个生产者和消费者开启一个新线程。生产者将一个新产品(在我们的示例中,新产品为一个 int 变量)放入一个环形缓冲区中,假设这个缓冲区没有满,然后通知等待的消费者:产品已经准备好。每个消费者从该缓冲区中取出产品,然后将实际情况通知给正在等待的生产者。
在任一特定时间,只有一个线程可以修改环形缓冲器。因此,在函数 bufPut()和 bufGet()间将存在线程同步问题,函数 bufPut()将一个元素插入到缓冲区,函数 buf-Get()将一个元素从缓冲区移除。
有两个条件变量:生产者等待其中一个条件变量,以判断缓冲器是否满了;消费者等待另一个条件变量,以判断缓冲器是否空了。缓冲区的所有必需元素都包括在结构 Buffer 中。函数 bufInit()初始化具有指定大小的 Buffer 对象,而函数 bufDestroy()销毁 Buffer 对象。
【例1】用于“生产者-消费者”模型的环形缓冲区
/* buffer.h * 用于线程安全缓冲区的所有声明 */ #include <stdbool.h> #include <threads.h> typedef struct Buffer { int *data; // 指向数据数组的指针 size_t size, count; // 元素数量的最大值和当前值 size_t tip, tail; // tip = 下一个空点的索引 mtx_t mtx; // 一个互斥 cnd_t cndPut, cndGet; // 两个条件变量 } Buffer; bool bufInit( Buffer *bufPtr, size_t size ); void bufDestroy(Buffer *bufPtr); bool bufPut(Buffer *bufPtr, int data); bool bufGet(Buffer *bufPtr, int *dataPtr, int sec); /* ------------------------------------------------------------- * buffer.c * 定义用于处理Buffer的函数 */ #include "buffer.h" #include <stdlib.h> // 为了使用malloc()和free() bool bufInit( Buffer *bufPtr, size_t size) { if ((bufPtr->data = malloc( size * sizeof(int))) == NULL) return false; bufPtr->size = size; bufPtr->count = 0; bufPtr->tip = bufPtr->tail = 0; return mtx_init( &bufPtr->mtx, mtx_plain) == thrd_success && cnd_init( &bufPtr->cndPut) == thrd_success && cnd_init( &bufPtr->cndGet) == thrd_success; } void bufDestroy(Buffer *bufPtr) { cnd_destroy( &bufPtr->cndGet ); cnd_destroy( &bufPtr->cndPut ); mtx_destroy( &bufPtr->mtx ); free( bufPtr->data ); } // 在缓冲区中插入一个新元素 bool bufPut(Buffer *bufPtr, int data) { mtx_lock( &bufPtr->mtx ); while (bufPtr->count == bufPtr->size) if (cnd_wait( &bufPtr->cndPut, &bufPtr->mtx ) != thrd_success) return false; bufPtr->data[bufPtr->tip] = data; bufPtr->tip = (bufPtr->tip + 1) % bufPtr->size; ++bufPtr->count; mtx_unlock( &bufPtr->mtx ); cnd_signal( &bufPtr->cndGet ); return true; } // 从缓冲区中移除一个元素 // 如果缓冲区是空的,则等待不超过sec秒 bool bufGet(Buffer *bufPtr, int *dataPtr, int sec) { struct timespec ts; timespec_get( &ts, TIME_UTC ); // 当前时间 ts.tv_sec += sec; // + sec秒延时 mtx_lock( &bufPtr->mtx ); while ( bufPtr->count == 0 ) if (cnd_timedwait(&bufPtr->cndGet, &bufPtr->mtx, &ts) != thrd_success) return false; *dataPtr = bufPtr->data[bufPtr->tail]; bufPtr->tail = (bufPtr->tail + 1) % bufPtr->size; --bufPtr->count; mtx_unlock( &bufPtr->mtx ); cnd_signal( &bufPtr->cndPut ); return true; }
例 2 中的 main()函数创建了一个缓冲区,并启动了若干个生产者和消费者线程,给予每个线程一个识别号码和一个指向缓冲区的指针。每个生产者线程创建一定数量的“产品”,然后用一个 return 语句退出。一个消费者线程如果在给定延时期间无法获得产品以进行消费,则直接返回。
【例2】启动生产者和消费者线程
// producer_consumer.c #include "buffer.h" #include <stdio.h> #include <stdlib.h> #define NP 2 // 生产者的数量 #define NC 3 // 消费者的数量 int producer(void *); // 线程函数 int consumer(void *); struct Arg { int id; Buffer *bufPtr; }; // 线程函数的参数 _Noreturn void errorExit(const char* msg) { fprintf(stderr, "%s\n", msg); exit(0xff); } int main(void) { printf("Producer-Consumer Demo\n\n"); Buffer buf; // 为5个产品创建一个缓冲区 bufInit( &buf, 5 ); thrd_t prod[NP], cons[NC]; // 线程 struct Arg prodArg[NP], consArg[NC]; // 线程的参数 int i = 0, res = 0; for ( i = 0; i < NP; ++i ) // 启动生产者 { prodArg[i].id = i+1, prodArg[i].bufPtr = &buf; if (thrd_create( &prod[i], producer, &prodArg[i] ) != thrd_success) errorExit("Thread error."); } for ( i = 0; i < NC; ++i ) // 启动消费者 { consArg[i].id = i+1, consArg[i].bufPtr = &buf; if ( thrd_create( &cons[i], consumer, &consArg[i] ) != thrd_success) errorExit("Thread error."); } for ( i = 0; i < NP; ++i ) // 等待线程结束 thrd_join(prod[i], &res), printf("\nProducer %d ended with result %d.\n", prodArg[i].id, res); for ( i = 0; i < NC; ++i ) thrd_join(cons[i], &res), printf("Consumer %d ended with result %d.\n", consArg[i].id, res); bufDestroy( &buf ); return 0; } int producer(void *arg) // 生产者线程函数 { struct Arg *argPtr = (struct Arg *)arg; int id = argPtr->id; Buffer *bufPtr = argPtr->bufPtr; int count = 0; for (int i = 0; i < 10; ++i) { int data = 10*id + i; if (bufPut( bufPtr, data )) printf("Producer %d produced %d\n", id, data), ++count; else { fprintf( stderr, "Producer %d: error storing %d\n", id, data); return -id; } } return count; } int consumer(void *arg) // 消费者线程函数 { struct Arg *argPtr = (struct Arg *)arg; int id = argPtr->id; Buffer *bufPtr = argPtr->bufPtr; int count = 0; int data = 0; while (bufGet( bufPtr, &data, 2 )) { ++count; printf("Consumer %d consumed %d\n", id, data); } return count; }
所有教程
- socket
- Python基础教程
- C#教程
- MySQL函数
- MySQL
- C语言入门
- C语言专题
- C语言编译器
- C语言编程实例
- GCC编译器
- 数据结构
- C语言项目案例
- C++教程
- OpenCV
- Qt教程
- Unity 3D教程
- UE4
- STL
- Redis
- Android教程
- JavaScript
- PHP
- Mybatis
- Spring Cloud
- Maven
- vi命令
- Spring Boot
- Spring MVC
- Hibernate
- Linux
- Linux命令
- Shell脚本
- Java教程
- 设计模式
- Spring
- Servlet
- Struts2
- Java Swing
- JSP教程
- CSS教程
- TensorFlow
- 区块链
- Go语言教程
- Docker
- 编程笔记
- 资源下载
- 关于我们
- 汇编语言
- 大数据
- 云计算
- VIP视频