QT多线程编程

Qt线程类



Qt 包含下面一些线程相关的类：

QThread 提供了开始一个新线程的方法

QThreadStorage 提供逐线程数据存储

QMutex  提供相互排斥的锁，或互斥量

QMutexLocker 是一个便利类，它可以自动对QMutex加锁与解锁

QReadWriterLock 提供了一个可以同时读操作的锁

QReadLocker与QWriteLocker 是便利类，它自动对QReadWriteLock加锁与解锁

QSemaphore 提供了一个整型信号量，是互斥量的泛化

QWaitCondition 提供了一种方法，使得线程可以在被另外线程唤醒之前一直休眠。



Qt线程的创建

Qt线程中有一个公共的抽象类，所有的线程都是从这个QThread抽象类中派生的，要实现QThread中的纯虚函数run(),run()函数是通过start()函数来实现调用的。



class MyThread : public QThread {

      public:

virtual void run();

};



void MyThread::run()

{

for( int count = 0; count < 20; count++ ) {

sleep( 1 );

qDebug( "Ping!" );

}

}



int main()

{

MyThread a;

MyThread b;



a.start();//自动调用run(),否则即使该线程创建，也是一开始就挂起

b.start();

//要等待线程a,b都退出

a.wait();

b.wait();

}



Qt线程同步



1．QMutex

QMutex ( bool recursive = FALSE )

virtual ~QMutex ()

void lock ()  //试图锁定互斥量。如果另一个线程已经锁定这个互斥量，那么这次调用将阻塞直到那个线程把它解锁。

void unlock ()

bool locked ()

bool tryLock () //如果另一个进程已经锁定了这个互斥量，这个函数返回假，而不是一直等到这个锁可用为止，比如，它不是阻塞的。

QMutex mutex;

      void someMethod()

      {

mutex.lock();

qDebug("Hello");

qDebug("World");

              mutex.unlock();

       }

  

       //用Java的术语，这段代码应该是：

       void someMethod()

       {

              synchronized {

                     qDebug("Hello");

                     qDebug("World");

              }

       }

    不过在Qt中我们可用通过另一个类来简化这种应用，因为如果使用QMutex.lock()而没有对应的使用QMutex.unlcok()的话就会造成死锁，别的线程永远也得不到接触该mutex锁住的共享资源的机会。尽管可以不使用lock()而使用tryLock(timeout)来避免因为死等而造成的死锁( tryLock(负值)==lock()),但是还是很有可能造成错误。

对于上述的情况MFC中用CSingleLock 或 MultiLock，Boost中用boost::mutex::scoped_lock来进行解决，而在Qt中用QMutexLocker来进行解决。下面是没有采用 QMutexLocker的例子和采用QMutexLocker的方案。



2．QMutexLocker

this complex function locks a QMutex upon entering the function and unlocks the mutex at all the exit points



       int complexFunction(int flag)

       {

              mutex.lock();

              int retVal = 0;

             switch (flag) {

             case 0:

              case 1:

                     mutex.unlock();

                     return moreComplexFunction(flag);

              case 2:

                     {

                            int status = anotherFunction();

                            if (status < 0) {

                                   mutex.unlock();

                                   return -2;

                            }

                            retVal = status + flag;

                     }

                     break;

              default:

                     if (flag > 10) {

                            mutex.unlock();

                            return -1;

                     }

                     break;

              }

              mutex.unlock();

              return retVal;

       }



    This example increases the likelihood that errors will occur.Using QMutexLocker greatly simplifies the code, and makes it more readable:

       int complexFunction(int flag)

       {

              QMutexLocker locker(&mutex);

              int retVal = 0;

              switch (flag) {

              case 0:

              case 1:

                     return moreComplexFunction(flag);

              case 2:

                     {

                            int status = anotherFunction();

                            if (status < 0)

                                   return -2;

                            retVal = status + flag;

                     }

                     break;

              default:

                     if (flag > 10)

                            return -1;

                     break;

              }

              return retVal;

       }



    Now, the mutex will always be unlocked when the QMutexLocker object is destroyed (when the function returns since locker is an auto variable).即使在抛出异常的情况下也可以使用。



