How check if a task is already in python Queue?

user_123927 0
Viewed 12900

I'm writing a simple crawler in Python using the threading and Queue modules. I fetch a page, check links and put them into a queue, when a certain thread has finished processing page, it grabs the next one from the queue. I'm using an array for the pages I've already visited to filter the links I add to the queue, but if there are more than one threads and they get the same links on different pages, they put duplicate links to the queue. So how can I find out whether some url is already in the queue to avoid putting it there again?

queue multithreading python
13 Answers

This is my code. Hopefully, it can help someone.

I added set as a cache to the Queue class. This cache was utilized for task uniqueness checking. Moreover, it was used to implement the __contains__ magic method in the Queue class.

Uniqueness can be defined in two ways. First, tasks are unique in the whole life of the queue. In other words, the queue rejects accepting a repeated task even after the task is done and removed from the queue. I implemented this as "be_unique_in_all_items". Second, tasks are unique only in the existing tasks in the queue. It means the task can be accepted after it is done. I implemented this as "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()

Note: set only can contain hashable objects. For unhashable objects use list instead.

user_130875800
Related