1. Bối cảnh vấn đề
Trong ứng dụng đơn luồng các collection như List<T>, Dictionary<T> hoạt động rất dự đoán được. Nhưng ngay khi nhiều thread truy cập cùng một collection cùng lúc, vấn nạn quen thuộc xuất hiện: race conditions (đua dữ liệu).
Nếu vài thread cùng đọc và/hoặc ghi vào cùng một collection mà không có đồng bộ đúng cách, bạn có thể gặp:
- Dữ liệu sai: phần tử có thể bị xóa bởi một thread trong khi thread khác đang cố cập nhật nó.
- Mất dữ liệu: một thread thêm phần tử, thread khác ghi đè lên mà không biết về ghi trước đó.
- Exception: collection có thể vào trạng thái không hợp lệ, và bạn sẽ thấy InvalidOperationException (ví dụ "Collection was modified; enumeration operation may not execute.") hoặc thậm chí NullReferenceException.
Ví dụ 1: Race condition trong List<T> (cộng đơn giản)
Hai thread cùng tăng cùng một phần tử của list.
using System.Collections.Generic;
using System.Threading.Tasks; // Cho Task.Run
class RaceConditionExample
{
static List<int> numbers = new List<int> { 0 }; // Danh sách với một phần tử
static void Main(string[] args)
{
Console.WriteLine("Giá trị ban đầu: " + numbers[0]); // 0
// Chạy hai task, mỗi task tăng numbers[0]
Task task1 = Task.Run(() => IncrementNumbers(500_000));
Task task2 = Task.Run(() => IncrementNumbers(500_000));
Task.WaitAll(task1, task2); // Chờ cả hai xong
Console.WriteLine("Giá trị cuối cùng: " + numbers[0]); // Mong đợi 1_000_000, nhưng...
// Kết quả gần như luôn nhỏ hơn 1_000_000!
}
static void IncrementNumbers(int count)
{
for (int i = 0; i < count; i++)
{
// Phép numbers[0]++ thực ra gồm 3 bước:
// 1. Đọc numbers[0]
// 2. Tăng giá trị lên 1
// 3. Ghi giá trị mới vào numbers[0]
numbers[0]++;
}
}
}
Tại sao đây là race? Nếu thread A đọc numbers[0] (giá trị 0), rồi thread B cũng đọc numbers[0] (cũng 0) trước khi A kịp ghi 1, thì cả hai sẽ tăng 0 lên 1 và ghi 1. Một lần tăng bị mất. Phép numbers[0]++ không phải là atomic.
Ví dụ 2: InvalidOperationException khi thay đổi Dictionary
Một thread lặp qua dictionary, thread khác thay đổi nó.
using System.Collections.Generic;
using System.Threading; // Cho Thread.Sleep
class DictionaryRaceExample
{
static Dictionary<int, string> users = new Dictionary<int, string>();
static void Main(string[] args)
{
// Khởi tạo dictionary
for (int i = 0; i < 5; i++) users.Add(i, $"User {i}");
// Thread đọc
Thread readerThread = new Thread(() =>
{
try
{
foreach (var user in users) // Lặp qua dictionary
{
Console.WriteLine($"Người đọc: {user.Key} - {user.Value}");
Thread.Sleep(10); // Giả lập công việc
}
}
catch (InvalidOperationException ex)
{
Console.WriteLine($"Người đọc: LỖI! {ex.Message}");
}
});
// Thread ghi
Thread writerThread = new Thread(() =>
{
Thread.Sleep(5); // Cho reader bắt đầu trước một chút
for (int i = 5; i < 10; i++)
{
users.Add(i, $"New User {i}"); // Thêm phần tử
Console.WriteLine($"Người ghi: Đã thêm User {i}");
Thread.Sleep(15);
}
});
readerThread.Start();
writerThread.Start();
readerThread.Join(); // Chờ các thread xong
writerThread.Join();
Console.WriteLine("Ví dụ kết thúc.");
}
}Tại sao xảy ra lỗi? Dictionary<TKey, TValue> (như List<T>) không được thiết kế để đọc và viết đồng thời bởi nhiều thread mà không đồng bộ. Khi thread ghi thay đổi cấu trúc nội bộ, thread đọc vẫn tiếp tục foreach trên dữ liệu đã bị thay đổi, dẫn đến InvalidOperationException.
2. Tại sao lock đơn giản không phải lúc nào cũng tối ưu?
Ý tưởng "bọc mọi thứ bằng lock" nghe đơn giản, nhưng có vài nhược điểm:
// Ví dụ tệ: lock quá nhiều
// (Chỉ để minh họa, đừng làm thế!)
static object _lock = new object();
static List<int> _sharedList = new List<int>();
void AddItem(int item)
{
lock (_lock)
{
_sharedList.Add(item);
}
}
int GetItemCount()
{
lock (_lock)
{
return _sharedList.Count;
}
}- Hiệu năng (bottleneck): lock chặn truy cập toàn bộ collection. Với 100 thread thì 99 thread sẽ chờ một thread, dù các thao tác có thể không xung đột trực tiếp.
- Độ phức tạp: phải nhớ đặt lock ở mọi chỗ dùng collection. Quên một chỗ là race quay lại.
- Deadlock: nhiều lock trên các object khác nhau dễ gây deadlock.
- Iterator: foreach không cứu bạn nếu thread khác sửa collection.
Vì vậy trong .NET người ta thêm các collection thread-safe đặc biệt.
Toán tử atomic
Collection thread-safe — đảm bảo hoạt động đúng khi nhiều thread truy cập cùng lúc mà không cần lock bên ngoài từ người dùng. Chìa khóa là toán tử atomic: hành động thực thi toàn bộ hoặc không thực thi — các thread khác không thấy trạng thái "nửa chừng".
- Thêm, xóa, đọc — hành xử như thể thực thi từng cái một.
- Bên trong dùng kỹ thuật thấp: các phép interlocked (Interlocked), Compare-And-Swap (CAS), các lock nhẹ — thay vì khoá toàn cục cả collection.
3. Tổng quan System.Collections.Concurrent
Namespace System.Collections.Concurrent cung cấp một bộ collection được thiết kế từ đầu cho đa luồng. Triết lý của chúng là tối đa song song, tối thiểu khoá.
- Hiệu năng: scale tốt khi tăng số core.
- Đơn giản: không cần tự thêm lock cho mỗi thao tác.
- Ít lỗi hơn: giảm các vấn đề do đồng bộ thủ công.
- Tối ưu cho cạnh tranh: hoạt động hiệu quả khi có nhiều thêm/xóa cùng lúc.
4. Các lớp chính
ConcurrentQueue<T> (queue thread-safe)
Nguyên tắc: FIFO — "đầu tiên vào, đầu tiên ra". Dùng cho: producer–consumer, logging, hàng đợi task.
using System.Collections.Concurrent;
ConcurrentQueue<string> messageQueue = new ConcurrentQueue<string>();
void Producer() => messageQueue.Enqueue("Thông điệp 1");
void Consumer()
{
if (messageQueue.TryDequeue(out string message))
{
Console.WriteLine($"Đã xử lý: {message}");
}
else
{
Console.WriteLine("Hàng đợi rỗng.");
}
}ConcurrentStack<T> (stack thread-safe)
Nguyên tắc: LIFO — "vào sau, ra trước". Dùng cho: lịch sử thao tác, DFS, object pools.
using System.Collections.Concurrent;
ConcurrentStack<int> historyStack = new ConcurrentStack<int>();
void PushAction(int value) => historyStack.Push(value);
void PopAction()
{
if (historyStack.TryPop(out int action))
{
Console.WriteLine($"Hành động đã hoàn tác: {action}");
}
else
{
Console.WriteLine("Stack rỗng.");
}
}ConcurrentBag<T> (bag thread-safe)
Collection không có thứ tự, không đảm bảo thứ tự truy xuất. Tối ưu cho kịch bản "thread thường lấy những gì chính nó đã thêm". Rất phù hợp cho object pool.
using System.Collections.Concurrent;
ConcurrentBag<System.Guid> objectPool = new ConcurrentBag<System.Guid>();
void AddObject() => objectPool.Add(System.Guid.NewGuid());
void TakeObject()
{
if (objectPool.TryTake(out System.Guid obj))
{
Console.WriteLine($"Lấy được object: {obj}");
}
else
{
Console.WriteLine("Pool rỗng.");
}
}ConcurrentDictionary<TKey, TValue> (dictionary thread-safe)
Hỗ trợ các thao tác thêm, cập nhật và lấy giá trị theo key một cách atomic. Rất tốt cho cache, session, counter.
using System.Collections.Concurrent;
ConcurrentDictionary<string, int> userScores = new ConcurrentDictionary<string, int>();
void UpdateScore(string user, int score)
{
// Thêm atomic nếu chưa có, hoặc cập nhật nếu có
userScores.AddOrUpdate(user, score, (key, existingVal) => existingVal + score);
Console.WriteLine($"Điểm {user}: {userScores[user]}");
}
void GetScore(string user)
{
if (userScores.TryGetValue(user, out int score))
{
Console.WriteLine($"Điểm hiện tại của {user}: {score}");
}
else
{
Console.WriteLine($"Không tìm thấy người dùng {user}.");
}
}5. Khi nào dùng mấy collection này thay cho bình thường?
- Ứng dụng đa luồng: nếu chỉ có một thread thì collection bình thường nhanh hơn (không có overhead).
- Một collection chung cho nhiều thread: dấu hiệu chính để dùng System.Collections.Concurrent.
- Cần hiệu năng cao và scale: các collection này thiết kế để giảm chờ đợi.
- Muốn code đơn giản hơn: không muốn lo lock tay quanh mọi thao tác.
- Cần các thao tác atomic: thêm/xóa/lấy sẽ không đưa collection vào trạng thái bất nhất.
Đừng dùng Concurrent-collection khi:
- Ứng dụng hoàn toàn đơn luồng.
- Cần "transactionality" cho nhiều thao tác liên quan (có thể cần đồng bộ bên ngoài hoặc cơ chế khác).
- Cần thứ tự rút ra nghiêm ngặt ở nơi mà nó không được đảm bảo (ví dụ ConcurrentBag<T>).
GO TO FULL VERSION