C# Actor模型開發實例：網絡爬蟲

作者：老趙 2009-08-05 16:04:27

本文通過建立一個網絡爬蟲的實例，介紹了適合C# Actor的開發方法。

之前的幾篇文章大都在擺一些“小道理”，有經驗的朋友容易想象出來其中的含義，不過對于那些還不了解Actor模型的朋友來說，這些內容似乎有些太過了。此外，乒乓測試雖然經典，但是不太容易說明問題。因此，今天我們就來看一個簡單的有些簡陋的網絡爬蟲，對于Actor模型的使用來說，它至少比乒乓測試能夠說明問題。對了，我們先來使用那“中看不中用”的消息執行方式。

功能簡介

這個網絡爬蟲的功能還是用于演示，先來列舉出它的實現目標吧：
◆給出一個初始鏈接，然后抓取它的HTML并分析出所有html鏈接，然后繼續爬，不斷爬，直到爬完所有鏈接為止。
◆多線程運行，我們可以指定由多少個爬蟲同時工作。
◆多個爬蟲組成一個“工作單元”，程序中可以同時出現多個工作單元，工作單元之間互相獨立。
◆能簡化的地方便簡化，如一切不涉及任何永久性存儲（也就是說，只使用內存），沒有太復雜的容錯機制。

的確很簡單吧？那么，現在您不妨先在腦海中想象一下，在不用Actor模型的時候您會怎么實現這個功能。然后，我們就要動手使用ActorLite這個小類庫了。

協議制定

正如我們不斷強調的那樣，在Actor模型中唯一的通信方式便是互相發送消息。于是使用Actor模型的第一步往往便是設計Actor類型，以及它們之間傳遞的消息。在這個簡單的場景中，我們會定義兩種Actor類型。一是Monitor，二是Crawler。一個Monitor便代表一個“工作單元”，它管理了多個爬蟲，即Crawler。

Monitor將負責在合適的時候創建Crawler，并向其發送一個消息，讓其開始工作。在我們的系統中，我們使用ICrawlRequestHandler接口來表示這個消息：

C# Actor模型·創建網絡爬蟲：

public interface ICrawlRequestHandler  
{  
    void Crawl(Monitor monitor, string url);  
}

在接受到上面的Crawl消息后，Crawler將去抓取指定的url對象，并將結果發還給Monitor。在這里我們要求報告Cralwer向Monitor報告“成功”和“失敗”兩種消息1：

C# Actor模型·爬蟲報告消息

public interface ICrawlResponseHandler  
{  
    void Succeeded(Crawler crawler, string url, List<string> links);  
    void Failed(Crawler crawler, string url, Exception ex);  
}

我們使用“接口”這種方式定義了“消息組”，把Succeeded和Failed兩種關系密切的消息綁定在一起。如果抓取成功，則Crawler會從抓取內容中獲得額外的鏈接，并發還給Monitor——失敗的時候自然就發還一個異常對象。此外，無論是成功還是失敗，我們都會把Crawler對象交給Monitor，Monitor會安排給Crawler新的抓取任務。

因此，Monitor和Cralwer類的定義大約應該是這樣的：

C# Actor模型·爬蟲和監控類的定義

public class Monitor : Actor<Action<ICrawlResponseHandler>>, ICrawlResponseHandler  
{  
    protected override void Receive(Action<ICrawlResponseHandler> message)  
    {  
        message(this);  
    }  
 
    #region ICrawlResponseHandler Members  
 
    void ICrawlResponseHandler.Succeeded(Crawler crawler, string url, List<string> links)  
    {  
        ...  
    }  
 
    void ICrawlResponseHandler.Failed(Crawler crawler, string url, Exception ex)  
    {  
        ...  
    }  
 
    #endregion  
}  
 
public class Crawler : Actor<Action<ICrawlRequestHandler>>, ICrawlRequestHandler  
{  
    protected override void Receive(Action<ICrawlRequestHandler> message)  
    {  
        message(this);  
    }  
 
    #region ICrawlRequestHandler Members  
 
    void ICrawlRequestHandler.Crawl(Monitor monitor, string url)  
    {  
        ...  
    }  
 
    #endregion  
}

