Лаб. работа “Организация потоков, параллельной обработки, синхронизации и распределённой обработки синхронизуемых участков кода”

Программирование сетевых приложений

Практика

Автор

Бизюк Андрей

Дата публикации

9 апреля 2026 г.

Цель работы

Реализовать паттерн «Производитель-потребитель» для асинхронной обработки сетевых пакетов. Организовать потокобезопасную очередь с использованием QMutex и QWaitCondition для синхронизации доступа со стороны потоков-производителей и потоков-потребителей.

Теоретические сведения

Потоки в Qt

QThread — класс Qt для создания и управления потоками выполнения. Два основных подхода:

Наследование от QThread — переопределение метода run()
Перемещение QObject в поток — через moveToThread()

class Worker : public QThread {
protected:
    void run() override {
        for (int i = 0; i < 10; i++) {
            // работа
            msleep(100);
        }
    }
};

Синхронизация: QMutex

QMutex — мьютекс для взаимного исключения. Гарантирует, что только один поток одновременно может выполнять защищённый участок кода.

QMutex mutex;

void threadSafeOperation() {
    mutex.lock();
    // критическая секция
    queue.push(data);
    mutex.unlock();
}

// Использование RAII (рекомендуется)
void threadSafeOperation() {
    QMutexLocker locker(&mutex);
    // критическая секция — мьютекс автоматически освобождается
    queue.push(data);
}

Ожидание и уведомление: QWaitCondition

QWaitCondition позволяет потокам ожидать определённого условия и быть разбуженными другими потоками.

QMutex mutex;
QWaitCondition condition;
bool dataReady = false;

// Поток-потребитель
void consumer() {
    mutex.lock();
    while (!dataReady)
        condition.wait(&mutex);  // ожидание, мьютекс освобождается
    // обработка данных
    dataReady = false;
    mutex.unlock();
}

// Поток-производитель
void producer() {
    mutex.lock();
    // добавление данных
    dataReady = true;
    condition.wakeAll();  // уведомление ожидающих потоков
    mutex.unlock();
}

Паттерн «Производитель-потребитель»

Конфигурация CMake для Qt

cmake_minimum_required(VERSION 3.16)
project(Lab5 LANGUAGES CXX)

set(CMAKE_CXX_STANDARD 17)
set(CMAKE_AUTOMOC ON)

find_package(Qt6 REQUIRED COMPONENTS Core)

add_executable(Lab5 main.cpp)
target_link_libraries(Lab5 PRIVATE Qt6::Core)

Задание для выполнения

Реализовать многопоточное приложение, моделирующее обработку сетевых пакетов по паттерну «Производитель-потребитель».

Требования к программе

Создать класс NetworkPacket:
- поля: sourceIp (string), destIp (string), payload (string), timestamp (QString), packetId (int, автоинкрементный)
- метод toString() — строковое представление пакета
Создать класс ThreadSafeQueue:
- закрытые поля: queue<NetworkPacket>, QMutex, QWaitCondition, bool running
- метод push(const NetworkPacket&) — добавить пакет с уведомлением потребителей (wakeOne())
- метод pop() — извлечь пакет (ожидание, если очередь пуста, через wait())
- метод size() — текущий размер очереди (под мьютексом)
- метод stop() — установить running = false и wakeAll() для завершения ожидающих потоков
- метод isEmpty() — проверка пустоты очереди
Создать класс Producer (наследник QThread):
- закрытые поля: ThreadSafeQueue& queue, string ip, int packetsCount, int intervalMs
- метод run() — генерирует packetsCount пакетов с указанным IP-адресом источника, случайным IP-адресом получателя и случайной нагрузкой, добавляет их в очередь с задержкой intervalMs. Выводит в консоль [Producer <ip>] Отправлен пакет #<id>.
Создать класс Consumer (наследник QThread):
- закрытые поля: ThreadSafeQueue& queue, string name, int processedCount
- метод run() — в цикле извлекает пакеты из очереди и «обрабатывает» их (имитация: задержка 50–200 мс, вывод [Consumer <name>] Обработан пакет #<id> от <sourceIp> → <destIp>). Завершается при stop() очереди и её опустошении.
- метод getProcessedCount() — количество обработанных пакетов
В функции main():
- Создать общую очередь ThreadSafeQueue
- Создать 3 потока-производителя (разные IP, по 10 пакетов каждый, интервал 100 мс)
- Создать 2 потока-потребителя
- Запустить все потоки
- Дождаться завершения производителей (wait())
- После завершения производителей вызвать queue.stop()
- Дождаться завершения потребителей
- Вывести статистику: сколько пакетов обработал каждый потребитель

Пример ожидаемого вывода

=== Запуск системы обработки пакетов ===
[Producer 192.168.1.1] Отправлен пакет #1
[Producer 10.0.0.5] Отправлен пакет #2
[Producer 172.16.0.1] Отправлен пакет #3
[Consumer Worker-1] Обработан пакет #1 от 192.168.1.1 → 10.0.0.2
[Consumer Worker-2] Обработан пакет #2 от 10.0.0.5 → 192.168.1.3
[Producer 192.168.1.1] Отправлен пакет #4
...
=== Все производители завершили работу ===
[Consumer Worker-1] Обработан пакет #29 от 172.16.0.1 → 10.0.0.1
[Consumer Worker-2] Обработан пакет #30 от 192.168.1.1 → 172.16.0.2

=== Статистика ===
Сгенерировано пакетов: 30
Worker-1 обработал: 16 пакетов
Worker-2 обработал: 14 пакетов

Индивидуальные задания

Вариант 1. Обработка DNS-запросов

Элемент очереди: DNSQuery (домен, тип записи (A/AAAA/CNAME/MX), IP-клиента, queryId — автоинкрементный). Метод toString(): [DNS] #<id> <domain> <type> от <clientIP>.