如何使用Python实现多线程队列

阅读更多：Python 教程

介绍..

在本例中，我们将创建一个任务队列，其中包含要执行的所有任务，以及与队列交互以单独处理其元素的线程池。

我们将从问题开始：什么是队列？队列是一种数据结构，是一系列不同元素按非常特定的顺序维护的集合。让我通过采取实际生活例子来解释。

假设您站在杂货店柜台付钱排队（别问我是哪家杂货店）。

在排队等候付账的人群中，您会注意到以下内容：

1.人们从队列的一端进入并从另一端退出。

2.如果A在B之前进入队列，除非B是名人或优先级更高，否则A将在B之前离开队列。

3.一旦每个人都付了账，排队的人就没有了。

好吧，回到编程，队列的工作方式类似。

1. enqueue – 将元素添加到队列的末尾。

2. dequeue – 从队列开头删除元素。

还有更多，先进先出（FIFO）-首先添加的元素将首先被删除。后进先出（LIFO）-添加的最后一个元素将首先被删除。

Python如何实现队列数据结构？

Python中的队列模块提供了队列数据结构的简单实现。每个队列可以具有以下方法。

get（）：返回下一个元素。
put（）：添加一个新元素。
qsize（）：队列中当前元素的数量。
empty（）：返回一个布尔值，指示队列是否为空。
full（）：返回一个布尔值，指示队列是否已满。

1.我们将创建一个函数，它接受一个参数x，然后遍历1到x之间的数字进行乘法。例如，当您将5传递给此函数时，它将遍历1到5并不断相乘，即1乘以5，2乘以5，3乘以5，4乘以5，5乘以5，最后将值作为列表返回。

示例

def print_multiply(x):
    output_value = []
    for i in range(1, x + 1):
        output_value.append(i * x)
    print(f"Output \n *** The multiplication result for the {x} is - {output_value}")

print_multiply(5)

输出

*** The multiplication result for the 5 is - [5, 10, 15, 20, 25]

2.我们将编写另一个名为process_queue（）的函数，该函数将尝试获取队列对象的下一个元素。这很简单的逻辑，不断传递元素，直到队列为空。我将使用sleep稍微延迟一下程序进行。

示例

def process_queue():
    while True:
        try:
            value = my_queue.get(block=False)
        except queue.Empty:
            return
        else:
            print_multiply(value)
            time.sleep(2)

3.创建一个类，初始化并启动新实例时将调用process_queue（）函数。

示例

class MultiThread(threading.Thread):
    def __init__(self, name):
        threading.Thread.__init__(self)
        self.name = name

    def run(self):
        print(f" ** Starting the thread - {self.name}")
        process_queue()
        print(f" ** Completed the thread - {self.name}")

4.最后，我们将传递数字的输入列表并填充队列。

# 设置变量
input_values = [5, 10, 15, 20]

# 填充队列
my_queue = queue.Queue()
for x in input_values:
    my_queue.put(x)

5.最后，将所有内容放在一起。

import queue
import threading
import time

# 类
class MultiThread(threading.Thread):
def __init__(self, name):
threading.Thread.__init__(self)
self.name = name

def run(self):
print(f"输出 \n ** 我们正在运行{self.name}")
process_queue()
print(f" ** 操作已完成 - {self.name}")

# 处理队列
def process_queue():
while True:
try:
value = my_queue.get(block=False) # 非阻塞方式取队列里的值
except queue.Empty:
return
else:
print_multiply(value)
time.sleep(2)

# 自定义函数：求乘积
def print_multiply(x):
output_value = []
for i in range(1, x + 1):
output_value.append(i * x)
print(f" \n *** {x}的乘积是 - {output_value}")

# 输入值
input_values = [2, 4, 6, 5,10,3]

# 将值加入队列
my_queue = queue.Queue()
for x in input_values:
my_queue.put(x)
# 初始化并开启3个线程
thread1 = MultiThread('线程1')
thread2 = MultiThread('线程2')
thread3 = MultiThread('线程3')
thread4 = MultiThread('线程4')

# 开启线程
thread1.start()
thread2.start()
thread3.start()
thread4.start()

# 合并线程
thread1.join()
thread2.join()
thread3.join()
thread4.join()

输出

** 我们正在运行线程1
*** 2的乘积是 - [2, 4]

输出

** 我们正在运行线程2
*** 4的乘积是 - [4, 8, 12, 16]

输出

** 我们正在运行线程3
*** 6的乘积是 - [6, 12, 18, 24, 30, 36]

输出

** 我们正在运行线程4
*** 5的乘积是 - [5, 10, 15, 20, 25]
*** 10的乘积是 - [10, 20, 30, 40, 50, 60, 70, 80, 90, 100]
*** 3的乘积是 - [3, 6, 9] ** 操作已完成 - 线程3
** 操作已完成 - 线程4
** 操作已完成 - 线程2 ** 操作已完成 - 线程1