Bdtjtk

I test multithreading in python, but the code is suspended:

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

    size = int(math.ceil(16 / 3))



    for e in range(size):

        for i in range(e * size, min((e + 1) * size, 16)):

            q.put([i], block=False)

        threads = 

        for i in range(size):

            threads.append(MongoInsertThread(q, i))

        for t in threads:

            t.start()

        for t in threads:

            t.join()

        print("+++++++++++++++++++++++")

the result is what I wanted, but the program is not ended, the result is:

0 :  [0]

1 :  [1]

2 :  [2]

3 :  [3]

4 :  [4]

5 :  [5]

+++++++++++++++++++++++

0 :  [6]

1 :  [7]

2 :  [8]

3 :  [9]

4 :  [10]

5 :  [11]

+++++++++++++++++++++++

0 :  [12]

1 :  [13]

2 :  [14]

3 :  [15]

But it does not print the last +++++++++++++++++++++++, How can I deal with it?

You are calling q.put() 16 times - i in range(e * 5, (e + 1)*5) for every e in range(5) produces

1. i in range(0, 5)


2. i in range(5, 10)


3. i in range(10, 15)


4. i in range(15, 16)


5. i in range(20, 16)


6. i in range(25, 16)

Which will append 16 values to q. However you create 25 threads with the associated 25 calls to q.get() - but once the first 16 elements have been removed get() blocks. If you use q.get(block=False) the code will finish successfully but you will get a number of warnings raising Empty from get(). Changing the loop condition of the loop

for i in range(size):

  threads.append(MongoInsertThread(q, i))

to range(e * size, min((e + 1) * size, 16))

Will fix the mismatch. However it is also useful to add a break statement to stop the outermost for loop after all 16 elements are added and removed.

Complete Example

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(block=False))



def save_to_mongo_with_thread():

    q = Queue.Queue()

    size = int(math.ceil(16 / 3))

    for e in range(5):

        if (e*size > 16): 

            break

        for i in range(e * size, min((e + 1) * size, 16)):

            q.put([i], block=False)

        threads = 

        for i in range(e * size, min((e + 1) * size, 16)):

            threads.append(MongoInsertThread(q, i))

        for t in threads:

            t.start()

        for t in threads:

            t.join()

        print("+++++++++++++++++++++++")

Output -

0: [0]

1: [1]

2: [2]

3: [3]

4: [4]

+++++++++++++++++++++++

5: [5]

6: [6]

7: [7]

8: [8]

9: [9]

+++++++++++++++++++++++

10: [10]

11: [11]

12: [12]

13: [13]

14: [14]

+++++++++++++++++++++++

15: [15]

+++++++++++++++++++++++

Also note that I am using get(block = False) to help illuminate future issues, but it is unnecessary for this example.