Növbələr və steklər Producer-Consumer

1. Giriş

Əsaslardan başlayaq. Növbə — FIFO (First-In, First-Out) prinsipi ilə işləyən fundamental data structure-dir, yəni "ilk gələn — ilk çıxır". Supermarketdəki adi növbəni təsəvvür edin: kim əvvəl dayandısa, onu əvvəl xidmət edirlər.

Çoxthreadli proqramlaşdırmada Producer-Consumer patterni ən geniş yayılmış və güclü patternlərdən biridir.

• Producers — datanı və ya taskları yaradan və ümumi növbəyə qoyan threadlər və ya tətbiqin hissələri. Onlar işi "istehsal" edirlər.
• Consumers — növbədən datanı və ya taskları götürən və onları işləyən threadlər və ya tətbiqin hissələri. Onlar işi "istehlak" edirlər.

Bu pattern məlumat axınını idarə etməyə, komponentləri bir-birindən ayırmağa kömək edir (producer kim və necə emal edəcəyini bilməlidir), tətbiqi daha reaktiv edir və threadlər arasında yükü bərabər bölür.

Nümunə: ConcurrentQueue — əlavə etmə və çıxarma

ConcurrentQueue<T>-ə elementlərin necə əlavə və çıxarılmasına baxaq.

using System.Collections.Concurrent;

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

// Əlavə etmə (producer)
tasks.Enqueue("Faylı yüklə");
tasks.Enqueue("Şəkili emal et");
Console.WriteLine($"Növbədəki tapşırıqlar: {tasks.Count}"); // Çıxış: Növbədəki tapşırıqlar: 2

// Çıxarma (consumer)
if (tasks.TryDequeue(out string task1))
{
    Console.WriteLine($"Tamamlanan tapşırıq: {task1}"); // Çıxış: Tamamlanan tapşırıq: Faylı yüklə
}

if (tasks.TryDequeue(out string task2))
{
    Console.WriteLine($"Tamamlanan tapşırıq: {task2}"); // Çıxış: Tamamlanan tapşırıq: Şəkili emal et
}

if (!tasks.TryDequeue(out string emptyTask))
{
    Console.WriteLine("Növbə boşdur, yeni tapşırıq yoxdur."); // Çıxış: Növbə boşdur, yeni tapşırıq yoxdur.
}

Əməliyyatların əsasları: Enqueue(), TryDequeue()

Enqueue(T item): elementi növbənin sonuna əlavə etmək üçün istifadə olunur. Bu əməliyyat thread-safe-dir. Siz eyni anda 10 müxtəlif thread-dən Enqueue çağıra bilərsiniz və bütün elementlər düzgün əlavə olunacaq.

TryDequeue(out T item): növbənin əvvəlindən elementi çıxarmağa cəhd etmək üçün istifadə olunur. Bu consumer-lər üçün əsas metoddur. Əgər element uğurla çıxarılıbsa true qaytarır (dəyişən item-ə dəyər düşür), əgər növbə boşdursa false qaytarır. Vacibdir ki, TryDequeue növbə boş olsa da thread-i block etmir.

2. TryDequeue() və əməliyyatların atomikliyi önəmi

TryDequeue() sadəcə rahat deyil; o kiritik əhəmiyyətə malikdir düzgün thread-safe işləmə üçün. O atomikdir: növbənin boş olub-olmaması yoxlanılması və elementin çıxarılması vahid, bölünməz əməliyyat kimi həyata keçirilir.

Əgər bizdə ayrı metodlar olsaydı — IsEmpty (növənın boş olub-olmamasını yoxlamaq) və Dequeue (elementi çıxarmaq) — onların çağırışları arasında başqa bir thread növbəni boşalda bilər. Nəticədə sizin Dequeue istisna atardı və ya səhv məlumat verərdi. TryDequeue belə vəziyyəti tam aradan qaldırır.

Nümunə: bir neçə thread ilə Producer-Consumer

Burada iki producer thread və bir consumer thread işə düşür.

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

ConcurrentQueue<int> dataQueue = new ConcurrentQueue<int>();
bool producersDone = false; // consumer-ə siqnal vermək üçün flag

void Producer(int start, int count)
{
    for (int i = 0; i < count; i++)
    {
        dataQueue.Enqueue(start + i);
        Console.WriteLine($"[P] Əlavə etdim: {start + i}");
        Thread.Sleep(10); 
    }
}

void Consumer()
{
    while (!producersDone || dataQueue.Count > 0) // Data var və ya producer-lər işləyir
    {
        if (dataQueue.TryDequeue(out int item))
        {
            Console.WriteLine($"[C] Emal olundu: {item}");
        }
        else
        {
            Thread.Sleep(50); // Növbə boşdursa gözləyirik
        }
    }
    Console.WriteLine("[C] İşini bitirdi.");
}

// Main-də nümunənin işə salınması:
// Task.Run(() => Producer(1, 5));
// Task.Run(() => Producer(100, 5)); // İkinci producer
// Task.Run(() => Consumer());
// Thread.Sleep(600); // Thread-lərə işləmək üçün vaxt veririk
// producersDone = true; // Producer-lərin bitdiyini siqnal veririk
// Thread.Sleep(200); // Consumer-ə qalanları götürmək üçün vaxt veririk

Qeyd edin ki, bu sadə nümunədə producersDone flag-i və Thread.Sleep bitişi imitasiya etmək üçün istifadə olunur. Real tətbiqlərdə daha etibarlı sinxronizasiya üçün adətən CancellationTokenSource və ya BlockingCollection<T> istifadə edilir.

