Linux C/C++编程——线程

线程是允许应用程序并发执行多个任务的一种机制，线程参与系统调度。

系统调度的最小单元是线程、而并非进程。

线程包含在进程之中，是进程中的实际运行单位。一个线程指的是进程中一个单一顺序的控制流（或者说是执行路线、执行流），一个进程中可以创建多个线程，多个线程实现并发运行，每个线程执行不同的任务。

同一进程中的各个线程，可以共享该进程所拥有的资源。

线程和子进程

子进程的缺点：

• 进程间切换开销大。
• 进程间通信较为麻烦。

多线程的有点：

•  同一进程的多个线程间切换开销比较小。
• 同一进程的多个线程间通信容易。

其它的概念，例如：并发并行什么的，具体查阅操作系统。

一、线程 

1. 创建线程

#include <pthread.h>
int pthread_create(pthread_t *thread, const pthread_attr_t *attr, void *(*start_routine) (void *), void *arg);

函数参数和返回值含义如下：

        thread： pthread_t 类型指针，当 pthread_create() 成功返回时，新创建的线程的线程 ID 会保存在参数 thread 所指向的内存中，后续的线程相关函数会使用该标识来引用此线程。

        attr： pthread_attr_t 类型指针，指向 pthread_attr_t 类型的缓冲区，pthread_attr_t 数据类型定义了线程的各种属性。如果将参数 attr 设置为 NULL，那么表示将线程的所有属性设置为默认值，以此创建新线程。

        start_routine： 参数 start_routine 是一个函数指针，指向一个函数，新创建的线程从 start_routine()函数开始运行，该函数返回值类型为 void * ，并且该函数的参数只有一个 void * ，其实这个参数就是pthread_create() 函数的第四个参数 arg 。

        arg： 传递给 start_routine() 函数的参数。一般情况下，需要将 arg 指向一个全局或堆变量。将参数 arg 设置为 NULL ，表示不需要传入参数给 start_routine() 函数。

案例1：pthread_create() 创建线程

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <string.h>
#include <unistd.h>
#include <sys/types.h>
#include <unistd.h>
static void *new_thread_start(void *arg)
{
 printf("新线程: 进程 ID<%d> 线程 ID<%lu>\n", getpid(), pthread_self());
 return (void *)0;
}
int main(void)
{
 pthread_t tid;
 int ret;
 ret = pthread_create(&tid, NULL, new_thread_start, NULL);
 if (ret) {
 fprintf(stderr, "Error: %s\n", strerror(ret));
 exit(-1);
 }
 printf("主线程: 进程 ID<%d> 线程 ID<%lu>\n", getpid(), pthread_self());
 sleep(1);
 exit(0);
}

编译的时候，需要加上-lpthread。因为 pthread 不在 gcc 的默认链接库中。

gcc -o pthread_create pthread_create.c -lpthread

2. 终止线程

案例2 pthread_exit() 终止线程

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <string.h>
#include <unistd.h>
#include <sys/types.h>
#include <unistd.h>

static void *new_thread_start(void *arg)
{
    printf("新线程 start\n");
    sleep(1);
    printf("新线程 end\n");
    /* 终止线程 */
    pthread_exit(NULL);
}

int main(void)
{
    pthread_t tid;
    int ret;
    ret = pthread_create(&tid, NULL, new_thread_start, NULL);
    if (ret){
        fprintf(stderr, "Error: %s\n", strerror(ret));
        exit(-1);
    }
    printf("主线程 end\n");
    pthread_exit(NULL);
    exit(0);
}

最三行的输出，会延迟一段时间输出。

就算主线程调用终止，整个进程也不会结束，即新线程还在继续运行。

3. 回收线程

在父、子进程当中，父进程可通过 wait() 函数（或其变体 waitpid()）阻塞等待子进程退出并获取其终止状态，回收子进程资源；

而在线程当中，也需要如此，通过调用 pthread_join()函数来阻塞等待线程的终止，并获取线程的退出码，回收线程资源；

#include <pthread.h>
int pthread_join(pthread_t thread, void **retval);

        thread： pthread_join() 等待指定线程的终止，通过参数 thread （线程 ID ）指定需要等待的线程；

        retval： 如果参数 retval 不为 NULL ，则 pthread_join()将目标线程的退出状态（即目标线程通过pthread_exit() 退出时指定的返回值或者在线程 start 函数中执行 return 语句对应的返回值）复制到 *retval 所指向的内存区域；

        如果目标线程被 pthread_cancel() 取消，则将 PTHREAD_CANCELED 放在 *retval 中。如果对目标线程的终止状态不感兴趣，则可将参数 retval 设置为 NULL 。

         返回值： 成功返回 0 ；失败将返回错误码。

