¿Cómo verificar si una tarea ya está en cola de Python?

Question

Escribo un rastreador simple en Python usando los módulos de subprocesamiento y cola. Busco una página, reviso los enlaces y los pongo en una cola, cuando un determinado hilo ha terminado de procesar la página, toma el siguiente de la cola. Estoy usando una matriz para las páginas que ya he visitado para filtrar los enlaces que agrego a la cola, pero si hay más de un subproceso y obtienen los mismos enlaces en páginas diferentes, ponen enlaces duplicados a la cola. Entonces, ¿cómo puedo averiguar si alguna url ya está en la cola para evitar ponerlo de nuevo allí?

Answer 1

Si no importa el orden en que se procesan los artículos, me gustaría probar una subclase de Queue que utiliza internamente 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() 

Como señaló Paul McGuire, esto permitiría agregar un elemento duplicado una vez que se ha eliminado del conjunto "por procesar" y aún no se ha agregado al conjunto "procesado". Para solucionar esto, puede almacenar ambos conjuntos en la instancia Queue, pero como está utilizando el conjunto más grande para comprobar si el elemento se ha procesado, puede volver al queue, que ordenará las solicitudes correctamente.

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) 

La ventaja de esto, en lugar de utilizar un conjunto separado, es que los métodos de la Queue 's son seguros para subprocesos, por lo que no necesita de bloqueo adicional para el control de la otra serie.

Answer 2

La forma en que resolví esto (en realidad lo hice en Scala, no en Python) fue utilizar un conjunto y una cola, solo agregando enlaces a la cola (y establecer) si no existían en el conjunto.

Tanto el conjunto como la cola se encapsularon en un único subproceso, lo que expone solo una interfaz tipo cola a los subprocesos del consumidor.

Editar: alguien más sugirió SQLite y eso también es algo que estoy considerando, si el conjunto de URL visitadas necesita crecer. (Actualmente, cada rastreo tiene solo unos cientos de páginas, por lo que cabe fácilmente en la memoria). Pero la base de datos es algo que también se puede encapsular dentro del propio conjunto, por lo que los hilos del consumidor no tienen que ser conscientes de ello.

Answer 3

SQLite es tan simple de usar y cabría perfectamente ... solo una sugerencia.

Answer 4

Además, en lugar de un conjunto puede intentar usar un diccionario. Las operaciones en conjuntos tienden a ser bastante lentas cuando son grandes, mientras que una búsqueda en el diccionario es agradable y rápida.

Mi 2c.

Answer 5

uso:

url in q.queue 

que devuelve Verdadero si y sólo si es url en la cola

Answer 6

¿Por qué sólo usar la matriz (idealmente, un diccionario sería aún mejor) para filtrar las cosas que ya has visitado? Agregue elementos a su matriz/diccionario tan pronto como los ponga en cola y solo agréguelos a la cola si aún no están en la matriz/dict. Entonces usted tiene 3 simples cosas separadas:

1. enlaces aún no se ve (ni en la cola ni matriz/dict)
2. Enlaces programados para ser visitado (tanto en la cola y la matriz/dict)
3. enlaces ya visitados (en orden de batalla/dict, no en la cola)

Answer 7

en lugar de "conjunto de páginas que ya visitó" crea un "conjunto de páginas ya se agrega a la cola"

Answer 8

El método put también necesita ser sobrescrito, si no una unión llamada se bloqueará siempre 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() 

Answer 9

Lamentablemente, no tengo ninguna calificación de enouch para comentar la mejor respuesta de Lukáš Lalinský.

para añadir soporte para SetQueue.task_done() y SetQueue.join() para la segunda variante de SetQueue de Lukáš Lalinský añadir otra brahch al caso:

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; 

probado y funciona con Python 3.4.

Answer 10

Esta es la versión completa de 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 

Answer 11

Lo que sigue es una mejora sobre Lukáš de este último solution Lalinský. La diferencia importante es que se put anulados a fin de garantizar unfinished_tasks es precisa y join funciona como se espera.

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) 

Answer 12

estoy de acuerdo con @Ben James.Try utilizar tanto deque y ajuste.

aquí código:

class SetUniqueQueue(Queue): 

    def _init(self, maxsize): 
     self.queue = deque() 
     self.setqueue = set() 

    def _put(self, item): 
     if item not in self.setqueue: 
      self.setqueue.add(item) 
      self.queue.append(item) 

    def _get(self): 
     return self.queue.popleft()