Concurrency Patterns trong .NET 10 — Xử lý Song song Hiệu quả

Posted on: 4/20/2026 7:10:47 PM

Trong thế giới backend hiện đại, khả năng xử lý hàng nghìn request đồng thời là yêu cầu cơ bản. .NET 10 cung cấp một bộ công cụ concurrency mạnh mẽ — từ async/await quen thuộc đến System.Threading.Channels và Parallel.ForEachAsync. Tuy nhiên, sử dụng sai cách có thể biến ứng dụng từ "concurrent" thành "deadlocked". Bài viết này đi sâu vào 7 pattern concurrency thực chiến, kèm code production-ready và phân tích khi nào nên (và không nên) dùng từng pattern.

Tổng quan Concurrency Model trong .NET 10

.NET 10 sử dụng mô hình concurrency dựa trên ThreadPool và Task. Không giống với model thread-per-request truyền thống, .NET tận dụng tối đa thread pool bằng cách "nhả" thread khi đợi I/O và "mượn" lại thread khác khi có kết quả. Điều này cho phép một ứng dụng ASP.NET Core phục vụ hàng chục nghìn concurrent request chỉ với vài chục thread.

graph TB
    subgraph ThreadPool["🏊 .NET ThreadPool"]
        T1["Thread 1"]
        T2["Thread 2"]
        T3["Thread N"]
    end

    subgraph Tasks["📋 Task Queue"]
        TA["Task A
HTTP Request"]
        TB["Task B
DB Query"]
        TC["Task C
File I/O"]
        TD["Task D
API Call"]
    end

    TA --> T1
    TB --> T2
    TC --> T1
    TD --> T3

    style ThreadPool fill:#f8f9fa,stroke:#e94560,color:#2c3e50
    style Tasks fill:#f8f9fa,stroke:#2c3e50,color:#2c3e50
    style T1 fill:#e94560,stroke:#fff,color:#fff
    style T2 fill:#e94560,stroke:#fff,color:#fff
    style T3 fill:#e94560,stroke:#fff,color:#fff
    style TA fill:#fff,stroke:#2c3e50,color:#2c3e50
    style TB fill:#fff,stroke:#2c3e50,color:#2c3e50
    style TC fill:#fff,stroke:#2c3e50,color:#2c3e50
    style TD fill:#fff,stroke:#2c3e50,color:#2c3e50

Pattern 1: async/await — Nền tảng của mọi thứ

async/await không tạo thread mới — nó giải phóng thread hiện tại khi đợi I/O. Đây là pattern cơ bản nhất nhưng cũng dễ mắc sai lầm nhất.

// ✅ ĐÚNG: Chạy concurrent 3 HTTP calls
public async Task<DashboardData> GetDashboardAsync(int userId)
{
    var userTask = _userService.GetByIdAsync(userId);
    var ordersTask = _orderService.GetRecentAsync(userId);
    var statsTask = _analyticsService.GetStatsAsync(userId);

    await Task.WhenAll(userTask, ordersTask, statsTask);

    return new DashboardData
    {
        User = userTask.Result,
        Orders = ordersTask.Result,
        Stats = statsTask.Result
    };
}

// ❌ SAI: Sequential — chậm gấp 3 lần
public async Task<DashboardData> GetDashboardSlowAsync(int userId)
{
    var user = await _userService.GetByIdAsync(userId);
    var orders = await _orderService.GetRecentAsync(userId);
    var stats = await _analyticsService.GetStatsAsync(userId);
    // Mỗi await đợi xong mới chạy tiếp → tổng = T1 + T2 + T3
    return new DashboardData { User = user, Orders = orders, Stats = stats };
}

// ❌ NGUY HIỂM — Deadlock trong ASP.NET
public ActionResult GetData()
{
    var data = _service.GetDataAsync().Result; // Block thread pool thread!
    return Ok(data);
}

// ✅ AN TOÀN — Luôn dùng async end-to-end
public async Task<ActionResult> GetData()
{
    var data = await _service.GetDataAsync();
    return Ok(data);
}

Pattern 2: SemaphoreSlim — Kiểm soát mức độ đồng thời

Khi cần giới hạn số lượng tác vụ chạy song song (ví dụ: gọi API bên thứ 3 có rate limit), SemaphoreSlim là lựa chọn nhẹ nhất và hiệu quả nhất.

public class RateLimitedHttpClient
{
    private readonly HttpClient _http;
    private readonly SemaphoreSlim _semaphore;

    public RateLimitedHttpClient(HttpClient http, int maxConcurrency = 10)
    {
        _http = http;
        _semaphore = new SemaphoreSlim(maxConcurrency, maxConcurrency);
    }

    public async Task<List<ApiResult>> FetchAllAsync(IEnumerable<string> urls)
    {
        var tasks = urls.Select(async url =>
        {
            await _semaphore.WaitAsync();
            try
            {
                return await _http.GetFromJsonAsync<ApiResult>(url);
            }
            finally
            {
                _semaphore.Release();
            }
        });

        var results = await Task.WhenAll(tasks);
        return results.ToList();
    }
}

Pattern 3: System.Threading.Channels — Pipeline Producer-Consumer

Channel<T> là cấu trúc dữ liệu thread-safe, lock-free, được thiết kế riêng cho pattern producer-consumer trong .NET. So với BlockingCollection<T> cũ, Channel nhanh hơn 10 lần nhờ tránh kernel-mode synchronization và hỗ trợ async natively.

graph LR
    P1["Producer 1
HTTP Receiver"] --> CH["Channel<T>
Bounded Buffer"]
    P2["Producer 2
File Watcher"] --> CH
    CH --> C1["Consumer 1
Processor"]
    CH --> C2["Consumer 2
Processor"]
    C1 --> DB["Database"]
    C2 --> DB

    style P1 fill:#4CAF50,stroke:#fff,color:#fff
    style P2 fill:#4CAF50,stroke:#fff,color:#fff
    style CH fill:#e94560,stroke:#fff,color:#fff
    style C1 fill:#2c3e50,stroke:#fff,color:#fff
    style C2 fill:#2c3e50,stroke:#fff,color:#fff
    style DB fill:#f8f9fa,stroke:#2c3e50,color:#2c3e50

public class OrderProcessingPipeline : BackgroundService
{
    private readonly Channel<Order> _channel;

    public OrderProcessingPipeline()
    {
        // Bounded channel: tối đa 1000 items,
        // producer sẽ await khi buffer đầy (backpressure)
        _channel = Channel.CreateBounded<Order>(new BoundedChannelOptions(1000)
        {
            FullMode = BoundedChannelFullMode.Wait,
            SingleReader = false,
            SingleWriter = false
        });
    }

    public ChannelWriter<Order> Writer => _channel.Writer;

    protected override async Task ExecuteAsync(CancellationToken ct)
    {
        // Chạy 4 consumer song song
        var consumers = Enumerable.Range(0, 4)
            .Select(_ => ConsumeAsync(ct));

        await Task.WhenAll(consumers);
    }

    private async Task ConsumeAsync(CancellationToken ct)
    {
        await foreach (var order in _channel.Reader.ReadAllAsync(ct))
        {
            await ProcessOrderAsync(order);
        }
    }

    private async Task ProcessOrderAsync(Order order)
    {
        // Validate → Calculate → Save → Notify
        await _validator.ValidateAsync(order);
        order.Total = await _calculator.CalculateAsync(order);
        await _repository.SaveAsync(order);
        await _notifier.NotifyAsync(order);
    }
}

Pattern 4: Parallel.ForEachAsync — Data Parallelism có kiểm soát

Khi cần xử lý một collection lớn song song với giới hạn concurrency, Parallel.ForEachAsync (có từ .NET 6) là lựa chọn tốt hơn so với tự quản lý SemaphoreSlim + Task.WhenAll.

public async Task ResizeImagesAsync(List<string> imagePaths)
{
    var options = new ParallelOptions
    {
        MaxDegreeOfParallelism = Environment.ProcessorCount,
        CancellationToken = _cts.Token
    };

    await Parallel.ForEachAsync(imagePaths, options, async (path, ct) =>
    {
        using var image = await Image.LoadAsync(path, ct);
        image.Mutate(x => x.Resize(800, 600));

        var outputPath = Path.Combine(_outputDir, Path.GetFileName(path));
        await image.SaveAsync(outputPath, ct);
    });
}

// So sánh: cách cũ phức tạp hơn
public async Task ResizeImagesOldWayAsync(List<string> imagePaths)
{
    var semaphore = new SemaphoreSlim(Environment.ProcessorCount);
    var tasks = imagePaths.Select(async path =>
    {
        await semaphore.WaitAsync();
        try { /* ... resize logic ... */ }
        finally { semaphore.Release(); }
    });
    await Task.WhenAll(tasks);
}

Pattern 5: Pipeline Pattern — Xử lý đa tầng

Pipeline pattern chia quá trình xử lý thành nhiều stage, mỗi stage chạy concurrent và truyền dữ liệu qua Channel. Đây là pattern lý tưởng cho ETL, image processing, và data ingestion.

graph LR
    S1["Stage 1
Download"] --> |Channel| S2["Stage 2
Parse"]
    S2 --> |Channel| S3["Stage 3
Transform"]
    S3 --> |Channel| S4["Stage 4
Load to DB"]

    style S1 fill:#e94560,stroke:#fff,color:#fff
    style S2 fill:#2c3e50,stroke:#fff,color:#fff
    style S3 fill:#4CAF50,stroke:#fff,color:#fff
    style S4 fill:#16213e,stroke:#fff,color:#fff

public class DataIngestionPipeline
{
    public async Task RunAsync(IEnumerable<string> sourceUrls, CancellationToken ct)
    {
        var downloadChannel = Channel.CreateBounded<RawData>(100);
        var parseChannel = Channel.CreateBounded<ParsedData>(100);
        var transformChannel = Channel.CreateBounded<TransformedData>(100);

        var pipeline = Task.WhenAll(
            DownloadStageAsync(sourceUrls, downloadChannel.Writer, ct),
            ParseStageAsync(downloadChannel.Reader, parseChannel.Writer, ct),
            TransformStageAsync(parseChannel.Reader, transformChannel.Writer, ct),
            LoadStageAsync(transformChannel.Reader, ct)
        );

        await pipeline;
    }

    private async Task DownloadStageAsync(
        IEnumerable<string> urls,
        ChannelWriter<RawData> writer,
        CancellationToken ct)
    {
        try
        {
            await Parallel.ForEachAsync(urls,
                new ParallelOptions { MaxDegreeOfParallelism = 5, CancellationToken = ct },
                async (url, token) =>
                {
                    var data = await _http.GetByteArrayAsync(url, token);
                    await writer.WriteAsync(new RawData(url, data), token);
                });
        }
        finally
        {
            writer.Complete();
        }
    }

    private async Task ParseStageAsync(
        ChannelReader<RawData> reader,
        ChannelWriter<ParsedData> writer,
        CancellationToken ct)
    {
        try
        {
            await foreach (var raw in reader.ReadAllAsync(ct))
            {
                var parsed = JsonSerializer.Deserialize<ParsedData>(raw.Bytes);
                await writer.WriteAsync(parsed, ct);
            }
        }
        finally
        {
            writer.Complete();
        }
    }

    // TransformStage và LoadStage tương tự...
}

Pattern 6: Periodic Background Processing

Thay vì dùng Timer cũ dễ gây reentrancy (callback chạy lại khi callback trước chưa xong), .NET 10 có PeriodicTimer — async-friendly và đảm bảo không overlap.

public class MetricsCollector : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken ct)
    {
        using var timer = new PeriodicTimer(TimeSpan.FromSeconds(30));

        while (await timer.WaitForNextTickAsync(ct))
        {
            try
            {
                var metrics = await CollectMetricsAsync(ct);
                await _metricsStore.PushAsync(metrics, ct);
            }
            catch (Exception ex) when (ex is not OperationCanceledException)
            {
                _logger.LogError(ex, "Metrics collection failed");
                // Không throw — timer tiếp tục tick
            }
        }
    }
}

// ❌ SAI: System.Timers.Timer có thể overlap
var timer = new System.Timers.Timer(30000);
timer.Elapsed += async (s, e) =>
{
    // Nếu CollectMetrics mất > 30s,
    // callback thứ 2 sẽ chạy song song → race condition
    await CollectMetricsAsync();
};

Pattern 7: Cancellation Token Propagation

CancellationToken không chỉ là tham số để "có cho đầy đủ" — nó là cơ chế graceful shutdown sống còn cho ứng dụng production. Khi Kubernetes gửi SIGTERM, ứng dụng cần dọn dẹp sạch sẽ trong thời gian grace period.

public class OrderService
{
    public async Task<OrderResult> ProcessWithTimeoutAsync(
        Order order,
        CancellationToken requestCt)
    {
        // Kết hợp: cancel khi request hủy HOẶC timeout 30s
        using var timeoutCts = new CancellationTokenSource(TimeSpan.FromSeconds(30));
        using var linkedCts = CancellationTokenSource
            .CreateLinkedTokenSource(requestCt, timeoutCts.Token);

        try
        {
            var result = await _processor.ProcessAsync(order, linkedCts.Token);
            return result;
        }
        catch (OperationCanceledException) when (timeoutCts.IsCancellationRequested)
        {
            _logger.LogWarning("Order {Id} processing timed out", order.Id);
            throw new TimeoutException($"Order {order.Id} exceeded 30s limit");
        }
        catch (OperationCanceledException) when (requestCt.IsCancellationRequested)
        {
            _logger.LogInformation("Order {Id} cancelled by client", order.Id);
            throw; // Re-throw — client đã disconnect
        }
    }
}

graph TD
    A["Request CancellationToken"] --> L["LinkedTokenSource"]
    B["Timeout CancellationToken
30 giây"] --> L
    L --> C["ProcessAsync()"]
    C --> D{"Token cancelled?"}
    D -->|Timeout| E["TimeoutException"]
    D -->|Client disconnect| F["OperationCanceledException"]
    D -->|Không| G["✅ Success"]

    style A fill:#4CAF50,stroke:#fff,color:#fff
    style B fill:#ff9800,stroke:#fff,color:#fff
    style L fill:#e94560,stroke:#fff,color:#fff
    style G fill:#4CAF50,stroke:#fff,color:#fff
    style E fill:#ff9800,stroke:#fff,color:#fff
    style F fill:#f8f9fa,stroke:#2c3e50,color:#2c3e50
    style C fill:#2c3e50,stroke:#fff,color:#fff
    style D fill:#f8f9fa,stroke:#e94560,color:#2c3e50

Bảng tổng hợp: Chọn Pattern nào?

Benchmark: Channel vs BlockingCollection vs ConcurrentQueue

Benchmark dưới đây chạy trên .NET 10, 1 triệu items, 4 producer + 4 consumer:

Real-world: Xây dựng Image Processing Service

Kết hợp tất cả pattern vào một service xử lý ảnh thực tế:

public class ImageProcessingService : BackgroundService
{
    private readonly Channel<ImageJob> _jobs;
    private readonly ILogger<ImageProcessingService> _logger;

    public ImageProcessingService(ILogger<ImageProcessingService> logger)
    {
        _logger = logger;
        _jobs = Channel.CreateBounded<ImageJob>(new BoundedChannelOptions(500)
        {
            FullMode = BoundedChannelFullMode.Wait
        });
    }

    // API controller gọi method này
    public async ValueTask EnqueueAsync(ImageJob job, CancellationToken ct)
        => await _jobs.Writer.WriteAsync(job, ct);

    protected override async Task ExecuteAsync(CancellationToken ct)
    {
        _logger.LogInformation("Image processing started with {Count} workers",
            Environment.ProcessorCount);

        var workers = Enumerable.Range(0, Environment.ProcessorCount)
            .Select(id => ProcessWorkerAsync(id, ct));

        await Task.WhenAll(workers);
    }

    private async Task ProcessWorkerAsync(int workerId, CancellationToken ct)
    {
        await foreach (var job in _jobs.Reader.ReadAllAsync(ct))
        {
            using var timeoutCts = new CancellationTokenSource(TimeSpan.FromMinutes(2));
            using var linked = CancellationTokenSource
                .CreateLinkedTokenSource(ct, timeoutCts.Token);

            try
            {
                await ProcessJobAsync(job, linked.Token);
                _logger.LogDebug("Worker {Id}: processed {File}", workerId, job.FileName);
            }
            catch (OperationCanceledException) when (timeoutCts.IsCancellationRequested)
            {
                _logger.LogWarning("Worker {Id}: {File} timed out", workerId, job.FileName);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Worker {Id}: failed {File}", workerId, job.FileName);
            }
        }
    }

    private static async Task ProcessJobAsync(ImageJob job, CancellationToken ct)
    {
        using var image = await Image.LoadAsync(job.SourcePath, ct);

        foreach (var size in job.TargetSizes)
        {
            ct.ThrowIfCancellationRequested();
            var clone = image.Clone(x => x.Resize(size.Width, size.Height));
            var path = Path.Combine(job.OutputDir, $"{size.Name}_{job.FileName}");
            await clone.SaveAsync(path, ct);
        }
    }
}

Những sai lầm phổ biến

// ❌ Exception bị nuốt, không ai biết task fail
_ = ProcessAsync(data);

// ✅ Dùng Channel hoặc IHostedService để track
await _channel.Writer.WriteAsync(data);

// ❌ Exception crash toàn bộ process
async void OnButtonClicked() { await DoWorkAsync(); }

// ✅ Luôn return Task
async Task OnButtonClickedAsync() { await DoWorkAsync(); }

// ❌ Parallel.ForEach KHÔNG hỗ trợ async delegate đúng cách
Parallel.ForEach(urls, async url =>
{
    await _http.GetAsync(url); // async void ngầm!
});

// ✅ Dùng Parallel.ForEachAsync
await Parallel.ForEachAsync(urls, async (url, ct) =>
{
    await _http.GetAsync(url, ct);
});

Tổng kết

Concurrency trong .NET 10 không phải là "thêm async vào mọi method" — mà là chọn đúng pattern cho đúng bài toán. Channel<T> cho producer-consumer, SemaphoreSlim cho throttling, Parallel.ForEachAsync cho data parallelism, và CancellationToken xuyên suốt mọi nơi. Nắm vững 7 pattern này, bạn đã có đủ công cụ để xây dựng hệ thống xử lý hàng triệu request mỗi giây mà không một thread nào bị block lãng phí.

Tham khảo

• Task-based Asynchronous Programming — Microsoft Learn
• 12 Production-Ready Async & Parallel Patterns — Mahesh Kumar Yadav
• Advanced C# Concurrency: Channels, Pipelines, and Parallel Processing — DEV Community
• Task Parallel Library (TPL) — Microsoft Learn
• Handling High-Concurrency Scenarios in C# — Or Ben Shmueli