案例3： pthread_join()等待线程终止

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <string.h>
#include <unistd.h>
#include <sys/types.h>
#include <unistd.h>
static void *new_thread_start(void *arg)
{
 printf("新线程 start\n");
 sleep(2);
 printf("新线程 end\n");
 pthread_exit((void *)10);
}
int main(void)
{
 pthread_t tid;
 void *tret;
 int ret;
 ret = pthread_create(&tid, NULL, new_thread_start, NULL);
 if (ret) {
 fprintf(stderr, "pthread_create error: %s\n", strerror(ret));
 exit(-1);
 }
 ret = pthread_join(tid, &tret);
 if (ret) {
 fprintf(stderr, "pthread_join error: %s\n", strerror(ret));
 exit(-1);
 }
 printf("新线程终止, code=%ld\n", (long)tret);
 exit(0);
}

4. 取消线程

案例4：pthread_cancel()取消线程

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <string.h>
#include <unistd.h>
#include <sys/types.h>
#include <unistd.h>

static void *new_thread_start(void *arg)
{
 printf("新线程--running\n");
    for ( ; ; )
        sleep(1);
        return (void *)0;
}

int main(void)
{
    pthread_t tid;
    void *tret;
    int ret;
    /* 创建新线程 */
    ret = pthread_create(&tid, NULL, new_thread_start, NULL);
    if (ret) {
        fprintf(stderr, "pthread_create error: %s\n", strerror(ret));
        exit(-1);
    }
    sleep(1);

    /* 向新线程发送取消请求 */
    ret = pthread_cancel(tid);
    if (ret) {
        fprintf(stderr, "pthread_cancel error: %s\n", strerror(ret));
        exit(-1);
    }

     /* 等待新线程终止 */
    ret = pthread_join(tid, &tret);
    if (ret) {
        fprintf(stderr, "pthread_join error: %s\n", strerror(ret));
        exit(-1);
    }

    printf("新线程终止, code=%ld\n", (long)tret);
    exit(0);
}

当主线程发送取消请求之后，新线程便退出了，而且退出码为 -1，也就是PTHREAD_CANCELED 。

二、线程同步

线程同步是为了对共享资源的访问进行保护。

案例1  两个线程并发访问同一全局变量

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <unistd.h>
#include <string.h>

static int g_count = 0;
static void *new_thread_start(void *arg)
{
    int loops = *((int *)arg);
    int l_count, j;
    for (j=0; j<loops; j++){
        l_count = g_count;
        l_count++;
        g_count = l_count;
    }
    return (void *)0;
}

static int loops;
int main(int argc, char *argv[])
{
    pthread_t tid1, tid2;
    int ret;
    /* 获取用户传递的参数 */
    if(2>argc)
        loops =10000000;
    else
        loops = atoi(argv[1]);
    /* 创建 2 个新线程*/
    ret = pthread_create(&tid1, NULL, new_thread_start, &loops);
    if(ret){
        fprintf(stderr, "pthread_create error: %s\n", strerror(ret));
        exit(-1);
    }

    ret = pthread_create(&tid2, NULL, new_thread_start, &loops);
    if (ret) {
        fprintf(stderr, "pthread_create error: %s\n", strerror(ret));
        exit(-1);
    }

     /* 等待线程结束 */
    ret = pthread_join(tid1, NULL);
    if (ret) {
        fprintf(stderr, "pthread_join error: %s\n", strerror(ret));
        exit(-1);
    }
    ret = pthread_join(tid2, NULL);
    if (ret) {
        fprintf(stderr, "pthread_join error: %s\n", strerror(ret));
        exit(-1);
    }

     /* 打印结果 */
    printf("g_count = %d\n", g_count);
    exit(0);
 
}

1. 互斥锁

所谓的信号量、互斥锁什么的，考完408后，都十分熟悉，但是都停留在伪代码阶段。

1.1 初始化互斥锁

宏初始化互斥锁

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

pthread_mutex_init() 函数初始化互斥锁

