Qt 多线程同步与通信

Posted 2020-10-14 云水

Qt 多线程同步与通信

 

1 多线程同步

Qt提供了以下几个类来完成这一点：QMutex、QMutexLocker、QSemphore、QWaitCondition。

当然可能还包含QReadWriteLocker、QReadLocker、QWriteLocker，但线程同步是应用很少，这里只做简单的讲解！

 

QMutex、QMutexLocker

QMutex类提供了一个保护一段临界区代码的方法，他每次只允许一个线程访问这段临界区代码。QMutex::lock()函数用来锁住互斥量，如果互斥量处于解锁状态，当前线程就会立即抓住并锁定它；否则当前线程就会被阻塞，直到持有这个互斥量的线程对其解锁。线程调用lock()函数后就会持有这个互斥量直到调用unlock()操作为止。QMutex还提供了一个tryLock()函数，如果互斥量已被锁定，就立即返回。

在问题出来了，如果要在stop()函数中使用mutex进行互斥操作，但unlock()操作写在那里？unlock()操作却不得不再return之后，从而导致unlock()操作永远也无法执行...

Qt提供了QMutexLocker类何以简化互斥量的处理，它在构造函数中接受一个QMutex对象作为参数并将其锁定，在析构函数中解锁这个互斥量。

 

  bool Thread::stop()  

         {  

             QMutexLocker locker(&mutex);  

             m_stop = true;  

             return m_stop;  

        } 

QReadWriteLocker、QReadLocker、QWriteLocker

下面是一段对QReadWriteLocker类的对象进行，读写锁的操作，比较简单，这里也不多做讲解了，自己看吧

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();  
  ...  
      } 

QSemphore

Qt中的信号量是由QSemaphore类提供的，信号量可以理解为互斥量功能的扩展，互斥量只能锁定一次而信号量可以获取多次，它可以用来保护一定数量的同种资源。

acquire(n)函数用于获取n个资源，当没有足够的资源时调用者将被阻塞直到有足够的可用资源。release(n)函数用于释放n个资源。

QSemaphore类还提供了一个tryAcquire(n)函数，在没有足够的资源是该函数会立即返回。

一个典型的信号量应用程序是在两个线程间传递一定数量的数据(DataSize)，而这两个线程使用一定大小(BufferSize)的共享循环缓存。

 

      const int DataSize = 100000;  

       const int BufferSize = 4096;  

       char buffer[BufferSize]; 

生产者线程向缓存中写入数据，直到它到达终点，然后在起点重新开始，覆盖已经存在的数据。消费者线程读取前者产生的数据。

生产者、消费者实例中对同步的需求有两处，如果生产者过快的产生数据，将会覆盖消费者还没有读取的数据，如果消费者过快的读取数据，将越过生产者并且读取到一些垃圾数据。

解决这个问题的一个有效的方法是使用两个信号量：

QSemaphore freeSpace(BufferSize);  

QSemaphore usedSpace(0); 

freeSpace信号量控制生产者可以填充数据的缓存部分。usedSpace信号量控制消费者可以读取的区域。这两个信号量是互补的。其中freeSpace信号量被初始化为BufferSize(4096)，表示程序一开始有BufferSize个缓冲区单元可被填充，而信号量usedSpace被初始化为0，表示程序一开始缓冲区中没有数据可供读取。

对于这个实例，每个字节就看作一个资源，实际应用中常会在更大的单位上进行操作，从而减小使用信号量带来的开销。

void Producer::run()  
{  
      for (int i = 0; i < DataSize; ++i) {  
         freeSpace.acquire();  
         buffer[i % BufferSize] = "MING"[uint(rand()) % 4];  
         usedSpace.release();  
       }  
 } 

 

在生产者中，我们从获取一个“自由的”字节开始。如果缓存被消费者还没有读取的数据填满，acquire()的调用就会阻塞，直到消费者已经开始消耗这些数据为止。一旦我们已经获取了这个字节，我们就用一些随机数据("M"、"I"、"N"或"G")填充它并且把这个字节释放为“使用的”,所以它可以被消费者线程使用。

