python 基于mongoDB的urls队列

作者: 4ffde5305e8f | 来源:发表于2017-04-12 20:19 被阅读0次

python 基于mongoDB的urls队列
我的爬虫日记
celery--turorial
MongoEngine所使用的操作符
celery redis rabbitMQ各是什么及之间的区别？
基于MongoDB的python日志功能
django+celery异步任务
Django中关于URL配置文件urls.py的理解
Python urls分发
使用Python连接mongoDB

普通的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状态。

将新添加进队列的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避免因重复数据对数据库的插入而造成的系统崩溃。

将从队列中取出的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']

下载结束后，将url设置为COMPLETE

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

下载错误时，将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)