C# 爬蟲多執行緒架構設計：生產者消費者、佇列去重、優雅停機、速率限制

前言

單執行緒爬蟲受限於網路 I/O 等待，CPU 利用率極低。本文設計一個**生產級多執行緒爬蟲架構**：BlockingCollection 佇列去重、SemaphoreSlim 速率限制、CancellationToken 優雅停機、HttpClient 連線池、錯誤重試與指數退避、進度監控。核心邏輯約 200 行，可直接擴充為分散式爬蟲。

架構圖

[種子 URL] → [URL 佇列 (去重)] → [N 個下載 Worker] → [HTML 解析] → [新 URL 入佇列] / [資料入庫]
                    ↑                    ↓
              [速率限制器]         [錯誤重試/退避]
                    ↓
            [CancellationToken] → 優雅停機

核心資料結構

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Net.Http;
using System.Threading;
using System.Threading.Tasks;

public class CrawlOptions
{
    public int MaxDegreeOfParallelism { get; set; } = 10;
    public int MaxUrls { get; set; } = 10000;
    public int RequestTimeoutSeconds { get; set; } = 30;
    public int MaxRetries { get; set; } = 3;
    public TimeSpan BaseRetryDelay { get; set; } = TimeSpan.FromSeconds(2);
    public double RequestsPerSecond { get; set; } = 5; // 速率限制
    public string UserAgent { get; set; } = "MyCrawler/1.0 (+https://example.com/bot)";
    public HashSet AllowedDomains { get; set; } = new();
    public Func ShouldCrawl { get; set; } = _ => true;
}

public class CrawlResult
{
    public string Url { get; set; }
    public string Html { get; set; }
    public HttpResponseMessage Response { get; set; }
    public Exception Error { get; set; }
    public int RetryCount { get; set; }
}

爬蟲核心類別

public class MultiThreadedCrawler : IDisposable
{
    private readonly CrawlOptions _opt;
    private readonly HttpClient _http;
    private readonly BlockingCollection _urlQueue;
    private readonly ConcurrentDictionary _visited;
    private readonly SemaphoreSlim _rateLimiter;
    private readonly CancellationTokenSource _cts = new();
    private readonly List _workers = new();
    private long _completed, _failed, _enqueued;

    public MultiThreadedCrawler(CrawlOptions opt)
    {
        _opt = opt;
        _http = new HttpClient(new HttpClientHandler
        {
            AutomaticDecompression = System.Net.DecompressionMethods.GZip | System.Net.DecompressionMethods.Deflate,
            MaxConnectionsPerServer = opt.MaxDegreeOfParallelism * 2
        })
        {
            Timeout = TimeSpan.FromSeconds(opt.RequestTimeoutSeconds)
        };
        _http.DefaultRequestHeaders.UserAgent.ParseAdd(_opt.UserAgent);

        _urlQueue = new BlockingCollection(new ConcurrentQueue(), opt.MaxUrls);
        _visited = new ConcurrentDictionary();
        _rateLimiter = new SemaphoreSlim((int)(opt.RequestsPerSecond * 2), (int)(opt.RequestsPerSecond * 2)); // 令牌桶近似
    }

    public async Task StartAsync(IEnumerable seeds, Func onData, Action onError = null)
    {
        // 1. 啟動速率補充器
        _ = Task.Run(RefillRateLimiterAsync, _cts.Token);

        // 2. 種子入佇列
        foreach (var url in seeds) EnqueueUrl(url);

        // 3. 啟動 Worker
        for (int i = 0; i < _opt.MaxDegreeOfParallelism; i++)
            _workers.Add(Task.Run(() => WorkerLoopAsync(onData, onError), _cts.Token));

        // 4. 等待完成或取消
        await Task.WhenAll(_workers).ConfigureAwait(false);
    }

    public void Stop() => _cts.Cancel();

    private void EnqueueUrl(string url)
    {
        if (_visited.TryAdd(NormalizeUrl(url), 0))
        {
            try { _urlQueue.Add(url, _cts.Token); Interlocked.Increment(ref _enqueued); }
            catch (OperationCanceledException) { }
        }
    }

    private async Task WorkerLoopAsync(Func onData, Action onError)
    {
        foreach (var url in _urlQueue.GetConsumingEnumerable(_cts.Token))
        {
            await _rateLimiter.WaitAsync(_cts.Token); // 速率限制
            try
            {
                var result = await DownloadWithRetryAsync(url);
                if (result.Error == null)
                {
                    await onData(result);
                    ExtractAndEnqueueLinks(result.Html, result.Response.RequestMessage.RequestUri);
                }
                else
                {
                    onError?.Invoke(result.Error);
                    Interlocked.Increment(ref _failed);
                }
            }
            catch (OperationCanceledException) { break; }
            catch (Exception ex) { onError?.Invoke(ex); Interlocked.Increment(ref _failed); }
            finally { Interlocked.Increment(ref _completed); }
        }
    }

