Xử lý dữ liệu song song

1. Введение

Hôm nay chúng ta sẽ tìm hiểu cách xử lý lượng dữ liệu lớn nhanh nhất có thể, tận dụng mọi lõi CPU có sẵn trên máy tính (hoặc server). Để làm điều này chúng ta sẽ dùng các lớp trong namespace System.Threading.Tasks.Parallel, cụ thể là các phương thức Parallel.For và Parallel.ForEach.

Как быть, если задача — чистый CPU-bound?

Vòng lặp for hoặc foreach truyền thống xử lý các phần tử lần lượt. Đơn giản và đáng tin cậy. Nhưng nếu bạn có CPU đa lõi, vòng lặp đó chỉ dùng một lõi, những lõi còn lại khá nhàn. Sao không chia mảng ra thành các phần và xử lý đồng thời trên nhiều lõi?

Пример:

// Tính tổng các bình phương từ 1 đến N
long sum = 0;
for (int i = 1; i <= 1_000_000; i++)
{
    sum += i * i;
}

Đoạn mã này đơn giản nhưng chạy tuần tự. Nếu phân tán công việc ra các lõi thì sao?

Meet the Family: Parallel.For и Parallel.ForEach

Что это такое?

• Parallel.For — hoạt động giống vòng lặp for thông thường, nhưng chia công việc thành các phần và tự động phân phối lên các thread, tận dụng các lõi có sẵn.
• Parallel.ForEach — xử lý một collection như foreach bình thường, nhưng chạy song song.

Официальная документация:

• Parallel.For
• Parallel.ForEach

Почему это удобно?

Bạn không phải tự tạo, khởi chạy và quản lý thread. Framework lo phần nặng cho bạn. Bạn viết code gần giống vòng lặp thông thường, còn song song hóa xảy ra tự động ở bên dưới.

2. Синтаксис: базовые примеры

Parallel.For

long total = 0;
Parallel.For(1, 1_000_001, i =>
{
    // Lambda này có thể được thực thi đồng thời trên nhiều thread
    Interlocked.Add(ref total, i * i); // Để tránh xung đột dữ liệu
});
Console.WriteLine($"Tổng bình phương: {total}");

Обратите внимание: biến total chúng ta cập nhật qua Interlocked.Add — để tránh race condition.

Parallel.ForEach

var numbers = Enumerable.Range(1, 10_000_000).ToArray();
long sum = 0;

Parallel.ForEach(numbers, num =>
{
    Interlocked.Add(ref sum, num * num); // Cộng an toàn
});
Console.WriteLine($"Tổng bình phương: {sum}");

Взгляд изнутри (Визуальная схема)

+-------------------+
|Tập hợp/Phạm vi    |
+---------+---------+
          |
          v
  +----------------------+
  |   Parallel.ForEach   |
  +----------+-----------+
             |
        +----+----+----+----+
        |         |         |
        v         v         v
  Task #1    Task #2    Task #3   ... (lõi có sẵn)
        |         |         |
     +--+----+  +--+-----+  +--+-----+
     |Xử lý   |  |Xử lý   |  |Xử lý   |
     +-------+  +--------+  +--------+
        \         |         /
         +--------+--------+
                  |
                  v
              Kết quả

3. Анализ больших файлов (CPU-bound обработка)

Giả sử ta có một file text với hàng chục nghìn dòng — mỗi dòng chứa một số. Cần đọc file, bình phương từng số và tính tổng các bình phương.

Синхронная версия

string[] lines = File.ReadAllLines("numbers.txt");
long sum = 0;

foreach (var line in lines)
{
    if (long.TryParse(line, out long n))
    {
        sum += n * n;
    }
}
Console.WriteLine($"Tổng bình phương: {sum}");

Параллельная версия с Parallel.For

string[] lines = File.ReadAllLines("numbers.txt");
long sum = 0;

Parallel.For(0, lines.Length, i =>
{
    if (long.TryParse(lines[i], out long n))
    {
        Interlocked.Add(ref sum, n * n);
    }
});
Console.WriteLine($"Tổng bình phương: {sum}");

Что изменилось: chúng ta thay vòng lặp thường bằng vòng lặp song song, và biến sum giờ tăng bằng Interlocked.Add — để tránh xung đột giữa các thread.

