问题描述

我正在尝试在 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?

推荐答案

虽然可以为队列创建一个新的独立线程，并按照您的方式单独处理该数据，但我相信这对于每个队列来说更常见独立的工作线程将消息发布到他们已经知道"的队列；关于.然后通过从该队列中拉出消息，从其他线程处理该队列.

While its possible to create a new independent thread for the queue, and process that data separately the way you are doing it, I believe it is more common for each independent worker thread to post messages to a queue that they already "know" about. Then that queue is processed from some other thread by pulling messages out of that queue.

我设想您的应用程序的方式是三个线程.主线程和两个工作线程.1 个工作线程将从数据库获取请求并将它们放入队列中.另一个工作线程将处理队列中的数据

The way I invision your application would be three threads. The main thread, and two worker threads. 1 worker thread would get requests from the database and put them in the queue. The other worker thread would process that data from the queue

主线程会使用线程函数 .join() 等待其他线程完成

The main thread would just waiting for the other threads to finish by using the thread functions .join()

您将保护线程可以访问的队列并通过使用互斥锁使其线程安全.我在其他语言的许多其他设计中也看到了这种模式.

You would protect queue that the threads have access to and make it thread safe by using a mutex. I have seen this pattern in many other designs in other languages as well.

有效的 Python"布雷特·斯拉特金 (Brett Slatkin) 就这个问题举了一个很好的例子.

"Effective Python" by Brett Slatkin has a great example of this very question.

他没有从 Queue 继承，而是在他的类中为其创建了一个包装器调用 MyQueue 并添加了 get() 和 put(message) 函数.

Instead of inheriting from Queue, he just creates a wrapper to it in his class called MyQueue and adds a get() and put(message) function.

他甚至在他的 Github 仓库中提供了源代码

He even provides the source code at his Github repo

https://github.com/bslatkin/effectivepython/blob/master/example_code/item_39.py

我不隶属于这本书或其作者，但我强烈推荐它，因为我从中学到了很多东西:)

I'm not affiliated with the book or its author, but I highly recommend it as I learned quite a few things from it :)

这篇关于在 python 中正确使用多线程队列?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持跟版网！