3．QReadWriteLock

    用mutex进行线程同步有一个问题就是mutex只允许某个时刻只允许一个线程对共享资源进行访问，如果同时有多个线程对共享资源进行读访问，而只有一个写操作线程，那么在这种情况下如果采用mutex就成为程序运行性能的瓶颈了。在这种情况下Qt下采用QReadWriteLock来实现多个线程读，一个线程写。写线程执行的时候会阻塞所有的读线程，而读线程之间的运行不需要进行同步。



       MyData data;

       QReadWriteLock lock;

       void ReaderThread::run()

       {

              lock.lockForRead();

              access_data_without_modifying_it(&data);

              lock.unlock();

       }

       void WriterThread::run()

       {

              lock.lockForWrite();

              modify_data(&data);

              lock.unlock();

       }

    QReadWriterLock 与QMutex相似，除了它对 "read","write"访问进行区别对待。它使得多个读者可以共时访问数据。使用QReadWriteLock而不是QMutex，可以使得多线程程序更具有并发性。



4．QReadLocker和QWriteLocker

    对于QMutex有QMutexLocker来简化使用，而对于QReadWriteLock有 QReadLocker和QWriteLocker。



    Here's an example that uses QReadLocker to lock and unlock a read-write lock for reading:  



       QReadWriteLock lock;

       QByteArray readData()

       {

              QReadLocker locker(&lock);

              return data;

       }



     It is equivalent to the following code:



       QReadWriteLock lock;

       QByteArray readData()

       {

              lock.lockForRead();

              lock.unlock();

              return data;

       }



5．QSemaphore

    QSemaphore 是QMutex的一般化，它可以保护一定数量的相同资源，与此相对，一个mutex只保护一个资源。下面例子中，使用QSemaphore来控制对环状缓冲区的访问，此缓冲区被生产者线程和消费者线程共享。生产者不断向缓冲写入数据直到缓冲末端，消费者从缓冲不断从缓冲头部读取数据。

    信号量比互斥量有更好的并发性，假如我们用互斥量来控制对缓冲的访问，那么生产者，消费者不能同时访问缓冲。然而，我们知道在同一时刻，不同线程访问缓冲的不同部分并没有什么危害。



       QSemaphore semaphore(1);   |      QMutex mutex;

       Qsemaphore.acquire();          |      Qmutex.lock();

       Qsemaphore.release();           |      Qmutex.unlock();



       Public Functions



       QSemaphore ( int n = 0 )

       ~QSemaphore ()

       void acquire ( int n = 1 )

       int available () const

       void release ( int n = 1 )

       bool tryAcquire ( int n = 1 )

       bool tryAcquire ( int n, int timeout )



    Semaphores support two fundamental operations, acquire() and release():

    acquire(n) tries to acquire n resources. If there aren't that many resources available, the call will block until this is the case.

    release(n) releases n resources.

    tryAcquire() returns immediately if it cannot acquire the resources

    available() returns the number of available resources at any time.



       Example:



       QSemaphore sem(5);      // sem.available() == 5

       sem.acquire(3);         // sem.available() == 2

       sem.acquire(2);         // sem.available() == 0

       sem.release(5);         // sem.available() == 5

       sem.release(5);         // sem.available() == 10

       sem.tryAcquire(1);      // sem.available() == 9, returns true

       sem.tryAcquire(250);    // sem.available() == 9, returns false



    生产者线程写数据到buffer直到缓冲末端，然后重新从buffer的头部开始写。显然producer线程和consumer线程是需要进行同步的，If the producer generates the data too fast, it will overwrite data that the consumer hasn't yet read; if the consumer reads the data too fast, it will pass the producer and read garbage.

    A crude way to solve this problem is to have the producer fill the buffer, then wait until the consumer has read the entire buffer, and so on. 显然这样做效率是比较低的。



       const int DataSize = 100000;

       const int BufferSize = 8192;

       char buffer[BufferSize];

       //When the application starts, the reader thread will start  

       //acquiring "free" bytes and convert them into "used" bytes

       QSemaphore freeBytes(BufferSize); //producer线程在此区域写入数据，初始资源数量为BufferSize

       QSemaphore usedBytes; //consumer线程读取此区域的数据,初始资源数量为0

       //For this example, each byte counts as one resource.

       //In a real-world application, we would probably operate on larger

       //units (for example, 64 or 256 bytes at a time)

      

       class Producer : public QThread

       {

       public:

              void run();

       };



       //生产者每acquire一次就，使用掉Buffer个资源中的一个，而写入的字符存入到buffer数组中

       //从而消费者可用读取字符，从而消费者获取一个资源



       void Producer::run()

       {

              //qsrand(QTime(0,0,0).secsTo(QTime::currentTime()));

              for (int i = 0; i < DataSize; ++i) {

                     freeBytes.acquire();

                     buffer[i % BufferSize] = "ACGT"[(int)qrand() % 4];

                     usedBytes.release();

              }

       }

      

       class Consumer : public QThread

       {

       public:

              void run();

       };



       void Consumer::run()

       {

              for (int i = 0; i < DataSize; ++i) {

                     usedBytes.acquire();

                     fprintf(stderr, "%c", buffer[i % BufferSize]);

                     freeBytes.release();

              }

              fprintf(stderr, "\n");

       }



       //Finally, in main(), we start the producer and consumer threads.

       //What happens then is that the producer converts some "free" space

       //into "used" space, and the consumer can then convert it back to //"free" space.



       int main(int argc, char *argv[])

       {

              QCoreApplication app(argc, argv);

              Producer producer;

              Consumer consumer;

              producer.start();

              consumer.start();

              producer.wait();

              consumer.wait();

              return 0;

       }

   