#include <pthread.h>
int pthread_mutex_init(pthread_mutex_t *mutex, const pthread_mutexattr_t *attr);

        mutex： 参数 mutex 是一个 pthread_mutex_t 类型指针，指向需要进行初始化操作的互斥锁对象；

        attr： 参数 attr 是一个 pthread_mutexattr_t 类型指针，指向一个 pthread_mutexattr_t 类型对象，该对象用于定义互斥锁的属性， 若将参数 attr 设置为 NULL ，则表示将互斥锁的属性设置为默认值。

1.2 互斥锁加锁和解锁

调用函数 pthread_mutex_lock()可以对互斥锁加锁、获取互斥锁，

而调用函数 pthread_mutex_unlock() 可以对互斥锁解锁、释放互斥锁。

案例2： 使用互斥锁保护全局变量的访问

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <unistd.h>
#include <string.h>

static pthread_mutex_t mutex;
static int g_count=0;

static void *new_thread_start(void *arg)
{
    int loops = *((int *)arg);
    int l_count, j;
    for(j = 0;j<loops;j++){
        /*上锁*/
        pthread_mutex_lock(&mutex);

        l_count = g_count;
        l_count++;
        g_count = l_count;

        /*解锁*/
        pthread_mutex_unlock(&mutex);
    }
    return (void *)0;
}

static int loops;
int main(int argc, char *argv[])
{
    pthread_t tid1, tid2;
    int ret;
    /* 获取用户传递的参数 */
    if (2 > argc)
        loops = 10000000; //没有传递参数默认为 1000 万次
    else
        loops = atoi(argv[1]);
    /* 初始化互斥锁 */
    pthread_mutex_init(&mutex, NULL);
    /* 创建 2 个新线程 */
    ret = pthread_create(&tid1, NULL, new_thread_start, &loops);
    if (ret) {
        fprintf(stderr, "pthread_create error: %s\n", strerror(ret));
        exit(-1);
    }
    ret = pthread_create(&tid2, NULL, new_thread_start, &loops);
    if (ret) {
        fprintf(stderr, "pthread_create error: %s\n", strerror(ret));
        exit(-1);
    }
    /* 等待线程结束 */
    ret = pthread_join(tid1, NULL);
    if (ret) {
        fprintf(stderr, "pthread_join error: %s\n", strerror(ret));
        exit(-1);
    }
    ret = pthread_join(tid2, NULL);
    if (ret) {
        fprintf(stderr, "pthread_join error: %s\n", strerror(ret));
        exit(-1);
    }
    /* 打印结果 */
    printf("g_count = %d\n", g_count);
    exit(0);
}

1.3 非阻塞加锁

当互斥锁已经被其它线程锁住时，调用 pthread_mutex_lock()函数会被阻塞，直到互斥锁解锁。

调用 pthread_mutex_trylock()加锁失败，不会阻塞。

案例3：非阻塞加锁

 while(pthread_mutex_trylock(&mutex)); //以非阻塞方式上锁
 l_count = g_count;
 l_count++;
 g_count = l_count;
 pthread_mutex_unlock(&mutex);//互斥锁解锁

1.4 销毁互斥锁

/* 销毁互斥锁 */
 pthread_mutex_destroy(&mutex)

1.5 死锁

// 线程 A
pthread_mutex_lock(mutex1);
pthread_mutex_lock(mutex2);
// 线程 B
pthread_mutex_lock(mutex2);
pthread_mutex_lock(mutex1);

A锁1，B锁2，导致A要锁2的时候阻塞，B要锁1的时候阻塞。

1.6 互斥锁的属性

互斥锁的类型属性控制着互斥锁的锁定特性，一共有 4 中类型：

• PTHREAD_MUTEX_NORMAL ： 一种标准的互斥锁类型，不做任何的错误检查或死锁检测。
• PTHREAD_MUTEX_ERRORCHECK ： 此类互斥锁会提供错误检查。
• PTHREAD_MUTEX_RECURSIVE ：此类互斥锁允许同一线程在互斥锁解锁之前对该互斥锁进行多次加锁，然后维护互斥锁加锁的次数，把这种互斥锁称为递归互斥锁。
• PTHREAD_MUTEX_DEFAULT ： 此类互斥锁提供默认的行为和特性 。

使用方式：

