Черги та стеки «Виробник-Споживач»

1. Вступ

Почнемо з основ. Черга — це фундаментальна структура даних, що працює за принципом FIFO (First-In, First-Out), тобто «перший прийшов — перший пішов». Уявіть звичайну чергу в супермаркеті: хто став першим, того першого й обслуговують.

У багатопоточному програмуванні патерн Виробник-Споживач (Producer-Consumer) — один із найпоширеніших і найпотужніших.

• Виробники (Producers) — це потоки або частини застосунку, які створюють дані чи завдання та кладуть їх до спільної черги. Вони «виробляють» роботу.
• Споживачі (Consumers) — це потоки або частини застосунку, які беруть дані чи завдання з черги та обробляють їх. Вони «споживають» роботу.

Цей патерн допомагає керувати потоком даних, роз’єднує компоненти системи (виробнику не потрібно знати, хто й як оброблятиме дані), робить систему швидшою у відгуку та дає змогу рівномірно розподіляти навантаження між потоками.

Приклад: ConcurrentQueue — додавання та вилучення

Подивімося, як додавати й вилучати елементи з ConcurrentQueue<T>.

using System.Collections.Concurrent;

ConcurrentQueue<string> tasks = new ConcurrentQueue<string>();

// Додавання елементів (виробник)
tasks.Enqueue("Завантажити файл");
tasks.Enqueue("Обробити зображення");
Console.WriteLine($"Завдань у черзі: {tasks.Count}"); // Вивід: Завдань у черзі: 2

// Вилучення елементів (споживач)
if (tasks.TryDequeue(out string task1))
{
    Console.WriteLine($"Виконано завдання: {task1}"); // Вивід: Виконано завдання: Завантажити файл
}

if (tasks.TryDequeue(out string task2))
{
    Console.WriteLine($"Виконано завдання: {task2}"); // Вивід: Виконано завдання: Обробити зображення
}

if (!tasks.TryDequeue(out string emptyTask))
{
    Console.WriteLine("Черга порожня, нових завдань немає."); // Вивід: Черга порожня, нових завдань немає.
}

Основи роботи: Enqueue(), TryDequeue()

Enqueue(T item): використовується для додавання елемента в кінець черги. Ця операція потокобезпечна. Ви можете одночасно викликати Enqueue з 10 різних потоків — усі елементи коректно додадуться.

TryDequeue(out T item): використовується для спроби вилучення елемента з початку черги. Це ключовий метод для споживачів. Він повертає true, якщо елемент було успішно вилучено (значення потрапляє у вихідний параметр item), і false, якщо черга порожня. Важливо, що TryDequeue не блокує потік, якщо черга порожня.

2. Важливість TryDequeue() та атомарність операцій

Метод TryDequeue() не просто зручний — він критично важливий для коректної потокобезпечної роботи. Він атомарний: перевірка на порожність черги та саме вилучення елемента відбуваються як одна неподільна операція.

Якби у нас були окремі методи IsEmpty (перевірити, чи порожня черга) і Dequeue (вилучити елемент), то між їхніми викликами інший потік міг би спорожнити чергу. У підсумку ваш Dequeue викинув би виняток або повернув некоректні дані. TryDequeue повністю прибирає таку ситуацію.

Приклад: Producer-Consumer з кількома потоками

Тут запускаємо два потоки-виробники й один потік-споживач.

using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Tasks;

ConcurrentQueue<int> dataQueue = new ConcurrentQueue<int>();
bool producersDone = false; // Прапорець для сигналізації споживачу

void Producer(int start, int count)
{
    for (int i = 0; i < count; i++)
    {
        dataQueue.Enqueue(start + i);
        Console.WriteLine($"[В] Додав: {start + i}");
        Thread.Sleep(10); 
    }
}

void Consumer()
{
    while (!producersDone || dataQueue.Count > 0) // Продовжуємо, доки є дані або виробники працюють
    {
        if (dataQueue.TryDequeue(out int item))
        {
            Console.WriteLine($"[С] Обробив: {item}");
        }
        else
        {
            Thread.Sleep(50); // Чекаємо, якщо черга порожня
        }
    }
    Console.WriteLine("[С] Завершив роботу.");
}

// Запуск прикладу в Main:
// Task.Run(() => Producer(1, 5));
// Task.Run(() => Producer(100, 5)); // Другий виробник
// Task.Run(() => Consumer());
// Thread.Sleep(600); // Даємо потокам трохи часу попрацювати
// producersDone = true; // Сигналізуємо, що виробники закінчили
// Thread.Sleep(200); // Даємо споживачу час забрати залишок

Зверніть увагу, що в цьому простому прикладі прапорець producersDone і Thread.Sleep використовуються для імітації завершення. У реальних застосунках для надійнішої синхронізації завершення часто застосовують CancellationTokenSource або BlockingCollection<T>.