Crawler實現

我們先從簡單的Crawler類的實現開始。Crawler類只需要實現ICrawlRequestHandler接口的Crawl方法即可：

C# Actor模型·爬蟲的實現

void ICrawlRequestHandler.Crawl(Monitor monitor, string url)  
{  
    try 
    {  
        string content = new WebClient().DownloadString(url);  
 
        var matches = Regex.Matches(content, @"href=""(http://[^""]+)""").Cast<Match>();  
        var links = matches.Select(m => m.Groups[1].Value).Distinct().ToList();  
        monitor.Post(m => m.Succeeded(this, url, links));  
    }  
    catch (Exception ex)  
    {  
        monitor.Post(m => m.Failed(this, url, ex));  
    }  
}

沒錯，使用WebClient下載頁面內容只需要一行代碼就可以了。然后便是使用正則表達式提取出頁面上所有的鏈接。很顯然這里是有問題的，因為我們我只分析出以“http://”開頭的地址，但是無視其他的“相對地址”——不過作為一個小實驗來說已經足夠說明問題了。最后自然是使用Post方法將結果發還給Monitor。在拋出異常的情況下，這幾行代碼的邏輯也非常自然。

Monitor實現

Monitor相對來說便略顯復雜了一些。我們知道，Monitor要負責控制Crawler的數量，那么必然需要負責維護一些必要的字段：

C# Actor模型·監控的實現

private HashSet<string> m_allUrls; // 所有待爬或爬過的url  
private Queue<string> m_readyToCrawl; // 待爬的url  
 
public int MaxCrawlerCount { private set; get; } // 最大爬蟲數目  
public int WorkingCrawlerCount { private set; get; } // 正在工作的爬蟲數目  
 
public Monitor(int maxCrawlerCount)  
{  
    this.m_allUrls = new HashSet<string>();  
    this.m_readyToCrawl = new Queue<string>();  
    this.MaxCrawlerCount = maxCrawlerCount;  
    this.WorkingCrawlerCount = 0;  
}

Monitor要處理的自然是ICrawlResponseHandler中的Succeeded或Failed方法：

void ICrawlResponseHandler.Succeeded(Crawler crawler, string url, List<string> links)  
{  
    Console.WriteLine("{0} crawled, {1} link(s).", url, links.Count);  
 
    foreach (var newUrl in links)  
    {  
        if (!this.m_allUrls.Contains(newUrl))  
        {  
            this.m_allUrls.Add(newUrl);  
            this.m_readyToCrawl.Enqueue(newUrl);  
        }  
    }  
 
    this.DispatchCrawlingTasks(crawler);  
}  
 
void ICrawlResponseHandler.Failed(Crawler crawler, string url, Exception ex)  
{  
    Console.WriteLine("{0} error occurred: {1}.", url, ex.Message);  
    this.DispatchCrawlingTasks(crawler);  
}

在抓取成功時，Monitor將遍歷links列表中的所有地址，如果發現新的url，則加入相關集合中。在抓取失敗的情況下，我們也只是簡單的繼續下去而已。而“繼續”則是由DispatchCrawlingTasks方法實現的，我們需要傳入一個“可復用”的Crawler對象：

C# Actor模型·爬蟲的傳入

private void DispatchCrawlingTasks(Crawler reusableCrawler)  
{  
    if (this.m_readyToCrawl.Count <= 0)  
    {  
        this.WorkingCrawlerCount--;  
        return;  
    }  
 
    var url = this.m_readyToCrawl.Dequeue();  
    reusableCrawler.Post(c => c.Crawl(this, url));  
 
    while (this.m_readyToCrawl.Count > 0 &&  
        this.WorkingCrawlerCount < this.MaxCrawlerCount)  
    {  
        var newUrl = this.m_readyToCrawl.Dequeue();  
        new Crawler().Post(c => c.Crawl(this, newUrl));  
 
        this.WorkingCrawlerCount++;  
    }  
}

如果已經沒有需要抓取的內容了，則直接拋棄Crawler對象即可，否則則分派一個新任務。接著便不斷創建新的爬蟲，分配新的抓取任務，直到爬蟲數額用滿，或者沒有需要抓取的內容位置。