    private async Task DownloadWithRetryAsync(string url)
    {
        for (int attempt = 0; attempt <= _opt.MaxRetries; attempt++)
        {
            try
            {
                var resp = await _http.GetAsync(url, _cts.Token);
                if (resp.IsSuccessStatusCode)
                {
                    var html = await resp.Content.ReadAsStringAsync();
                    return new CrawlResult { Url = url, Html = html, Response = resp, RetryCount = attempt };
                }
                if ((int)resp.StatusCode >= 500 || resp.StatusCode == System.Net.HttpStatusCode.TooManyRequests)
                    throw new HttpRequestException($"HTTP {(int)resp.StatusCode}");
            }
            catch (Exception ex) when (attempt < _opt.MaxRetries)
            {
                var delay = TimeSpan.FromSeconds(_opt.BaseRetryDelay.TotalSeconds * Math.Pow(2, attempt)); // 指數退避
                await Task.Delay(delay, _cts.Token);
            }
        }
        return new CrawlResult { Url = url, Error = new Exception("Max retries exceeded") };
    }

    private void ExtractAndEnqueueLinks(string html, Uri baseUri)
    {
        // 簡易 Regex 抽取，生產環境建議用 AngleSharp / HtmlAgilityPack
        var matches = System.Text.RegularExpressions.Regex.Matches(html, @"href\s*=\s*[""'](?[^""']+)[""']", System.Text.RegularExpressions.RegexOptions.IgnoreCase);
        foreach (System.Text.RegularExpressions.Match m in matches)
        {
            if (Uri.TryCreate(baseUri, m.Groups["url"].Value, out var abs) && _opt.ShouldCrawl(abs.AbsoluteUri))
                EnqueueUrl(abs.AbsoluteUri);
        }
    }

    private async Task RefillRateLimiterAsync()
    {
        var interval = TimeSpan.FromSeconds(1.0 / _opt.RequestsPerSecond);
        while (!_cts.Token.IsCancellationRequested)
        {
            await Task.Delay(interval, _cts.Token);
            _rateLimiter.Release(Math.Min(10, _rateLimiter.CurrentCount + 10)); // 批次補充
        }
    }

    public (long Completed, long Failed, long Enqueued, int QueueRemaining) GetStats()
        => (_completed, _failed, _enqueued, _urlQueue.Count);

    public void Dispose()
    {
        _cts.Cancel();
        _urlQueue.CompleteAdding();
        Task.WhenAll(_workers).Wait(TimeSpan.FromSeconds(10));
        _http.Dispose();
        _rateLimiter.Dispose();
        _cts.Dispose();
    }

    private string NormalizeUrl(string url)
    {
        if (Uri.TryCreate(url, UriKind.Absolute, out var u))
            return u.Scheme + "://" + u.Host + u.PathAndQuery.TrimEnd('/').ToLowerInvariant();
        return url.ToLowerInvariant();
    }
}

使用範例

var opt = new CrawlOptions
{
    MaxDegreeOfParallelism = 20,
    MaxUrls = 50000,
    RequestsPerSecond = 10,
    AllowedDomains = new HashSet { "example.com" },
    ShouldCrawl = url => new Uri(url).Host.EndsWith("example.com")
};

var crawler = new MultiThreadedCrawler(opt);

await crawler.StartAsync(new[] { "https://example.com" },
    onData: async result =>
    {
        // 解析、存資料庫、推送訊息佇列
        Console.WriteLine($"[OK] {result.Url} ({result.Html.Length} chars)");
        await Task.CompletedTask;
    },
    onError: ex => Console.WriteLine($"[ERR] {ex.Message}"));

// 監控迴圈
var timer = new System.Timers.Timer(5000);
timer.Elapsed += (s, e) =>
{
    var (c, f, q, r) = crawler.GetStats();
    Console.WriteLine($"完成:{c} 失敗:{f} 佇列:{r} 總入隊:{q}");
};
timer.Start();

Console.ReadLine();
crawler.Stop();

關鍵設計細節解析

進階擴充方向

1. 分散式：URL 佇列換成 Redis Stream / RabbitMQ，`_visited` 換成 Redis Bloom Filter / Redis Set。
2. 斷點續爬：定期將 `_visited`、佇列快照持久化到資料庫/檔案，啟動時恢復。
3. 動態速率：根據 429/5xx 比例自動調整 `RequestsPerSecond` (AIMD 算法)。
4. JS 渲染：整合 PuppeteerSharp / PlaywrightSharp 處理 SPA 頁面。
5. 監控面板：推送統計到 Prometheus + Grafana，設定告警規則。
6. 禮貌爬蟲：遵守 `robots.txt`、`Crawl-delay`、User-Agent 識別、聯繫資訊。

總結

以 `BlockingCollection` + `SemaphoreSlim` + `CancellationToken` 為核心，構建出**高併發、可控速率、優雅停機、可觀測**的爬蟲骨架。核心邏輯不超過 200 行，卻涵蓋生產環境 90% 的剛需。下一步視業務複雜度，引入分散式佇列、瀏覽器自動化、智慧解析器，即可演進為企業級爬蟲平台。