ConcurrentQueue<T> ідеально підходить для сценаріїв, де:

• Порядок обробки елементів важливий (FIFO).
• Багато потоків додають елементи та/або багато потоків їх забирають.
• Потрібна висока продуктивність без ручного керування блокуваннями.

3. Стек для «Виробник-Споживач» (LIFO)

Стек — це інша фундаментальна структура даних, що працює за принципом LIFO (Last-In, First-Out), тобто «останній прийшов — перший пішов». Уявіть стос тарілок: ви завжди берете найверхню, і нову тарілку завжди кладете на сам верх стосу.

ConcurrentStack<T> так само потокобезпечний, як і ConcurrentQueue<T>, і теж може використовуватися в патерні «виробник-споживач», але з протилежним порядком обробки.

Приклад: ConcurrentStack — додавання та вилучення

using System.Collections.Concurrent;

ConcurrentStack<string> commandStack = new ConcurrentStack<string>();

// Додавання команд (виробник)
commandStack.Push("Виділити текст");
commandStack.Push("Змінити шрифт");
commandStack.Push("Зберегти документ");
Console.WriteLine($"Команд у стеку: {commandStack.Count}"); // Вивід: Команд у стеку: 3

// Вилучення команд (споживач)
if (commandStack.TryPop(out string cmd1))
{
    Console.WriteLine($"Скасовано команду: {cmd1}"); // Вивід: Скасовано команду: Зберегти документ
}

if (commandStack.TryPop(out string cmd2))
{
    Console.WriteLine($"Скасовано команду: {cmd2}"); // Вивід: Скасовано команду: Змінити шрифт
}

if (!commandStack.TryPop(out string emptyCmd))
{
    Console.WriteLine("Стек команд порожній."); // Вивід: Стек команд порожній.
}

4. Основи роботи: Push(), TryPop()

Push(T item): використовується для додавання елемента на вершину стеку. Операція потокобезпечна.

TryPop(out T item): використовується для спроби вилучення елемента з вершини стеку. Повертає true, якщо елемент було успішно вилучено, і false, якщо стек порожній. Як і TryDequeue, це атомарна операція, що запобігає станам гонки.

Приклад: використання ConcurrentStack для пулу об’єктів

Стек чудово підходить для реалізації пулів об’єктів: взяв — використав — повернув.

using System.Collections.Concurrent;

class Connection { /* Проста заглушка */ public Guid Id { get; } = Guid.NewGuid(); }

ConcurrentStack<Connection> connectionPool = new ConcurrentStack<Connection>();

// Заповнюємо пул початковими зʼєднаннями
for (int i = 0; i < 3; i++)
{
    connectionPool.Push(new Connection());
}
Console.WriteLine($"Зʼєднань у пулі: {connectionPool.Count}"); // Вивід: Зʼєднань у пулі: 3

void UseConnection()
{
    if (connectionPool.TryPop(out Connection conn))
    {
        Console.WriteLine($"[Пул] Використано зʼєднання: {conn.Id}");
        // Імітація роботи зі зʼєднанням
        Thread.Sleep(50); 
        connectionPool.Push(conn); // Повертаємо у пул
        Console.WriteLine($"[Пул] Повернено зʼєднання: {conn.Id}. У пулі: {connectionPool.Count}");
    }
    else
    {
        Console.WriteLine("[Пул] Пул зʼєднань порожній. Створюємо нове.");
        // Зазвичай тут створюють нове зʼєднання, якщо пул порожній
        connectionPool.Push(new Connection()); 
    }
}

// Запуск прикладу в Main:
Task.Run(() => UseConnection());
Task.Run(() => UseConnection());
Task.Run(() => UseConnection());
Thread.Sleep(500);

У цьому прикладі кілька потоків можуть безпечно брати й повертати з’єднання у спільний пул.

5. Приклади застосування та порівняння з ConcurrentQueue

ConcurrentStack<T> застосовується, коли:

• Порядок LIFO критичний (наприклад, історія операцій для функції «Скасувати»).
• Потрібен швидкий доступ до найостанніших доданих елементів (часто лишаються «гарячими» в кеші процесора).
• Реалізуються алгоритми на стеках (обхід графів у глибину, синтаксичний аналіз виразів).

Порівняння та вибір відповідної колекції

ConcurrentQueue

ConcurrentStack

Вибір між ConcurrentQueue і ConcurrentStack повністю залежить від потрібного порядку обробки елементів у вашому сценарії «Виробник-Споживач». Обидві колекції забезпечують високу продуктивність і потокобезпечність відразу, не вимагають ручної синхронізації та допомагають будувати масштабовані багатопотокові системи.