producer的run函数：

    当producer线程执行run函数，如果buffer中已经满了，而没有consumer线程没有读，这样producer就不能再往buffer中写字符。此时在 freeBytes.acquire处就阻塞直到consumer线程读(consume）数据。一旦producer获取到一个字节(资源）就写如一个随机的字符，并调用 usedBytes.release从而consumer线程获取一个资源可以读一个字节的数据了。



consumer的run函数：

    当consumer线程执行run函数，如果buffer中没有数据，就是资源=0，则consumer线程在此处阻塞。直到producer线程执行写操作，写入一个字节，并执行usedBytes.release从而使得consumer线程的可用资源数=1。则consumer线程从阻塞状态中退出，并将 usedBytes资源数-1，当前资源数=0。



6．QWaitCondition

       QWaitCondition ()

       virtual ~QWaitCondition ()

       bool wait ( QMutex * mutex, unsigned long time = ULONG_MAX )

       void wakeOne ()

       void wakeAll ()

       Public function:

       bool QWaitCondition::wait ( QMutex * mutex, unsigned long time = ULONG_MAX )

       1) 释放锁定的mutex

       2) 在线程对象上等待

    mutex必须由调用线程进行初锁定 。注意调用wait的话，会自动调用unlock解锁之前锁住的资源，不然会造成死锁。线程1等待线程2来改变共享资源，从而达到一定的条件然后发出信号，使得线程1从wait中的阻塞状态中被唤醒。但是线程2想改变资源，却无法办到，因为线程1调用lock之后就在wait中blocking，了但是没有及时的unlock，那么这就构成了死锁的条件。所以说wait函数除了使调用线程切换到内核态之外，还自动unlock(&mutex)mutex将被解锁，并且调用线程将会阻塞，直到下列条件之一满足时才醒来：另一个线程使用wakeOne()或wakeAll()传输信号给它。在这种情况下，这个函数将返回真。

