Hàng đợi và ngăn xếp Producer-Consumer

1. Giới thiệu

Bắt đầu từ cơ bản. Hàng đợi — là cấu trúc dữ liệu cơ bản, hoạt động theo nguyên tắc FIFO (First-In, First-Out), tức là "vào trước — ra trước". Hãy tưởng tượng hàng đợi ở siêu thị: ai đứng trước sẽ được phục vụ trước.

Trong lập trình đa luồng, mẫu Producer-Consumer — là một trong những pattern phổ biến và mạnh mẽ.

• Producers — là các luồng hoặc phần của ứng dụng tạo dữ liệu hoặc tác vụ và đặt chúng vào hàng đợi chung. Họ "sản xuất" công việc.
• Consumers — là các luồng hoặc phần của ứng dụng lấy dữ liệu hoặc tác vụ từ hàng đợi và xử lý chúng. Họ "tiêu thụ" công việc.

Mẫu này giúp điều tiết luồng dữ liệu, tách rời các thành phần hệ thống (producer không cần biết ai và làm thế nào sẽ xử lý dữ liệu), làm hệ thống phản hồi tốt hơn và giúp phân phối tải đều giữa các luồng.

Ví dụ: ConcurrentQueue — thêm và lấy phần tử

Xem cách thêm và lấy phần tử từ ConcurrentQueue<T>.

using System.Collections.Concurrent;

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

// Thêm phần tử (nhà sản xuất)
tasks.Enqueue("Tải tệp");
tasks.Enqueue("Xử lý hình ảnh");
Console.WriteLine($"Số tác vụ trong hàng đợi: {tasks.Count}"); // Kết quả: Số tác vụ trong hàng đợi: 2

// Lấy phần tử (người tiêu thụ)
if (tasks.TryDequeue(out string task1))
{
    Console.WriteLine($"Đã hoàn thành tác vụ: {task1}"); // Kết quả: Đã hoàn thành tác vụ: Tải tệp
}

if (tasks.TryDequeue(out string task2))
{
    Console.WriteLine($"Đã hoàn thành tác vụ: {task2}"); // Kết quả: Đã hoàn thành tác vụ: Xử lý hình ảnh
}

if (!tasks.TryDequeue(out string emptyTask))
{
    Console.WriteLine("Hàng đợi rỗng, không có tác vụ mới."); // Kết quả: Hàng đợi rỗng, không có tác vụ mới.
}

Nguyên lý cơ bản: Enqueue(), TryDequeue()

Enqueue(T item): Dùng để thêm phần tử vào cuối hàng đợi. Thao tác này là thread-safe. Bạn có thể cùng lúc gọi Enqueue từ 10 luồng khác nhau, và tất cả phần tử sẽ được thêm đúng.

TryDequeue(out T item): Dùng để thử lấy phần tử từ đầu hàng đợi. Đây là method then chốt cho consumer. Nó trả về true nếu phần tử được lấy thành công (giá trị nằm trong tham số xuất item), và false nếu hàng đợi rỗng. Quan trọng là TryDequeue không chặn luồng khi hàng đợi rỗng.

2. Tầm quan trọng của TryDequeue() và tính nguyên tử của các thao tác

Method TryDequeue() không chỉ tiện; nó rất quan trọng để đảm bảo hoạt động thread-safe đúng đắn. Nó có tính nguyên tử: kiểm tra rỗng và chính việc lấy phần tử diễn ra như một thao tác không thể chia cắt.

Nếu chúng ta có các method riêng biệt IsEmpty (kiểm tra hàng đợi rỗng) và Dequeue (lấy phần tử), thì giữa hai lời gọi này một luồng khác có thể làm cạn hàng đợi. Kết quả là Dequeue của bạn sẽ ném exception hoặc trả dữ liệu không đúng. TryDequeue loại trừ hoàn toàn tình huống đó.

Ví dụ: Producer-Consumer với nhiều luồng

Ở đây ta chạy hai producer và một consumer.

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

ConcurrentQueue<int> dataQueue = new ConcurrentQueue<int>();
bool producersDone = false; // Cờ để báo cho consumer

void Producer(int start, int count)
{
    for (int i = 0; i < count; i++)
    {
        dataQueue.Enqueue(start + i);
        Console.WriteLine($"[P] Đã thêm: {start + i}");
        Thread.Sleep(10); 
    }
}

void Consumer()
{
    while (!producersDone || dataQueue.Count > 0) // Tiếp tục chừng nào còn dữ liệu hoặc producers đang chạy
    {
        if (dataQueue.TryDequeue(out int item))
        {
            Console.WriteLine($"[C] Đã xử lý: {item}");
        }
        else
        {
            Thread.Sleep(50); // Chờ nếu hàng đợi rỗng
        }
    }
    Console.WriteLine("[C] Kết thúc.");
}

// Chạy ví dụ trong Main:
// Task.Run(() => Producer(1, 5));
// Task.Run(() => Producer(100, 5)); // Producer thứ hai
// Task.Run(() => Consumer());
// Thread.Sleep(600); // Cho các luồng thời gian làm việc
// producersDone = true; // Báo hiệu producers đã xong
// Thread.Sleep(200); // Cho consumer thời gian lấy phần còn lại

