ITPub博客

首页 > Linux操作系统 > Linux操作系统 > 适合C# Actor的消息执行方式(5):一个简单的网络爬虫

适合C# Actor的消息执行方式(5):一个简单的网络爬虫

原创 Linux操作系统 作者:iDotNetSpace 时间:2009-07-28 11:05:41 0 删除 编辑

之前的几篇文章大都在摆一些“小道理”,有经验的朋友容易想象出来其中的含义,不过对于那些还不了解Actor模型的朋友来说,这些内容似乎有些太过了。此外,乒乓测试虽然经典,但是不太容易说明问题。因此,今天我们就来看一个简单的有些简陋的网络爬虫,对于Actor模型的使用来说,它至少比乒乓测试能够说明问题。对了,我们先来使用那“中看不中用”的消息执行方式。

<div xmlns:cc="http://creativecommons.org/ns#" about="http://www.cnblogs.com/JeffreyZhao/archive/2009/07/02/embarrassed-mvp.html"><a rel="cc:attributionURL" property="cc:attributionName" href="http://www.cnblogs.com/JeffreyZhao/">赵劼a> / <a rel="license" href="http://creativecommons.org/licenses/by/2.5/cn/">CC BY 2.5a>div>

功能简介

  这个网络爬虫的功能还是用于演示,先来列举出它的实现目标吧:

  • 给出一个初始链接,然后抓取它的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>,已经没有办法接受另一个接口作为消息了,不是吗?

  这就是一个致命的限制:一个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自己发送一条消息表示抓取完毕……呃,但是我们现在做不到啊。

  嗯,下次再说吧。

原文地址http://www.cnblogs.com/JeffreyZhao/archive/2009/07/27/message-execution-model-for-c-sharp-actor-5-a-simple-web-crawler.html

来自 “ ITPUB博客 ” ,链接:http://blog.itpub.net/12639172/viewspace-610622/,如需转载,请注明出处,否则将追究法律责任。

请登录后发表评论 登录
全部评论

注册时间:2008-01-04

  • 博文量
    2376
  • 访问量
    5297366