time毫秒过去了。如果time为ULONG_MAX（默认值），那么这个等待将永远不会超时（这个事件必须被传输）。如果等待的事件超时，这个函数将会返回假互斥量将以同样的锁定状态返回。这个函数提供的是允许从锁定状态到等待状态的原子转换。

       void QWaitCondition::wakeAll ()

    这将会唤醒所有等待QWaitCondition的线程。这些线程被唤醒的顺序依赖于操组系统的调度策略，并且不能被控制或预知。

       void  QWaitCondition::wakeOne ()

    这将会唤醒所有等待QWaitCondition的线程中的一个线程。这个被唤醒的线程依赖于操组系统的调度策略，并且不能被控制或预知。

    假定每次用户按下一个键，我们有三个任务要同时执行，每个任务都可以放到一个线程中，每个线程的run()都应该是这样：

       QWaitCondition key_pressed;

       for (;;) {

              key_pressed.wait(); // 这是一个QWaitCondition全局变量

              // 键被按下，做一些有趣的事

              do_something();

       }



       或是这样：

       forever {

              mutex.lock();

              keyPressed.wait(&mutex);

              do_something();

              mutex.unlock();

       }



    第四个线程回去读键按下并且每当它接收到一个的时候唤醒其它三个线程，就像这样：

       QWaitCondition key_pressed;

       for (;;) {

              getchar();

              // 在key_pressed中导致引起任何一个线程。wait()将会从这个方法中返回并继续执行

              key_pressed.wakeAll();

       }

    注意这三个线程被唤醒的顺序是未定义的，并且当键被按下时，这些线程中的一个或多个还在do_something()，它们将不会被唤醒（因为它们现在没有等待条件变量）并且这个任务也就不会针对这次按键执行操作。这种情况是可以避免得，比如，就像下面这样做：



       QMutex mymutex;

       QWaitCondition key_pressed;

       int mycount=0;

       //Worker线程代码

       for (;;) {

              key_pressed.wait(); // 这是一个QWaitCondition全局变量

              //keyPressed.wait(&mutex);

              mymutex.lock();

              mycount++;

              mymutex.unlock();

              do_something();

              mymutex.lock();

              mycount--;

              mymutex.unlock();

       }

       // 读取按键线程代码

       for (;;) {

              getchar();

              mymutex.lock();

              // 睡眠，直到没有忙碌的工作线程才醒来。count==0说明没有Worker线程在do something

              while( count > 0 ) {

                     mymutex.unlock();

                     sleep( 1 );

                     mymutex.lock();

              }

              mymutex.unlock();

              key_pressed.wakeAll();

       }

    应用条件变量对前面用信号量进行保护的环状缓冲区的例子进行改进：

下面的例子中：

    1）生产者首先必须检查缓冲是否已满(numUsedBytes==BufferSize)，如果是，线程停下来等待bufferNotFull条件。如果不是，在缓冲中生产数据，增加numUsedBytes,激活条件 bufferNotEmpty。

    2）使用mutex来保护对numUsedBytes的访问。

    另外，QWaitCondition::wait()接收一个mutex作为参数，这个mutex应该被调用线程初始化为锁定状态。在线程进入休眠状态之前，mutex会被解锁。而当线程被唤醒时，mutex会再次处于锁定状态。

    而且，从锁定状态到等待状态的转换是原子操作，这阻止了竞争条件的产生。当程序开始运行时，只有生产者可以工作。消费者被阻塞等待bufferNotEmpty条件，一旦生产者在缓冲中放入一个字节，bufferNotEmpty条件被激发，消费者线程于是被唤醒。



       const int DataSize = 100000;

       const int BufferSize = 8192;

       char buffer[BufferSize];

       QWaitCondition bufferNotEmpty;

       QWaitCondition bufferNotFull;

       QMutex mutex;

       int numUsedBytes = 0;



       class Producer : public QThread

       {

       public:

              void run();

       };

      

       void Producer::run()

       {

              qsrand(QTime(0,0,0).secsTo(QTime::currentTime()));

              for (int i = 0; i < DataSize; ++i) {

                     mutex.lock();

                     //producer线程首先检查缓冲区是否已满

                     if (numUsedBytes == BufferSize)//缓冲区已满，等待consumer来减少numUsedBytes

                     // bufferNotFull.wait(&mutex)先调用mutex.unlock()然后收到信号时调用mutex.lock()

                     bufferNotFull.wait(&mutex);//缓冲区已满等待bufferNotFull的条件变量成立变为有信号

                     mutex.unlock();

                     buffer[i % BufferSize] = "ACGT"[(int)qrand() % 4];

                     mutex.lock();

                     ++numUsedBytes; //producer用掉一个Bytes，表示producer写入buffer中的字节数

                     bufferNotEmpty.wakeAll();

                     mutex.unlock();

              }

       }



       class Consumer : public QThread

       {

       public:

              void run();

       };



       void Consumer::run()

       {

              for (int i = 0; i < DataSize; ++i) {

                     mutex.lock();

                     if (numUsedBytes == 0)

                            bufferNotEmpty.wait(&mutex);

                     mutex.unlock();

                     fprintf(stderr, "%c", buffer[i % BufferSize]);

                     mutex.lock();

                     --numUsedBytes;

                     bufferNotFull.wakeAll();

                     mutex.unlock();

              }

              fprintf(stderr, "\n");

       }



       int main(int argc, char *argv[])

       {

              QCoreApplication app(argc, argv);

              Producer producer;

              Consumer consumer;

              producer.start();

              consumer.start();

              producer.wait();

              consumer.wait();

              return 0;

       }



