普通的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)
网友评论