我正在使用Python的线程和队列模块编写一个简单的网络爬虫。我获取页面,检查链接并将它们放入队列中。当某个线程完成对页面的处理后,它从队列中获取下一个页面。为了过滤掉已经访问过的页面,我使用一个数组来保存这些页面,并将要添加到队列中的链接与该数组进行比较。但是,如果有多个线程同时在不同的页面上获取相同的链接,那么它们会向队列中添加重复的链接。那么,如何判断某个URL是否已经在队列中,以避免将其再次放入队列中呢?
我正在使用Python的线程和队列模块编写一个简单的网络爬虫。我获取页面,检查链接并将它们放入队列中。当某个线程完成对页面的处理后,它从队列中获取下一个页面。为了过滤掉已经访问过的页面,我使用一个数组来保存这些页面,并将要添加到队列中的链接与该数组进行比较。但是,如果有多个线程同时在不同的页面上获取相同的链接,那么它们会向队列中添加重复的链接。那么,如何判断某个URL是否已经在队列中,以避免将其再次放入队列中呢?
如果您不关心处理项目的顺序,我建议使用Queue
的一个子类,在内部使用set
:
class SetQueue(Queue):
def _init(self, maxsize):
self.maxsize = maxsize
self.queue = set()
def _put(self, item):
self.queue.add(item)
def _get(self):
return self.queue.pop()
正如Paul McGuire所指出的那样,这将允许在从“待处理”集合中删除并尚未添加到“已处理”集合中之后添加重复项。为了解决这个问题,您可以在Queue
实例中存储两个集合,但由于您正在使用更大的集合来检查项目是否已被处理,因此您可以同样回到正确排序请求的queue
。
class SetQueue(Queue):
def _init(self, maxsize):
Queue._init(self, maxsize)
self.all_items = set()
def _put(self, item):
if item not in self.all_items:
Queue._put(self, item)
self.all_items.add(item)
与单独使用集合相比,这样做的优点在于,Queue
的方法是线程安全的,因此您无需额外锁定即可检查其他集合。
SetQueue().put(["foo", "bar"])
会失败。 - azmeuk如果不重写put方法,join调用将会永远阻塞。请参考https://github.com/python/cpython/blob/master/Lib/queue.py#L147
class UniqueQueue(Queue):
def put(self, item, block=True, timeout=None):
if item not in self.queue: # fix join bug
Queue.put(self, item, block, timeout)
def _init(self, maxsize):
self.queue = set()
def _put(self, item):
self.queue.add(item)
def _get(self):
return self.queue.pop()
put
以确保unfinished_tasks
准确,并且join
按预期工作。from queue import Queue
class UniqueQueue(Queue):
def _init(self, maxsize):
self.all_items = set()
Queue._init(self, maxsize)
def put(self, item, block=True, timeout=None):
if item not in self.all_items:
self.all_items.add(item)
Queue.put(self, item, block, timeout)
SetQueue
。import Queue
class SetQueue(Queue.Queue):
def _init(self, maxsize):
Queue.Queue._init(self, maxsize)
self.all_items = set()
def _put(self, item):
if item not in self.all_items:
Queue.Queue._put(self, item)
self.all_items.add(item)
def _get(self):
item = Queue.Queue._get(self)
self.all_items.remove(item)
return item
super(SetQueue, self)._init()
此外,_init
函数没有 maxsize
参数 - __init__
接收该参数。 - lucidyan使用:
url in q.queue
如果url
在队列中,则返回True
我解决这个问题的方法(实际上我是用Scala而不是Python)是同时使用Set和Queue,只有当它们在Set中不存在时才将链接添加到Queue(和Set)中。
Set和Queue都封装在单个线程中,仅向消费者线程公开类似于队列的接口。
编辑:其他人建议使用SQLite,如果访问的URL集合需要增长,那么我也正在考虑这一点。(目前每次爬行只有几百页,因此轻松适合内存。)但是数据库也可以封装在Set本身中,因此消费者线程无需知道它。
SQLite非常简单易用,而且非常适合...只是一个建议。
这是我的代码。希望它能帮助到某个人。
我将 set 添加为队列类的缓存。此缓存用于任务唯一性检查。此外,它还用于在队列类中实现 __contains__
魔法方法。
唯一性可以有两种定义。 首先,任务在整个队列的生命周期内是唯一的。换句话说,即使任务完成并从队列中删除,队列也会拒绝接受重复任务。我将其实现为“be_unique_in_all_items”。其次,任务仅在队列中现有任务中是唯一的。这意味着任务可以在完成后被接受。我将其实现为“be_unique_in_existing_items”。
from queue import Queue
from traceback import print_exc
class MQueue(Queue):
def __init__(self,
**kwargs):
super().__init__(maxsize=kwargs.get("maxsize", 0))
self._be_unique_in_existing_items = kwargs.get("be_unique_in_existing_items", False)
self._be_unique_in_all_items = kwargs.get("be_unique_in_all_items", False)
if self._be_unique_in_existing_items and self._be_unique_in_all_items:
raise ValueError("Choose one criteria")
self.cache = set()
def get(self, *args, **kwargs):
result = super().get(*args, **kwargs)
if result:
if self._be_unique_in_existing_items:
self.cache.remove(result)
return result
def put(self, item, *args, **kwargs):
if self._be_unique_in_existing_items or self._be_unique_in_all_items:
if item in self.cache:
raise ValueError("The given item exists in cache.")
self.cache.add(item)
return super().put(item, *args, **kwargs)
def __contains__(self, item):
if self._be_unique_in_existing_items or self._be_unique_in_all_items:
return self.cache.__contains__(item)
else:
return Queue.__contains__(item) # will raise you error
if __name__ == "__main__":
# ordinary queue
ordinary_queue_obj = MQueue(maxsize=0)
ordinary_queue_obj.put(1)
ordinary_queue_obj.put(1)
try:
print(1 in ordinary_queue_obj)
except Exception:
print_exc()
# be unique in existing queue
unique_in_existing_queue_obj = MQueue(maxsize=0,
be_unique_in_existing_items=True)
unique_in_existing_queue_obj.put(1)
print(1 in unique_in_existing_queue_obj)
try:
unique_in_existing_queue_obj.put(1)
except ValueError:
print_exc()
task = unique_in_existing_queue_obj.get()
unique_in_existing_queue_obj.task_done()
unique_in_existing_queue_obj.put(task)
# be unique in all queue
unique_in_all_queue_obj = MQueue(maxsize=0,
be_unique_in_all_items=True)
unique_in_all_queue_obj.put(1)
print(1 in unique_in_all_queue_obj)
try:
unique_in_all_queue_obj.put(1)
except ValueError:
print_exc()
task = unique_in_all_queue_obj.get()
unique_in_all_queue_obj.task_done()
try:
print(task in unique_in_all_queue_obj)
unique_in_all_queue_obj.put(task)
except ValueError:
print_exc()
注意:set 只能包含可哈希对象。对于不可哈希的对象,请使用 list。
遗憾的是,我没有足够的评分来评论Lukáš Lalinský最好的答案。
为了支持Lukáš Lalinský的SetQueue的第二个变体的SetQueue.task_done()
和SetQueue.join()
,请在if语句中添加else分支:
def _put(self, item):
if item not in self.all_items:
Queue._put(self, item);
self.all_items.add(item);
else:
self.unfinished_tasks -= 1;
已测试并可与Python 3.4一起使用。