在 python 中正确使用多线程队列?-Python问题

Using Multithreaded queue in python the correct way?(在 python 中正确使用多线程队列?)

本文介绍了在 python 中正确使用多线程队列?的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！

问题描述

我正在尝试在 python 中使用多线程的队列.我只是想知道我使用的方法是否正确.如果我在做一些多余的事情，或者如果有更好的方法我应该使用.

I am trying to use The Queue in python which will be multithreaded. I just wanted to know the approach I am using is correct or not. And if I am doing something redundant or If there is a better approach that I should use.

我正在尝试从表中获取新请求并使用一些逻辑来安排它们以执行某些操作，例如运行查询.

I am trying to get new requests from a table and schedule them using some logic to perform some operation like running a query.

所以我在这里从主线程为队列生成了一个单独的线程.

So here from the main thread I spawn a separate thread for the queue.

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)

现在在 Setqueue 类中，我有一个 process_queue 函数，用于处理每次运行中添加到队列中的前 2 个请求.

Now in the Setqueue class I have a process_queue function which is used for processing the top 2 requests in every run that were added to the queue.

'''
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)

我想看看我的理解是否正确，是否有任何问题.

I want to see if my understanding is correct and if there can be any issues with it.

所以当主函数中的 True 连接到数据库时运行的主线程获取新请求并将其放入队列中.队列的工作线程(守护进程)不断从队列中获取新请求并派生执行处理的非守护线程，并且由于连接超时为 1，工作线程将继续接受新请求而不会被阻塞，并且它的子线程将继续在后台处理.对吗?

So the main thread running in while True in the main func connects to database gets new requests and puts it in the queue. The worker thread(daemon) for the queue keeps on getting new requests from the queue and fork non-daemon threads which do the processing and since timeout for the join is 1 the worker thread will keep on taking new requests without getting blocked, and its child thread will keep on processing in the background. Correct?

因此，如果主进程退出，这些将在完成工作之前不会被杀死，但工作守护线程将退出.疑问:如果父进程是守护进程，子进程是非守护进程，如果父进程退出，子进程是否退出?).

So in case if the main process exit these won`t be killed until they finish their work but the worker daemon thread would exit. Doubt : If the parent is daemon and child is non daemon and if parent exits does child exit?).

我也在这里阅读:- David beazley 多处理

大卫比兹利在使用池作为线程协处理器部分试图解决类似的问题.所以我应该按照他的步骤:-1. 创建一个进程池.2.像我为request_queue做的那样打开一个线程3.在那个线程中

By david beazley in using a Pool as a Thread Coprocessor section where he is trying to solve a similar problem. So should I follow his steps :- 1. Create a pool of processes. 2. Open a thread like I am doing for request_queue 3. In that thread

  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")

使用池中的进程并并行运行 verify_func.这会给我带来更好的性能吗?

Use a process from the pool and run the verify_func in parallel. Will this give me more performance?

问题描述

推荐答案