另外一个例子：



       #include

       #include

       // 全局条件变量

       QWaitCondition mycond;

       // Worker类实现

       class Worker : public QPushButton, public QThread

       {

       Q_OBJECT

       public:

              Worker(QWidget *parent = 0, const char *name = 0): QPushButton(parent, name)

              {

                     setText("Start Working");

                     // 连接从QPushButton继承来的信号和我们的slotClicked()方法

                     connect(this, SIGNAL(clicked()), SLOT(slotClicked()));

                     // 调用从QThread继承来的start()方法……这将立即开始线程的执行

                     QThread::start();

              }

       public slots:

              void slotClicked()

              {

                     // 唤醒等待这个条件变量的一个线程

                     mycond.wakeOne();

              }

       protected:

              void run()

              {

                     // 这个方法将被新创建的线程调用……

                     while ( TRUE ) {

                            // 锁定应用程序互斥锁，并且设置窗口标题来表明我们正在等待开始工作

                            qApp->lock();

                            setCaption( "Waiting" );

                            qApp->unlock();

                            // 等待直到我们被告知可以继续

                            mycond.wait();

                            // 如果我们到了这里，我们已经被另一个线程唤醒……让我们来设置标题来表明我们正在工作

                            qApp->lock();

                            setCaption( "Working!" );

                            qApp->unlock();

                            // 这可能会占用一些时间，几秒、几分钟或者几小时等等，因为这个一个和GUI线程分开的线程，在处理事件时，GUI线程不会停下来……

                            do_complicated_thing();

                     }

              }

       };

      

       // 主线程——所有的GUI事件都由这个线程处理。

       int main( int argc, char **argv )

       {

              QApplication app( argc, argv );

              // 创建一个worker……当我们这样做的时候，这个worker将在一个线程中运行

              Worker firstworker( 0, "worker" );

              app.setMainWidget( &worker );

              worker.show();

              return app.exec();

       }



7．线程安全类



非线程安全类

    这个类不是线程安全的，因为假如多个线程都试图修改数据成员 n,结果未定义。这是因为c++中的++和--操作符不是原子操作。实际上，它们会被扩展为三个机器指令：

       1，把变量值装入寄存器

       2，增加或减少寄存器中的值

       3，把寄存器中的值写回内存

       假如线程A与B同时装载变量的旧值，在寄存器中增值，回写。他们写操作重叠了，导致变量值仅增加了一次。很明显，访问应该串行化：A执行123步骤时不应被打断。

       class Counter

       {

       public:

              Counter() {n=0;}

              void increment() {++n;}

              void decrement() {--n;}

              int value() const {return n;}

       private:

              int n;

       };



线程安全类

       class Counter

       {

       public:

              Counter() { n = 0; }

              void increment() { QMutexLocker locker(&mutex); ++n; }

              void decrement() { QMutexLocker locker(&mutex); --n; }

              int value() const { QMutexLocker locker(&mutex); return n; }

       private:

              mutable QMutex mutex;

              int n;

       };

    QMutexLocker类在构造函数中自动对mutex进行加锁，在析构函数中进行解锁。随便一提的是，mutex使用了mutable关键字来修饰，因为我们在value()函数中对mutex进行加锁与解锁操作，而value()是一个const函数。