python 基于mongoDB的urls队列

普通的urls队列储存在内存之中,不适用与多进程情况。我们扩展一个支持多进程的健壮性的urls队列—–基于MongoDB的urls队列。
  更加健壮性的队列,则需要考虑使用专用的消息传输工具,如:Celery。

我们首先给urls队列中的url定义四种状态:
1. OUTSTANAING = 0 未处理 当添加一个新url时,其url状态。
2. PROCESSING = 1 待处理 当url从队列中取出准备下载是,其url状态。
3. COMPLETE = 2 已处理 当下载结束后,其url状态。
4. ERROR = 3 处理失败 当下载失败时,其url状态。

  1. 将新添加进队列的url设置为 OUTSTANAING
    def push(self, url):
        """添加一个url到队列中,设置url状态为 未处理
            :param  url 添加到队列的url"""
        try:
            self.crawl_queue.insert(
                {'url': url, 'status': self.OUTSTANAING}
            )
        except errors.DuplicateKeyError as e:
            pass

对字段属性为 unique(唯一性) 的字段 进行多次重复插入,则会产生DuplicateKeyError错误,我们捕获DuplicateKeyError避免因重复数据对数据库的插入而造成的系统崩溃。

  1. 将从队列中取出的url设置为 PROCESSING
def pop(self): 
       """取得一个url,并更新url状态为待处理
         :return 一个未处理url
         :raise 没有未处理url则抛出KeyError异常"""
        record = self.crawl_queue.find_and_modify(
            query={'status': self.OUTSTANAING},
            update={'$set': {'status': self.PROCESSING, 'wait_time': datetime.datetime.now()}}
        )
        if record:
            return record['url']
        else:
            self.repair()
            raise KeyError()

队列中没有OUTSTANDING状态的url时,则抛出 KeyError,表示当前队列处理完成。
  我们使用了一个timeout参数,默认值为300秒。如果某个URL处理时间超过这个timeout值,我们就认定处理过程中出现错误,URL状态被设置为OUTSTANDING,以便再次处理。

    def repair(self):
        """重置url状态为 未处理"""
        record = self.crawl_queue.find_and_modify(
            query={
                'wait_time': {'%lt': datetime.datetime.now() - datetime.timedelta(seconds=self.timeout)},
                'status': {'$ne': self.COMPLETE},
            },
            update={'$set': {'status': self.OUTSTANAING}}
        )
        if record:
            print '重置url状态', record['url']
  1. 下载结束后,将url设置为COMPLETE
    def complete(self, url):
        """更新url状态为 已处理"""
        self.crawl_queue.update({'url': url}, {'$set': {'status': self.COMPLETE}})
  1. 下载错误时,将url设置为ERROR
    def set_error(self, url):
        """更新url状态,处理异常URL"""
        self.crawl_queue.update({'url': url}, {'$set': {'status': self.ERROR}})

我们对队列进行抽象,以便适应与不同的数据库以及不同的表单字段。

        #mongoDB数据库集合(表单)
        self.crawl_queue = db_table
        #表单字段
        self.table_field = table_field

完整代码如下

class MongoQueue(object):
       """MongoQueue类 基于mongodb的urls队列
         定义了urls的四种状态信息
            OUTSTANAING = 0   未处理
            PROCESSING = 1    待处理
            COMPLETE = 2      已处理
            ERROR = 3         产生异常的url
        当添加一个新url是,其状态为 OUTSTANAING,
        当从队列中取出url准备下载时,其状态为 PROCESSING
        当url下载完成时,其状态为 COMPLETE
        当url处理中(包括下载以及后续解析下载的html)产生异常是,其状态为 ERROR"""

    OUTSTANAING = book_config.OUTSTANAING  # 未处理
    PROCESSING = book_config.PROCESSING  # 待处理
    COMPLETE = book_config.COMPLETE  # 已处理
    ERROR = book_config.ERROR  # 产生异常的url

    def __init__(self, db_table,table_field,timeout=300, mgDB=None):
        """初始化队列
            :param  db_table 需要操作的mongodb数据库集合(表单)
            :param  table_field 数据库集合字段
            :param  timeout url处理时间,如果处理时间大于设定值,则url状态重新设置为OUTSTANAING
            """
        self.crawl_queue = db_table
        self.timeout = timeout
        self.table_field = table_field
        #日志,输出到控制台
        self.console_logger = book_config.CONSOLE_LOGGER

    def __nonzero__(self):
        """查询数据库中是否有状态为 OUTSTANAING 的未处理url
            __nonzero__ 用于将类转换为布尔值。通常在用类进行判断和将类转换成布尔值时调用
            例如 if 对象 相当于 if 对象.__nonzero__()
            :return 如果存在未处理url则返回True,否则返回False"""

        record = self.crawl_queue.find_one(
            {'$or': [{'status': {'$ne': self.COMPLETE}}, {'status': {'$ne': self.ERROR}}]})
        return True if record else False

    def push(self, url):
        """添加一个url到队列中,设置url状态为 未处理
            :param  url 添加到队列的url"""
        try:
            self.crawl_queue.insert(
                {self.table_field: url, 'status': self.OUTSTANAING}
            )
        except errors.DuplicateKeyError as e:
            pass

    def pop(self):
        """取得一个url,并更新url状态为待处理
            :return 一个未处理url
            :raise 没有未处理url则抛出KeyError异常"""
        record = self.crawl_queue.find_and_modify(
            query={'status': self.OUTSTANAING},
            update={'$set': {'status': self.PROCESSING, 'wait_time': datetime.datetime.now()}}
        )
        if record:
            return record[self.table_field]
        else:
            self.repair()
            raise KeyError()

    def repair(self):
        """重置url状态为 未处理"""
        record = self.crawl_queue.find_and_modify(
            query={
                'wait_time': {'%lt': datetime.datetime.now() - datetime.timedelta(seconds=self.timeout)},
                'status': {'$ne': self.COMPLETE},
            },
            update={'$set': {'status': self.OUTSTANAING}}
        )
        if record:
            #print '重置url状态', record[self.table_field]
            self.console_logger.info(u'重置 url : ' + record[self.table_field] + u' 状态')

    def complete(self, url):
        """更新url状态为 已处理"""
        self.crawl_queue.update({self.table_field: url}, {'$set': {'status': self.COMPLETE}})

    def set_error(self, url):
        """更新url状态,处理异常URL"""
        self.crawl_queue.update({self.table_field: url}, {'$set': {'status': self.ERROR}})
if __name__ == '__main__:
       client = pymongo.MongoClient('localhost',27017)
       urls_table = client['urls_table']
       table_field = 'url'
       mq = MongoQueue(urls_table,table_field)
       # 添加一个url
       mq.push(url)
       # 取出一个url
       url = mq.pop()
       # 设置url处理完成
       mq.complete(url)
       # 设置url 处理失败
       mq.set_error(url)
    原文作者:有点d伤
    原文地址: https://www.jianshu.com/p/ef43c16c2fd0
    本文转自网络文章,转载此文章仅为分享知识,如有侵权,请联系博主进行删除。
点赞