pthread_mutex_t mutex;
pthread_mutexattr_t attr;
/* 初始化互斥锁属性对象 */
pthread_mutexattr_init(&attr);
/* 将类型属性设置为 PTHREAD_MUTEX_NORMAL */
pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_NORMAL);
/* 初始化互斥锁 */
pthread_mutex_init(&mutex, &attr);
......
/* 使用完之后 */
pthread_mutexattr_destroy(&attr);
pthread_mutex_destroy(&mutex);

2. 条件变量

几种模式不赘述了，考过408的都懂，详细看操作系统。

案例4：无条件变量，生产者---消费者

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <unistd.h>
#include <string.h>
static pthread_mutex_t mutex;
static int g_avail = 0;

/* 消费者 线程*/
static void *consumer_thread(void *arg)
{
    for ( ; ; ){
        pthread_mutex_lock(&mutex);

        while (g_avail > 0)
        {
            g_avail--;  //消费
            
            pthread_mutex_unlock(&mutex);
        }
        return (void *)0;
    }
}

/* 主线程（生产者）*/
int main(int argc, char *argv[])
{
    pthread_t tid;
    int ret;
    /*初始化互斥锁*/
    pthread_mutex_init(&mutex,  NULL);
    /* 创建新线程*/
    ret = pthread_create(&tid, NULL, consumer_thread, NULL);
    if(ret){
        fprintf(stderr, "pthread_create error: %s\n", strerror(ret));
        exit(-1);
    }
    for( ; ; ){
        pthread_mutex_lock(&mutex);
        g_avail++;  //生产者
        pthread_mutex_unlock(&mutex);
    }
    exit(0);
}

上述代码虽然可行，但由于新线程中会不停的循环检查全局变量 g_avail 是否大于 0 ，故而造成 CPU 资源的浪费。

2.1 条件变量初始化

#include <pthread.h>
int pthread_cond_destroy(pthread_cond_t *cond);
int pthread_cond_init(pthread_cond_t *cond, const pthread_condattr_t *attr);

2.2 条件变量的通知和等待

#include <pthread.h>
int pthread_cond_broadcast(pthread_cond_t *cond);
int pthread_cond_signal(pthread_cond_t *cond);

int pthread_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex);

案例5：使用条件变量，生产者---消费者

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <unistd.h>
#include <string.h>
static pthread_mutex_t mutex; //定义互斥锁
static pthread_cond_t cond; //定义条件变量
static int g_avail = 0; //全局共享资源
/* 消费者线程 */
static void *consumer_thread(void *arg)
{
 for ( ; ; ) {
 pthread_mutex_lock(&mutex);//上锁
 while (0 >= g_avail)
 pthread_cond_wait(&cond, &mutex);//等待条件满足
 while (0 < g_avail)
 g_avail--; //消费
 pthread_mutex_unlock(&mutex);//解锁
 }
 return (void *)0;
}
/* 主线程（生产者） */
int main(int argc, char *argv[])
{
 pthread_t tid;
 int ret;
 /* 初始化互斥锁和条件变量 */
 pthread_mutex_init(&mutex, NULL);
 pthread_cond_init(&cond, NULL);
 /* 创建新线程 */
 ret = pthread_create(&tid, NULL, consumer_thread, NULL);
if (ret) {
 fprintf(stderr, "pthread_create error: %s\n", strerror(ret));
 exit(-1);
 }
 for ( ; ; ) {
 pthread_mutex_lock(&mutex);//上锁
 g_avail++; //生产
 pthread_mutex_unlock(&mutex);//解锁
 pthread_cond_signal(&cond);//向条件变量发送信号
 }
 exit(0);
}

三、Qt 的多线程

1. 多线程实现

Qt 有 QThread 类，下面是案例。

案例1：多线程打印数字

1. 声明控件

//threaddlg.h

class ThreadDlg : public QDialog
{
    Q_OBJECT

public:
    ThreadDlg(QWidget *parent = 0);
    ~ThreadDlg();
private:
    QPushButton *startBtn;
    QPushButton *stopBtn;
    QPushButton *quitBtn;
}

//threaddlg.cpp

ThreadDlg::ThreadDlg(QWidget *parent)
    : QDialog(parent)
{
    setWindowTitle(tr("线程"));
    startBtn = new QPushButton(tr("开始"));
    stopBtn = new QPushButton(tr("停止"));
    quitBtn = new QPushButton(tr("退出"));
    QHBoxLayout *mainLayout = new QHBoxLayout(this);
    mainLayout->addWidget(startBtn);
    mainLayout->addWidget(stopBtn);
    mainLayout->addWidget(quitBtn);
}