Lưu ý rằng trong ví dụ đơn giản này cờ producersDone và Thread.Sleep được dùng để giả lập việc kết thúc. Trong ứng dụng thực tế để đồng bộ kết thúc đáng tin cậy hơn thường dùng CancellationTokenSource hoặc BlockingCollection<T>.

ConcurrentQueue<T> phù hợp cho các kịch bản:

• Thứ tự xử lý phần tử quan trọng (FIFO).
• Nhiều luồng cùng thêm phần tử, và/hoặc nhiều luồng cùng lấy phần tử.
• Cần hiệu năng cao mà không muốn tự quản lý lock.

3. Ngăn xếp cho producer-consumer (LIFO)

Ngăn xếp — là cấu trúc dữ liệu khác, hoạt động theo nguyên tắc LIFO (Last-In, First-Out), tức là "vào sau — ra trước". Hãy tưởng tượng một chồng đĩa: bạn luôn lấy chiếc trên cùng, và đặt phần tử mới lên trên cùng.

ConcurrentStack<T> cũng thread-safe như ConcurrentQueue<T>, và có thể dùng trong pattern producer-consumer, nhưng với thứ tự xử lý đảo ngược.

Ví dụ: ConcurrentStack — thêm và lấy phần tử

using System.Collections.Concurrent;

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

// Thêm lệnh (producer)
commandStack.Push("Bôi đen văn bản");
commandStack.Push("Thay đổi font");
commandStack.Push("Lưu tài liệu");
Console.WriteLine($"Số lệnh trong ngăn xếp: {commandStack.Count}"); // Kết quả: Số lệnh trong ngăn xếp: 3

// Lấy lệnh (consumer)
if (commandStack.TryPop(out string cmd1))
{
    Console.WriteLine($"Lệnh đã hoàn tác: {cmd1}"); // Kết quả: Lệnh đã hoàn tác: Lưu tài liệu
}

if (commandStack.TryPop(out string cmd2))
{
    Console.WriteLine($"Lệnh đã hoàn tác: {cmd2}"); // Kết quả: Lệnh đã hoàn tác: Thay đổi font
}

if (!commandStack.TryPop(out string emptyCmd))
{
    Console.WriteLine("Ngăn xếp lệnh trống."); // Kết quả: Ngăn xếp lệnh trống.
}

4. Nguyên lý cơ bản: Push(), TryPop()

Push(T item): Dùng để thêm phần tử lên đỉnh ngăn xếp. Thao tác là thread-safe.

TryPop(out T item): Dùng để thử lấy phần tử từ đỉnh ngăn xếp. Trả về true nếu phần tử được lấy thành công, và false nếu ngăn xếp rỗng. Giống như TryDequeue, đây là thao tác nguyên tử, tránh race condition.

Ví dụ: dùng ConcurrentStack để làm object pool

Ngăn xếp phù hợp để implement object pool: lấy — dùng — trả về.

using System.Collections.Concurrent;

class Connection { /* Lớp placeholder đơn giản */ public Guid Id { get; } = Guid.NewGuid(); }

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

// Khởi tạo pool với vài kết nối
for (int i = 0; i < 3; i++)
{
    connectionPool.Push(new Connection());
}
Console.WriteLine($"Số kết nối trong pool: {connectionPool.Count}"); // Kết quả: Số kết nối trong pool: 3

void UseConnection()
{
    if (connectionPool.TryPop(out Connection conn))
    {
        Console.WriteLine($"[Pool] Đã sử dụng kết nối: {conn.Id}");
        // Mô phỏng công việc với kết nối
        Thread.Sleep(50); 
        connectionPool.Push(conn); // Trả lại vào pool
        Console.WriteLine($"[Pool] Trả lại kết nối: {conn.Id}. Trong pool: {connectionPool.Count}");
    }
    else
    {
        Console.WriteLine("[Pool] Pool kết nối rỗng. Tạo mới.");
        // Thường thì ở đây tạo kết nối mới nếu pool rỗng
        connectionPool.Push(new Connection()); 
    }
}

// Chạy ví dụ trong Main:
Task.Run(() => UseConnection());
Task.Run(() => UseConnection());
Task.Run(() => UseConnection());
Thread.Sleep(500);

Trong ví dụ này nhiều luồng có thể an toàn lấy và trả kết nối vào pool chung.

5. Ví dụ ứng dụng và so sánh với ConcurrentQueue

ConcurrentStack<T> thường dùng khi:

• Thứ tự LIFO quan trọng (ví dụ: lịch sử hành động cho chức năng "Undo").
• Cần truy cập nhanh tới các phần tử vừa được thêm gần đây (thường "nóng" trong cache CPU).
• Triển khai các thuật toán dựa trên stack (duyệt đồ thị theo chiều sâu, phân tích cú pháp biểu thức).

So sánh và chọn collection phù hợp

ConcurrentQueue

ConcurrentStack

Việc chọn giữa ConcurrentQueue và ConcurrentStack phụ thuộc hoàn toàn vào thứ tự xử lý bạn cần trong scenario producer-consumer. Cả hai collection đều cung cấp hiệu năng cao và tính thread-safe "out of the box", giải phóng bạn khỏi việc lo lock thủ công và giúp xây dựng hệ thống đa luồng có thể mở rộng.

2

Nhiệm vụ

C# SELF, mức độ, bài học

Đã khóa

Triển khai Producer-Consumer sử dụng hàng đợi

Triển khai Producer-Consumer sử dụng hàng đợi

Mở