4. Что происходит под капотом?

Khi bạn gọi Parallel.For hoặc Parallel.ForEach, .NET tự chia công việc thành các mảnh và phân phối lên các lõi CPU có sẵn, dùng thread pool. Mỗi mảnh được xử lý độc lập trên một thread riêng.

Преимущество: nếu bạn có 4 lõi, công việc có thể nhanh gần 4 lần (nếu tác vụ không phụ thuộc tài nguyên ngoài và không bị tắc ở các nút cổ chai khác như bộ nhớ hay tốc độ đọc đĩa).

Сравним время выполнения

var numbers = Enumerable.Range(1, 100_000_000).ToArray();
long sumSync = 0;
var sw = System.Diagnostics.Stopwatch.StartNew();

foreach (var n in numbers)
    sumSync += n * n;

sw.Stop();
Console.WriteLine($"Sync: {sw.ElapsedMilliseconds} ms, tổng: {sumSync}");

long sumParallel = 0;
sw.Restart();

Parallel.ForEach(numbers, n =>
    Interlocked.Add(ref sumParallel, n * n)
);

sw.Stop();
Console.WriteLine($"Parallel: {sw.ElapsedMilliseconds} ms, tổng: {sumParallel}");

Попробуйте сами! Trên máy mạnh tốc độ có thể tăng nhiều lần, nhưng tất cả phụ thuộc vào loại công việc và các nút cổ chai.

5. Полезные нюансы

Управление степенью параллелизма

Đôi khi bạn muốn giới hạn số thread dùng (ví dụ để không làm quá tải hệ thống). Dùng MaxDegreeOfParallelism:

using System.Threading.Tasks;
long sum = 0;
var options = new ParallelOptions { 
    MaxDegreeOfParallelism = 2 
};
Parallel.For(0, 100, options, i =>
{
    Interlocked.Add(ref sum, i * i);
});
Console.WriteLine($"Tổng bình phương: {sum}");

Где это полезно: nếu biết một phần tính toán tải nhiều lên đĩa hơn CPU — đặt số thread thấp hơn và đo ảnh hưởng tới hiệu năng.

Когда использовать параллельные циклы

Расширение: что ещё умеет Parallel?

Parallel.Invoke() — chạy nhiều method độc lập cùng lúc:

static void DoTask1() => Console.WriteLine("Tác vụ 1 hoàn tất");
static void DoTask2() => Console.WriteLine("Tác vụ 2 hoàn tất");
static void DoTask3() => Console.WriteLine("Tác vụ 3 hoàn tất");

Parallel.Invoke(
    () => DoTask1(),
    () => DoTask2(),
    () => DoTask3()
);

Mỗi method sẽ được chạy trên lõi khác nếu có thể.

Применение в реальной жизни

• Обработка изображений: xử lý đồng thời các khối khác nhau (ví dụ áp filter).
• Вычисления по массивам: tính toán tài chính, mô phỏng (đánh giá portfolio theo kịch bản).
• Работа с большими журналами логов: tìm kiếm và tổng hợp trên nhiều lõi.
• Машинное обучение: chia nhỏ tác vụ độc lập (batch dữ liệu, feature engineering).

Và đương nhiên, khi phỏng vấn bạn có thể không chỉ giải thích vòng lặp song song là gì mà còn thẳng thắn nêu ưu và nhược của chúng.

6. Типичные ошибки при работе с Parallel.For и Parallel.ForEach

Ошибка №1: Игнорирование гонок данных.
Cập nhật biến chung mà không dùng Interlocked hoặc lock dẫn tới kết quả sai do truy cập đồng thời giữa các thread (xung đột dữ liệu).

Ошибка №2: Использование для I/O-bound задач.
Vòng lặp song song không tăng tốc các tác vụ phụ thuộc đĩa hoặc mạng, thậm chí có thể làm chậm vì overhead.

Ошибка №3: Предположение о порядке выполнения.
Vòng lặp song song không đảm bảo thứ tự xử lý phần tử, điều này có thể phá vỡ logic nếu bạn phụ thuộc vào thứ tự.

Ошибка №4: Игнорирование побочных эффектов.
Thay đổi trạng thái chung (ví dụ collection) trong vòng lặp song song có thể gây lỗi nếu không dùng cấu trúc thread-safe.