ConcurrentQueue<T> aşağıdakı senarilərə uyğundur:

• Elementlərin emal sırası vacibdir (FIFO).
• Çoxlu thread-lər elementlər əlavə edir və/və ya çoxlu thread-lər onları götürür.
• Manual lock-larla məşğul olmadan yüksək performans lazımdır.

3. Producer-consumer üçün Stek (LIFO)

Stek — LIFO (Last-In, First-Out) prinsipi ilə işləyən başqa bir fundamental data structure-dir, yəni "sonuncu gələn — ilk çıxır". Boğazın yuxarısındakı boşqab yığmağını təsəvvür edin: həmişə ən yuxarıdakı götürülür və yeni boşqab da yuxarıya qoyulur.

ConcurrentStack<T> ConcurrentQueue<T>-də olduğu kimi thread-safe-dir və producer-consumer patternində istifadə oluna bilər, amma emal sırası tərsinədir.

Nümunə: ConcurrentStack — əlavə etmə və çıxarma

using System.Collections.Concurrent;

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

// Komandaların əlavə edilməsi (producer)
commandStack.Push("Mətni seç");
commandStack.Push("Şrift dəyişdir");
commandStack.Push("Sənədi saxla");
Console.WriteLine($"Stekdəki komandalar: {commandStack.Count}"); // Çıxış: Stekdəki komandalar: 3

// Komandaların çıxarılması (consumer)
if (commandStack.TryPop(out string cmd1))
{
    Console.WriteLine($"Əmr ləğv olundu: {cmd1}"); // Çıxış: Əmr ləğv olundu: Sənədi saxla
}

if (commandStack.TryPop(out string cmd2))
{
    Console.WriteLine($"Əmr ləğv olundu: {cmd2}"); // Çıxış: Əmr ləğv olundu: Şrift dəyişdir
}

if (!commandStack.TryPop(out string emptyCmd))
{
    Console.WriteLine("Əmrlər steki boşdur."); // Çıxış: Əmrlər steki boşdur.
}

4. Əməliyyatların əsasları: Push(), TryPop()

Push(T item): elementi stekin üstünə əlavə etmək üçün istifadə olunur. Əməliyyat thread-safe-dir.

TryPop(out T item): stekin üstündən elementi çıxarmağa cəhd etmək üçün istifadə olunur. Əgər element uğurla çıxarılıbsa true, əks halda (stek boşdursa) false qaytarır. TryDequeue kimi, bu da atomik əməliyyatdır və data race-ləri əngəlləyir.

Nümunə: ConcurrentStack-dən object pool üçün istifadə

Stek object pool-un implementasiyası üçün əla uyğundur: götürdün — istifadə etdin — qaytardın.

using System.Collections.Concurrent;

class Connection { /* Sadə stub */ public Guid Id { get; } = Guid.NewGuid(); }

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

// Pool-u ilkin bağlantılarla doldururuq
for (int i = 0; i < 3; i++)
{
    connectionPool.Push(new Connection());
}
Console.WriteLine($"Pool-da bağlantılar: {connectionPool.Count}"); // Çıxış: Pool-da bağlantılar: 3

void UseConnection()
{
    if (connectionPool.TryPop(out Connection conn))
    {
        Console.WriteLine($"[Pool] İstifadə edilən bağlantı: {conn.Id}");
        // Bağlantı ilə işin imitassiyası
        Thread.Sleep(50); 
        connectionPool.Push(conn); // Pool-a geri qaytarırıq
        Console.WriteLine($"[Pool] Qaytarılan bağlantı: {conn.Id}. Pool-da: {connectionPool.Count}");
    }
    else
    {
        Console.WriteLine("[Pool] Bağlantı pool-u boşdur. Yeni yaradıram.");
        // Adətən burada pool boşdursa yeni bağlantı yaradırlar
        connectionPool.Push(new Connection()); 
    }
}

// Main-də nümunənin işə salınması:
Task.Run(() => UseConnection());
Task.Run(() => UseConnection());
Task.Run(() => UseConnection());
Thread.Sleep(500);

Bu nümunədə bir neçə thread təhlükəsiz şəkildə ümumi pool-dan bağlantı götürə və geri qaytara bilər.

5. Tətbiq nümunələri və ConcurrentQueue-lə müqayisə

ConcurrentStack<T> aşağıdakı hallarda istifadə olunur:

• LIFO sırası kritikdirsə (məsələn, "Undo" üçün əməliyyat tarixi).
• Son əlavə edilmiş elementlərə sürətli giriş lazım olduqda (CPU cache-də "hot" qalırlar).
• Stack-based alqoritmlər tətbiq olunursa (DFS, ifadələrin sintaksis analizi).

Müqayisə və uyğun kolleksiyanın seçimi

ConcurrentQueue

ConcurrentStack

Seçim ConcurrentQueue və ConcurrentStack arasında tamamilə sizin producer-consumer senarinizdə tələb olunan emal sırasından asılıdır. Hər iki kolleksiya yüksək performans və thread-safe davranış təklif edir, manual sinxronizasiyadan azad edir və miqyaslana bilən çoxthreadli sistemlər qurmağa kömək edir.

2

Tapşırıq

C# SELF, səviyyə, dərs

Bağlanıb

Producer-Consumer-in Queue ilə reallaşdırılması

Producer-Consumer-in Queue ilə reallaşdırılması

Açıq