Producer-Consumer на QWaitCondition: классический паттерн в Qt-коде

Введение

Паттерн «производитель-потребитель» — один из самых распространённых в многопоточном программировании: один или несколько потоков генерируют данные (производители) и помещают их в общую очередь, а другой поток (или потоки) забирают данные из очереди и обрабатывают (потребители). Naive-реализация с постоянным опросом («активное ожидание», busy-waiting) расходует процессорное время впустую. В Qt для эффективного ожидания есть QWaitCondition — примитив, позволяющий потоку «уснуть» до явного сигнала о появлении новых данных. В статье построим классическую реализацию этого паттерна.

Концепция

QWaitCondition всегда используется в паре с QMutex: поток-потребитель блокирует мьютекс, проверяет условие (есть ли данные в очереди), и если условия нет — вызывает wait(&mutex), который атомарно освобождает мьютекс и засыпает поток до пробуждения. Производитель, добавив данные, блокирует тот же мьютекс и вызывает wakeOne() (разбудить одного ожидающего) или wakeAll() (разбудить всех). После пробуждения wait() снова автоматически захватывает мьютекс перед возвратом — поэтому код после wait() снова работает под защитой блокировки.

Обязательное правило: проверка условия должна происходить в цикле while, а не в одиночном if, потому что между пробуждением потока и повторным захватом мьютекса другой поток может успеть забрать данные первым (так называемые «ложные пробуждения» и конкуренция нескольких потребителей).

Пример кода

#include <QCoreApplication>
#include <QThread>
#include <QMutex>
#include <QWaitCondition>
#include <QQueue>
#include <QDebug>

class BoundedQueue
{
public:
    explicit BoundedQueue(int capacity) : m_capacity(capacity) {}

    void produce(int item)
    {
        QMutexLocker locker(&m_mutex);
        while (m_queue.size() >= m_capacity) {
            m_notFull.wait(&m_mutex); // ждём, пока появится место
        }
        m_queue.enqueue(item);
        qDebug() << "Произведено:" << item << "(в очереди:" << m_queue.size() << ")";
        m_notEmpty.wakeOne();
    }

    int consume()
    {
        QMutexLocker locker(&m_mutex);
        while (m_queue.isEmpty()) {
            m_notEmpty.wait(&m_mutex); // ждём, пока появятся данные
        }
        int item = m_queue.dequeue();
        m_notFull.wakeOne();
        return item;
    }

private:
    QMutex m_mutex;
    QWaitCondition m_notEmpty;
    QWaitCondition m_notFull;
    QQueue<int> m_queue;
    int m_capacity;
};

int main(int argc, char *argv[])
{
    QCoreApplication app(argc, argv);
    BoundedQueue queue(5); // ограниченный буфер на 5 элементов

    auto *producer = QThread::create([&queue]() {
        for (int i = 0; i < 15; ++i) {
            queue.produce(i);
            QThread::msleep(50);
        }
    });

    auto *consumer = QThread::create([&queue]() {
        for (int i = 0; i < 15; ++i) {
            int item = queue.consume();
            qDebug() << "Потреблено:" << item;
            QThread::msleep(150); // потребитель медленнее производителя
        }
    });

    producer->start();
    consumer->start();
    producer->wait();
    consumer->wait();
    delete producer;
    delete consumer;

    return 0;
}

Пояснения к коду

BoundedQueue реализует ограниченную по размеру очередь с двумя условиями ожидания: m_notEmpty (потребитель ждёт появления данных) и m_notFull (производитель ждёт освобождения места, если буфер заполнен до m_capacity). Оба метода используют цикл while для проверки условия — это защищает от ситуации, когда поток был разбужен, но к моменту повторного захвата мьютекса условие уже изменилось из-за другого потока.

Поскольку в примере производитель работает быстрее потребителя (50 мс против 150 мс на операцию), буфер регулярно заполняется до предела, и производитель будет видимо «притормаживать», ожидая на m_notFull.wait() — это наглядно демонстрирует механизм обратного давления (backpressure) ограниченной очереди.

Подводные камни

Использование if вместо while для проверки условия после wait() — частая ошибка, которая работает «почти всегда» в тестах с одним потребителем, но ломается при добавлении второго потребителя или при ложных пробуждениях, специфичных для ОС.
wakeOne() против wakeAll(). Если несколько потоков ждут на одном QWaitCondition, но логически должен быть разбужен только один (например, чтобы забрать ровно один элемент), wakeAll() разбудит всех, и только один из них реально получит данные — остальные снова заснут, что не является ошибкой, но создаёт лишние переключения контекста; для производительности в таких случаях предпочтительнее wakeOne().
Использование разных мьютексов с одним QWaitCondition в разных вызовах wait() — недопустимо и приводит к неопределённому поведению; один QWaitCondition должен быть всегда связан с одним конкретным мьютексом на протяжении всего его использования.
Очередь без ограничения размера (m_capacity отсутствует) при медленном потребителе и быстром производителе может неограниченно расти, потребляя всю доступную память — ограниченный буфер с обратным давлением, как в примере, — стандартное решение этой проблемы.