Estoy tratando de usar The Queue en python que será multiproceso. Solo quería saber si el enfoque que estoy usando es correcto o no. Y si estoy haciendo algo redundante o si hay un mejor enfoque que debería usar.

Estoy tratando de obtener nuevas solicitudes de una tabla y programarlas usando cierta lógica para realizar alguna operación, como ejecutar una consulta.

Así que aquí, desde el hilo principal, engendro un hilo separado para la cola.

if __name__=='__main__':

  request_queue = SetQueue(maxsize=-1)
  worker = Thread(target=request_queue.process_queue)
  worker.setDaemon(True)
  worker.start()


  while True:
    try:
      #Connect to the database get all the new requests to be verified
      db = Database(username_testschema, password_testschema, mother_host_testschema, mother_port_testschema, mother_sid_testschema, 0)
      #Get new requests for verification
      verify_these = db.query("SELECT JOB_ID FROM %s.table WHERE     JOB_STATUS='%s' ORDER BY JOB_ID" %
                             (username_testschema, 'INITIATED'))

      #If there are some requests to be verified, put them in the queue.
      if len(verify_these) > 0:
        for row in verify_these:
          print "verifying : %s" % row[0]
          verify_id = row[0]
          request_queue.put(verify_id)
    except Exception as e:
      logger.exception(e)
    finally:
      time.sleep(10)

Ahora en la clase Setqueue tengo una función process_queue que se utiliza para procesar las 2 solicitudes principales en cada ejecución que se agregaron a la cola.

'''
Overridding the Queue class to use set as all_items instead of list to ensure unique items added and processed all the time,
'''

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)

  '''
  The Multi threaded queue for verification process. Take the top two items, verifies them in a separate thread and sleeps for 10 sec.
  This way max two requests per run will be processed.
  '''
  def process_queue(self):
    while True:
      scheduler_obj = Scheduler()

      try:
        if self.qsize() > 0:
          for i in range(2):
            job_id = self.get()
            t = Thread(target=scheduler_obj.verify_func, args=(job_id,))
            t.start()

          for i in range(2):
            t.join(timeout=1)
            self.task_done()

      except Exception as e:
        logger.exception(
          "QUEUE EXCEPTION : Exception occured while processing requests in the VERIFICATION QUEUE")
      finally:
        time.sleep(10)

Quiero ver si mi comprensión es correcta y si puede haber algún problema con ella.

Entonces, el subproceso principal que se ejecuta mientras True en el func principal se conecta a la base de datos recibe nuevas solicitudes y lo pone en la cola. El subproceso de trabajo (daemon) para la cola sigue recibiendo nuevas solicitudes de los subprocesos de cola y bifurcación que no son daemon que realizan el procesamiento y, dado que el tiempo de espera para la unión es 1, el subproceso de trabajo seguirá tomando nuevas solicitudes sin bloquearse, y su el subproceso secundario seguirá procesándose en segundo plano. ¿Correcto?

Entonces, en caso de que el proceso principal salga, estos no se eliminarán hasta que terminen su trabajo, pero el hilo del demonio de trabajo se cerrará. Duda: si el padre es daemon y el niño no es daemon y si el padre sale, ¿sale el niño?).

También leo aquí: -David beazley multiprocesamiento

Por David Beazley al usar un Pool como una sección de Coprocesador de subprocesos donde está tratando de resolver un problema similar. Entonces, debo seguir sus pasos: 1. Crear un grupo de procesos. 2. Abra un hilo como lo estoy haciendo para request_queue 3. En ese hilo

  def process_verification_queue(self):
    while True:
      try:
        if self.qsize() > 0:
          job_id = self.get()
          pool.apply_async(Scheduler.verify_func, args=(job_id,))
      except Exception as e:
        logger.exception("QUEUE EXCEPTION : Exception occured while    processing requests in the VERIFICATION QUEUE")

Utilice un proceso del grupo y ejecute generate_func en paralelo. ¿Esto me dará más rendimiento?