python多线程无法输出一个完整的结果

Question

我使用多线程向数据库中插入数据，但无法返回正确的结果，代码如下:

class MongoInsertThread(threading.Thread):
    def __init__(self,  queue, thread_id):
        super(MongoInsertThread, self).__init__()
        self.thread_id = thread_id
        self.queue = queue

    def run(self):
        print(self.thread_id,': ', self.queue.get())

def save_to_mongo_with_thread():
    q = queue.Queue()

    for e in range(3):
        for i in range(10):
            q.put([i], block=False)
        threads = []
        for i in range(5): ##(1)
            threads.append(MongoInsertThread(q, i))
        for t in threads:
            t.start()
        for t in threads:
            t.join()
        print("+++++++++++++++++++++++")

但是代码生成的结果是:

0 :  [0]
1 :  [1]
2 :  [2]
3 :  [3]
4 :  [4]
+++++++++++++++++++++++
0 :  [5]
1 :  [6]
2 :  [7]
3 :  [8]
4 :  [9]
+++++++++++++++++++++++
0 :  [0]
1 :  [1]
2 :  [2]
3 :  [3]
4 :  [4]
+++++++++++++++++++++++

这不是我想要的，我希望结果是:

0 :  [0]
1 :  [1]
2 :  [2]
3 :  [3]
4 :  [4]
0 :  [5]
1 :  [6]
2 :  [7]
3 :  [8]
4 :  [9]
+++++++++++++++++++++++
0 :  [0]
1 :  [1]
2 :  [2]
3 :  [3]
4 :  [4]
0 :  [5]
1 :  [6]
2 :  [7]
3 :  [8]
4 :  [9]
+++++++++++++++++++++++
0 :  [0]
1 :  [1]
2 :  [2]
3 :  [3]
4 :  [4]
0 :  [5]
1 :  [6]
2 :  [7]
3 :  [8]
4 :  [9]
+++++++++++++++++++++++

也许在 for 循环中出了点问题，但我找不到任何解决方案来处理它。我的代码哪里有问题？我该如何处理呢？而我用11代替了##(1)中的5，挂了，请问如何处理？

Answer 1

使用线程池怎么样？这是我的方法:

• 更改您的 MongoInsertThread.run() 方法，使其永远运行，直到它看到一些返回点(None 工作，例如)。
• 制作一个线程池 MongoInsertThread 线程。
• 更新 save_to_mongo_with_threads:启动线程池 -> 将作业放入队列 -> 停止线程池。

更新:在此解决方案中更多地解释线程池

• 线程池是多个线程的集合。
• 线程池中的线程共享相同的作业队列。
• 每个线程永远运行(MongoInsertThread.run() 方法):
  • (1) 从共享队列中获取一个作业。
  • (2) 如果作业是None -> 中断forever 循环(即停止当前线程)
  • (3) else(作业不是None)-> 处理作业。
  • (4) 转到 (1)。

代码:

import threading
import queue
import time


class MongoInsertThread(threading.Thread):
    def __init__(self, queue, thread_id):
        super(MongoInsertThread, self).__init__()
        self.thread_id = thread_id
        self.queue = queue

    def run(self):
        while True:
            job = self.queue.get()
            if job is None:
                return
            print(self.thread_id, ": ", job)


class ThreadPool:
    def __init__(self, queue, thread_count):
        self._queue = queue
        self._thread_count = thread_count
        self._workers = []
        for thread_id in range(thread_count):
            worker_thrd = MongoInsertThread(queue, thread_id)
            self._workers.append(worker_thrd)

    def start(self):
        for worker_thrd in self._workers:
            worker_thrd.start()

    def stop(self):
        # put None job, each worker thread picks one then stops
        for worker_thrd in self._workers:
            self._queue.put(None)
        # wait for worker threads
        for worker_thrd in self._workers:
            worker_thrd.join()


def save_to_mongo_with_threads():
    q = queue.Queue()

    pool = ThreadPool(q, 5)
    pool.start()

    time.sleep(1.0)

    for e in range(3):
        for i in range(10):
            q.put([e, i])
        print("+++++++++++++++++++++++")

    time.sleep(1.0)

    pool.stop()


save_to_mongo_with_threads()

注意:作业可能在线程之间分布不均。一种可能的输出:

+++++++++++++++++++++++
0 :  [0, 0]
2 :  [0, 1]
4 :  [0, 3]
1 :  [0, 4]
0 :  [0, 5]
2 :  [0, 6]
4 :  [0, 7]
3 :  [0, 2]
+++++++++++++++++++++++
2 :  [0, 9]
0 :  [0, 8]
+++++++++++++++++++++++
2 :  [1, 2]
0 :  [1, 3]
4 :  [1, 0]
4 :  [1, 7]
1 :  [1, 4]
2 :  [1, 5]
4 :  [1, 8]
1 :  [1, 9]
2 :  [2, 0]
4 :  [2, 1]
1 :  [2, 2]
2 :  [2, 3]
4 :  [2, 4]
3 :  [1, 1]
2 :  [2, 6]
4 :  [2, 7]
3 :  [2, 8]
2 :  [2, 9]
1 :  [2, 5]
0 :  [1, 6]