使用

我們使用區區幾十行代碼遍實現了一個簡單的多線程爬蟲，其中一個關鍵便是使用了Actor模型。使用Actor模型，對象之間通過消息傳遞進行交互。而且對于單個Actor對象來說，消息的執行完全是線程安全的。因此，我們只要作用最直接的邏輯便可以完成整個實現，從而回避了內存共享的并行模式中所使用的互斥體、鎖等各類組件。

不過有沒有發現，我們沒有一個入口可以“開啟”一個抓取任務啊，Monitor類中還缺少了點什么。好吧，那么我們補上一個Start方法：

C# Actor模型·爬蟲開始工作

public class Monitor : Actor<Action<ICrawlResponseHandler>>, ICrawlResponseHandler  
{  
    ...  
 
    public void Start(string url)  
    {  
        this.m_allUrls.Add(url);  
        this.WorkingCrawlerCount++;  
        new Crawler().Post(c => c.Crawl(this, url));  
    }  
}

于是，我們便可以這樣打開一個或多個抓取任務：

static class Program  
{  
    static void Main(string[] args)  
    {  
        new Monitor(5).Start("http://www.cnblogs.com/");  
        new Monitor(10).Start("http://www.csdn.net/");  
 
        Console.ReadLine();  
    }  
}

這里我們新建兩個工作單元，也就是啟動了兩個抓取任務。一是使用5個爬蟲抓取cnblogs.com，二是使用10個爬蟲抓取csdn.net。

缺陷

這里的缺陷是什么？其實很明顯，您發現了嗎？

使用Actor模型可以保證消息執行的線程安全，不過很明顯Start方法并非如此，我們只能用它來“開啟”一個抓取任務。但是如果我們想再次“手動”提交一個需要抓取的URL怎么辦？所以理想的方法，其實也應該是向Monitor發送一個消息來啟動第一個URL抓取任務。需要補充，則發送多個URL即可。可是，這個消息定義在什么地方才合適呢？我們的Monitor類已經實現了Actor<Action<ICrawlResponseHandler>>，已經沒有辦法接受另一個接口作為消息了，不是嗎？

這就是一個致命的限制：一個Actor雖然可以實現多個接口，但只能接受其中一個作為消息。同樣的，如果我們要為Monitor提供其他功能，例如“查詢”某個URL的抓取狀態，也因為同樣的原因而無法實現。還有，便是在前幾篇文章中談到的問題了。Crawler和Monitor直接耦合，我們向Crawler發送的消息只能攜帶一個Monitor對象。

最后，便是一個略顯特別的問題了。我們這里使用WebClient的DownloadString方法來獲取網頁的內容，但是這是個同步IO操作，理想的做法中我們應該使用異步的方法。所以，我們可以這么寫：

void ICrawlRequestHandler.Crawl(Monitor monitor, string url)  
{  
    WebClient webClient = new WebClient();  
    webClient.DownloadStringCompleted += (sender, e) =>  
    {  
        if (e.Error == null)  
        {  
            var matches = Regex.Matches(e.Result, @"href=""(http://[^""]+)""").Cast<Match>();  
            var links = matches.Select(m => m.Groups[1].Value).Distinct().ToList();  
            monitor.Post(m => m.Succeeded(this, url, links));  
        }  
        else 
        {  
            monitor.Post(m => m.Failed(this, url, e.Error));  
        }  
    };  
    webClient.DownloadStringAsync(new Uri(url));  
}

如果您還記得老趙在最近一篇文章中關于IO線程池的討論，就可以了解到DownloadStringCompleted事件的處理方法會在統一的IO線程池中運行，這樣我們無法控制其運算能力。因此，我們應該在回調函數中向Crawler自己發送一條消息表示抓取完畢……呃，但是我們現在做不到埃

嗯，下次再說吧。

【編輯推薦】

強類型和Actor：ActorLite的演示
C#的Tag Message回顧：繁瑣而危險
Erlang的Actor回顧：將消息轉化為邏輯執行
Actor模型的本質：究竟是要解決什么問題
順暢的使用C# Actor：另一個解決方案

責任編輯：yangsai 來源：老趙點滴

C# Actor模型