2. 打印功能

//workthread.h

#include <QThread>
class WorkThread : public QThread
{
    Q_OBJECT
public:
    WorkThread();
protected:
    void run();
};

 WorkThread 类继承自 QThread 类，run()函数需要重新实现。

//workthread.cpp

#include "workthread.h"
#include <QtDebug>
WorkThread::WorkThread()
{

}
void WorkThread::run()
{
    while(true)
    {
        for(int n=0;n<10;n++)
            qDebug()<<n<<n<<n<<n<<n<<n<<n<<n;
    }
}

run() 函数：不断打印0~9数字，且每一个数字重复8次。

3. 槽函数

WorkThread

//threaddlg.h
private:
    WorkThread *workThread[MAXSIZE];
public slots:
    void slotStart();						//槽函数用于启动线程
    void slotStop();						//槽函数用于终止线程

指向工作线程的私有指针数组 workThread，记录了所启动的全部线程。 

连接槽函数：

//threaddlg.cpp

    connect(startBtn,SIGNAL(clicked()),this,SLOT(slotStart()));
    connect(stopBtn,SIGNAL(clicked()),this,SLOT(slotStop()));
    connect(quitBtn,SIGNAL(clicked()),this,SLOT(close()));

 “开始”槽函数的实现：

//threaddlg.cpp

void ThreadDlg::slotStart()
{
    for(int i=0;i<MAXSIZE;i++)
    {
        workThread[i]=new WorkThread();	//(a)
    }
    for(int i=0;i<MAXSIZE;i++)
    {
        workThread[i]->start();			//(b)
    }
    startBtn->setEnabled(false);
    stopBtn->setEnabled(true);
}

（a）创建指定数目的线程

（b）调用 QThread 基类的 start() 函数，此函数将启动 run() 函数。从此线程开始运行。

void ThreadDlg::slotStop()
{
    for(int i=0;i<MAXSIZE;i++)
    {
        workThread[i]->terminate();
        workThread[i]->wait();
    }
    startBtn->setEnabled(true);
    stopBtn->setEnabled(false);
}

“停止”槽函数实现：

void ThreadDlg::slotStop()
{
    for(int i=0;i<MAXSIZE;i++)
    {
        workThread[i]->terminate();    //a
        workThread[i]->wait();         //b
    }
    startBtn->setEnabled(true);
    stopBtn->setEnabled(false);
}

（a）调用 QThread 基类的 terminate() 函数，依次终止数组中的 WorkThread 类实例。

（b）调用 QThread 基类的 wait() 函数，使得线程阻塞等待直到退出或超时。

4. 测试

//threaddlg.h

#define MAXSIZE 1

线程数为1的时候：

但是改成线程数为5的时候，依然是这个结果。。。

2. 多线程控制

实现线程的互斥与同步常用类有：

QMutex、QMutexLocker、QReadWriteLocker、QReadLocker、QWriteLocker、QSemaphore 和 QWaitCondition。

class Key
{
public:
    key()   {key=0;}
    int createKey() {++key; return key;}
    int value()const {return key;}
private:
    int key;
}

上面代码中，每创建一个Key类，key的值就会递增，也就是不重复的Key类。

但是在多线程环境下，这个类是不安全的，因为存在多个线程同时修改成员变量 key 的值。

在 C++ 中，“++”操作符不是原子操作。编译后，展开为以下3条命令

1）将变量值载入寄存器

2）将寄存器中的值加1

3）将寄存器中的值写回主存

设当前 key=0，如果 pth1 和 pth2 同时将0载入到寄存器，执行加1操作后写回主存。此时key=1，只是进行了一次加1操作。

2.1 QMutex 类

class Key
{
public:
    key()   {key=0;}
    int createKey() { mutex.lock(); ++key; return key; mutex.unlock();}
    int value()const { mutex.lock(); return key; mutex.unlock();}
private:
    int key;
    QMutex mutex;
}

在自加的命令进行互斥锁的加解锁。QMutex 类具有 lock()、unlock() 以及 tryLock() 函数等。

但是上面的代码，是有问题的。unlock() 操作不得已在 return 之后（两个函数都有 return key的操作），这样会导致 unlock() 操作永远无法执行。

2.2 QMutexLocker 类

简化了互斥量的处理。

class Key
{
public:
    key()   {key=0;}
    int createKey() { QmutexLocker locker(&mutex); ++key; return key;}
    int value()const { QmutexLocker locker(&mutex); return key; mutex.unlock();}
private:
    int key;
    QMutex mutex;
}

