适合C# Actor的消息执行方式（5）：一个简单的网络爬虫
2009-07-27 09:13 by 老赵, 7928 visits之前的几篇文章大都在摆一些“小道理”，有经验的朋友容易想象出来其中的含义，不过对于那些还不了解Actor模型的朋友来说，这些内容似乎有些太过了。此外，乒乓测试虽然经典，但是不太容易说明问题。因此，今天我们就来看一个简单的有些简陋的网络爬虫，对于Actor模型的使用来说，它至少比乒乓测试能够说明问题。对了，我们先来使用那“中看不中用”的消息执行方式。
功能简介
这个网络爬虫的功能还是用于演示，先来列举出它的实现目标吧：
- 给出一个初始链接，然后抓取它的HTML并分析出所有html链接，然后继续爬，不断爬，直到爬完所有链接为止。
- 多线程运行，我们可以指定由多少个爬虫同时工作。
- 多个爬虫组成一个“工作单元”，程序中可以同时出现多个工作单元，工作单元之间互相独立。
- 能简化的地方便简化，如一切不涉及任何永久性存储（也就是说，只使用内存），没有太复杂的容错机制。
的确很简单吧？那么，现在您不妨先在脑海中想象一下，在不用Actor模型的时候您会怎么实现这个功能。然后，我们就要动手使用ActorLite这个小类库了。
协议制定
正如我们不断强调的那样，在Actor模型中唯一的通信方式便是互相发送消息。于是使用Actor模型的第一步往往便是设计Actor类型，以及它们之间传递的消息。在这个简单的场景中，我们会定义两种Actor类型。一是Monitor，二是Crawler。一个Monitor便代表一个“工作单元”，它管理了多个爬虫，即Crawler。
Monitor将负责在合适的时候创建Crawler，并向其发送一个消息，让其开始工作。在我们的系统中，我们使用ICrawlRequestHandler接口来表示这个消息：
public interface ICrawlRequestHandler { void Crawl(Monitor monitor, string url); }
在接受到上面的Crawl消息后，Crawler将去抓取指定的url对象，并将结果发还给Monitor。在这里我们要求报告Cralwer向Monitor报告“成功”和“失败”两种消息1：
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类的定义大约应该是这样的：
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方法即可：
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的数量，那么必然需要负责维护一些必要的字段：
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对象：
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方法：
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自己发送一条消息表示抓取完毕……呃，但是我们现在做不到啊。
嗯，下次再说吧。
相关文章
- 适合C# Actor的消息执行方式（1）：Erlang中的模式匹配
- 适合C# Actor的消息执行方式（2）：C# Actor的尴尬
- 适合C# Actor的消息执行方式（3）：中看不中用的解决方案
- 适合C# Actor的消息执行方式（4）：阶段性总结
- 适合C# Actor的消息执行方式（5）：一个简单的网络爬虫
- 适合C# Actor的消息执行方式（6）：协变与逆变
境界不到,无法做出评论
佩服楼主的强大