void Consumer::run()  
  {  
   for (int i = 0; i < DataSize; ++i) {  
          usedSpace.acquire();  
        cerr << buffer[i % BufferSize];  
         freeSpace.release();  
           }  
        cerr << endl;  
  } 

 

   int main()  
   {  
       Producer producer;  
       Consumer consumer;  
       producer.start();  
       consumer.start();  
       producer.wait();  
       consumer.wait();  
       return 0;  
    } 

QWaitCondition

对生产者和消费者问题的另一个解决方法是使用QWaitCondition,它允许线程在一定条件下唤醒其他线程。其中wakeOne()函数在条件满足时随机唤醒一个等待线程，而wakeAll()函数则在条件满足时唤醒所有等待线程。
下面重写生产者和消费者实例，以QMutex为等待条件，QWaitCondition允许一个线程在一定条件下唤醒其他线程。

 

   const int DataSize = 100000;  

   const int BufferSize = 4096;  

   char buffer[BufferSize];  

   QWaitCondition bufferIsNotFull;  

   QWaitCondition bufferIsNotEmpty;  

   QMutex mutex;  

   int usedSpace = 0;

在缓存之外，我们声明了两个QWaitCondition、一个QMutex和一个存储了在缓存中有多少个“使用的”字节的变量。

   void Producer::run()  
   {  
       for (int i = 0; i < DataSize; ++i) {  
           mutex.lock();  
           if (usedSpace == BufferSize)  
               bufferIsNotFull.wait(&mutex);  
           buffer[i % BufferSize] = "MING"[uint(rand()) % 4];  
           ++usedSpace;  
           bufferIsNotEmpty.wakeAll();  
        mutex.unlock();  
        }  
   }

在生产者中，我们从检查缓存是否充满开始。如果是充满的，我们等待“缓存不是充满的”条件。当这个条件满足时，我们向缓存写入一个字节，增加usedSpace，并且在唤醒任何等待这个“缓存不是空白的”条件变为真的线程。

for循环中的所有语句需要使用互斥量加以保护，以保护其操作的原子性。

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

这个函数做下说明，该函数将互斥量解锁并在此等待，它有两个参数，第一个参数为一个锁定的互斥量，第二个参数为等待时间。如果作为第一个参数的互斥量在调用是不是锁定的或出现递归锁定的情况，wait()函数将立即返回。

调用wait()操作的线程使得作为参数的互斥量在调用前变为锁定状态，然后自身被阻塞变成为等待状态直到满足以下条件：

其他线程调用了wakeOne()或者wakeAll()函数，这种情况下将返回"true"值。

第二个参数time超时(以毫秒记时)，该参数默认情况是ULONG_MAX，表示永不超时，这种情况下将返回"false"值。

wait()函数返回前会将互斥量参数重新设置为锁定状态，从而保证从锁定状态到等待状态的原则性转换。

void Consumer::run()  
 {  
     forever {  
         mutex.lock();  
         if (usedSpace == 0)  
             bufferIsNotEmpty.wait(&mutex);  
         cerr << buffer[i % BufferSize];  
         --usedSpace;  
         bufferIsNotFull.wakeAll();  
      mutex.unlock();  
  }  
  cerr << endl;  
  }

消费者做的和生产者正好相反，他等待“缓存不是空白的”条件并唤醒任何等待“缓存不是充满的”的条件的线程。

main()函数与上面的基本相同，这个不再多说。

在QThread类的静态函数currentThread()，可以返回当前线程的线程ID。在X11环境下，这个ID是一个unsigned long类型的值。

自己总结：

原子类型  QAtomicInt QAtomicPointer 

 

2.多线程通信

  a . 全局变量 即共享内存

  b.  管道

  c.  信号槽

3.多进程通信

  信号槽  socket 文件映射