在构造函数中接收一个 QMutex 对象作为参数并将其锁定，在析构函数中解锁这个互斥量，这样就不会有之前的问题。

2.3 信号量

又要碰到生产者——消费者模型了。这边是利用信号量的方式：

//main.cpp

#include <QCoreApplication>
#include <QSemaphore>
#include <QThread>
#include <stdio.h>

const int DataSize=1000;
const int BufferSize=80;
int buffer[BufferSize];    //a
QSemaphore freeBytes(BufferSize);    //b
QSemaphore usedBytes(0);    //c

（a）生产者向 buffer 写入数据，直到最大。然后从起点重新写入，也就覆盖已经存在的数据。消费者读取 buffer 的数据，每个 int 字长都被看成一个资源，当然会更改单位。

（b） freeBytes 信号量控制可被生产者填充的缓冲区部分。

（c）usedBytes 信号量控制可被消费者读取的缓冲区部分。

 Producer 类继承自 QThread 类，作为生产者。

//main.cpp

class Producer : public QThread
{
public:
    Producer();
    void run();
};

生产者的构造函数，以及 run() 函数：

Producer::Producer()
{
}

void Producer::run()
{
    for(int i=0;i<DataSize;i++)
    {
       freeBytes.acquire();    //a
       buffer[i%BufferSize]=(i%BufferSize);    //b
       usedBytes.release();    //c
    }
}

（a）生产者线程首先获取一个空闲单元，如果消费者占用整个缓冲区，那么该函数的调用会被阻塞，直到消费者读取数据为止。当然可以用 tryAcquire(n) 函数。

（b）一旦生产者获取了空闲单元，就使用氮气的缓冲区单元序号填充这个缓冲区单元。

（c）调用该函数，将可用资源+1.也就是生产者生产了数据，且消费者可以读取。

消费者就不赘述了：

class Consumer : public QThread
{
public:
    Consumer();
    void run();
};

Consumer::Consumer()
{
}

void Consumer::run()
{
    for(int i=0;i<DataSize;i++)
    {
        usedBytes.acquire();

        fprintf(stderr,"%d",buffer[i%BufferSize]);
        if(i%16==0&&i!=0)
            fprintf(stderr,"\n");

        freeBytes.release();
    }
    fprintf(stderr,"\n");
}

int main(int argc, char *argv[])
{
    QCoreApplication a(argc, argv);

    Producer producer;
    Consumer consumer;

    producer.start();
    consumer.start();

    producer.wait();
    consumer.wait();

    return a.exec();
}

2.4 等待与唤醒

生产者——消费者还有一个方式：使用 QWaitCondition 类。其实有点像理发师模型？

QWaitCondition bufferEmpty;
QWaitCondition bufferFull;
QMutex mutex;								//(a)
int numUsedBytes=0;							//(b)
int rIndex=0;								//(c)

 （a）使用互斥量保证对线程操作的原子性

（b）numUsedBytes 表示存在多少“可用字节”

（c）rIndex 用于表示当前所读取缓冲区的位置

Producer::Producer()
{
}
void Producer::run()
{
    for(int i=0;i<DataSize;i++)				//(a)
    {
       mutex.lock();
       if(numUsedBytes==BufferSize)			//(b)
          bufferEmpty.wait(&mutex);			//(c)
       buffer[i%BufferSize]=numUsedBytes;	//(d)
       ++numUsedBytes;						//增加numUsedBytes变量
       bufferFull.wakeAll();				//(e)
       mutex.unlock();
    }
}

（a）for 循环中的所有语句都要进行保护

（b）检查缓冲区是否被填满

（c）如果被填满了，则等待 缓冲区有位置（等待消费者消费）。

（d）如果没有被填满，则想缓冲区写入一个数值。

（e）最后唤醒等待，也就是跟别人说有数据可以获取

消费者就不赘述了：

void Consumer::run()
{
    forever
    {
        mutex.lock();
        if(numUsedBytes==0)
            bufferFull.wait(&mutex);			
        printf("%ul::[%d]=%d\n",currentThreadId(),rIndex,buffer[rIndex]);
                                                
        rIndex=(++rIndex)%BufferSize;			
        --numUsedBytes;							
        bufferEmpty.wakeAll();					
        mutex.unlock();
    }
    printf("\n");
}