muduo库MutexLock、MutexLockGuard、Contidion、CountDownLatch分析

MutexLock

MutexLockGuard

Condition

CountDownLatch

MutexLock

互斥量是线程同步常用的变量，但在实际中一般都是使用封装的函数，这样便于操作。其类图如下：




共有两个变量，mutex_是互斥量，holder_是用来表示给互斥量上锁线程的tid。

在构造函数中初始化互斥量mutex_和holder_(0)，在析构函数中给销毁mutex_。对我接口根据名字很容易看出用法。

bool isLockByThisThread()是用来检查是否是当前线程给这个MutexLock对象加锁的，原理为比较holder_和 CurrentThread::tid()的值。

assingnHolder和unassignHolder分别在上锁时给holder_赋值，解锁是给holder_置零。assignHolder在上锁后调用，而unassignHolder在解锁前调用。

pthread_mutex_t* getPthreadMutex()可以返回指向类对象中互斥量的指针，在类外对互斥量操作，这个主要用在条件变量中。

在MutexLock中，还有一个类UnassignGuard，这个类中有一个MutexLock对象的引用，在其构造函数调用unassignHolder，析构函数中assignHolder，这个是为条件变量pthread_cond_wait()调用时设计的。在调用pthread_cond_wait()会解锁MuteLock，等待条件（其他线程会给MutexLock上锁）。

MutexLockGuard

在使用mutex时，有时会忘记给mutex解锁，为了防止这种情况发生，常常使用RAII手法。MutexLockGuard就是为此设计的，源码非常简单：

class MutexLockGuard : boost::noncopyable
{
 public:
  explicit MutexLockGuard(MutexLock& mutex)
    : mutex_(mutex)
  {
    mutex_.lock();
  }

  ~MutexLockGuard()
  {
    mutex_.unlock();
  }

 private:

  MutexLock& mutex_;
};

内部有一个MutexLock变量的引用，在构造函数初始化，并上锁；在析构函数解锁。

Condition

Condition类封装了条件变量，给出了几个接口，其源码很简单：

class Condition : boost::noncopyable
{
 public:
  explicit Condition(MutexLock& mutex)
    : mutex_(mutex)
  {
    MCHECK(pthread_cond_init(&pcond_, NULL));//初始化条件变量
  }

  ~Condition()
  {
    MCHECK(pthread_cond_destroy(&pcond_));//销毁
  }

  void wait()
  {
    MutexLock::UnassignGuard ug(mutex_);//这里先给mutex_的holder_置零。在等其析构时会给holder_赋值
    MCHECK(pthread_cond_wait(&pcond_, mutex_.getPthreadMutex()));//给mutex_解锁
  }

  // returns true if time out, false otherwise.
  bool waitForSeconds(int seconds);

  void notify()
  {
    MCHECK(pthread_cond_signal(&pcond_));
  }

  void notifyAll()
  {
    MCHECK(pthread_cond_broadcast(&pcond_));
  }

 private:
  MutexLock& mutex_;
  pthread_cond_t pcond_;
};

在这里可以看到MutexLock::UnassignGuard的应用。

pthread_cond_wait(&pcond_, mutex_.getPthreadMutex())会先给mutex_解锁，然后等待条件。这两步是原子操作。在条件成立后，它会给mutex_加锁，然后返回，这两步也是原子操作。

在学了上面三个类后，可以使用封装的库来写生产者消费者模型了。生产者消费者模型可以参考这里。

#include <muduo/base/Mutex.h>
#include <muduo/base/Thread.h>
#include <muduo/base/Condition.h>
#include <boost/bind.hpp>

#include <stdio.h>  

using namespace  muduo;

struct msg{  
    struct msg *next;  
    int num;  
};  

struct msg *head;  

MutexLock mutex;
Condition con(mutex);

void consumer()//消费者  
{  
    struct msg *mp;  
    for(;;)
    { 
        {
            MutexLockGuard lock(mutex);
            while(head==NULL)//无货的话，等待生产者生产  
                con.wait();
            mp=head;  
            head=mp->next;   
        }

        printf("Consume %d\n",mp->num);  
        free(mp);  
        sleep(rand()%5);  
    }  
}  
void producer()//生产者  
{  
    struct msg *mp;  
    for(;;){  
        mp=static_cast<msg*> (malloc(sizeof(struct msg)));  
        mp->num=rand()%1000+1;  
        printf("Produce %d\n",mp->num);  
        {
            MutexLockGuard lock(mutex);
            mp->next=head;  
            head=mp;  
        }
        con.notify();
        sleep(rand()%5);  
    }  
}  
int main(int argc, char **argv)  
{  

    Thread t1(boost::bind(producer), "Producer");
    Thread t2(boost::bind(consumer), "Consumer");
    t1.start();
    t2.start();
    t1.join();
    t2.join();
return 0;  
}

CountDownLatch

这个是参考Java的一个同步辅助类，在完成一组正在其他线程中执行的操作之前，它允许一个或多个线程一直等待。

例如一组线程等待一个命令，让命令到来时，这些线程才开始运行。或者一个线程等待多个线程运行结束后才可以运行。

其源码很简单：

class CountDownLatch : boost::noncopyable
{
 public:

  explicit CountDownLatch(int count);

  void wait();

  void countDown();

  int getCount() const;

 private:
  mutable MutexLock mutex_;
  Condition condition_;
  int count_;
};

先看私有变量，count_为一个数值，当这个数值为零时，才会通知阻塞在调用wait()的线程。函数void countDown()每次调用都会给count_减1。

以一个例子说明，主线程创建了2个子线程，这两个子线程调用的函数阻塞在wait()上。当主线程调用countDown()后，子线程才运行。

#include <muduo/base/CountDownLatch.h>
#include <muduo/base/Thread.h>
#include <boost/bind.hpp>

#include <vector>
#include <stdio.h>
using namespace muduo;
CountDownLatch latch_(1);
void Function()
{
    latch_.wait();//wait for latch_ countDown
    printf("Thread ID=%d, Name=%s\n", CurrentThread::tid(), CurrentThread::name());
}
int main()
{
    Thread t1(boost::bind(Function), "Thread 1");
    Thread t2(boost::bind(Function), "Thread 2");
    t1.start();
    t2.start();

    printf("main thread running, before countDown\n");
    latch_.countDown();
    sleep(3);//wait for thread t1, t2
    printf("main thread running, after countDown\n");

    t1.join();
    t2.join();

    return 0;
}

运行结果：

main thread running, before countDown

Thread ID=6994, Name=Thread 2

Thread ID=6993, Name=Thread 1

main thread running, after countDown