xiaoxinks

一、前置知识

1. 队列基础

  • 如果不指定队列是什么,请自行查阅
  • 在Python中,队列是最常用的线程间的通信方法,因为它是线程安全的
from queue import Queue

# 创建队列
#   -- 限制队中最多有 maxsize 个元素
#   -- 如果省略参数,默认元素个数无限制
q = Queue(100)
q1 = Queue()

# 元素入队
q.put(1)
q.put(True)
q.put('abc')

# 队列的大小
print(q.qsize())

# 判断队满
print(q.full())

# 判断队空
print(q.empty())

# 元素出队
#	注意:如果队空,取元素时,会陷入阻塞状态,知道再往队中加入数据为止【***】
while not q.empty():
    print(q.get())

2. 多线程

(1) 进程与线程的关系

'''
    1. 一个进程可以有多个线程,但是必须有一个主线程
    
    2. 进程之间互不影响,资源不共享
    
    3. 线程之间,资源可以共享,共享的是线程所属的进程的内容
    
    4. 线程必须依赖于进程存在
'''

(2) 创建线程——方式1

from threading import Thread

def add(n1, n2):
    print('结果为:' + n1 + n2)

def main():
    # 创建一个线程
    #   -- target 函数的名称
    #   -- args 以元组的形式,传入函数所需的参数
    t = Thread(target=add, args=(1, 2,))
    # 开启线程
    t.start()

if __name__ == '__main__':
    main()

(3) 创建线程——方式2

'''
    1. 通过继承 Thread类 创建线程的步骤
        (1) 定义一个类
        (2) 继承 Thread类
        (3) 重写 run() 方法
        (4) 在 run() 方法中写逻辑代码

    2. 注意事项
        (1) 子类继承 Thread类 后,实例化对象时,会自动执行父类中的 run()方法
            所以我们可以重写 run(),然后在 run() 中执行我们自己的代码
        (2) 一个子类继承了 Thread类,那么在对线程执行任何其他操作之前
            它必须确保已调用基类的构造函数
            -- 比如:传参时,需要调用的父类的构造函数
'''
from threading import Thread

class MyThread(Thread):
    # 构造函数
    def __init__(self, n1, n2):
        # 调用父类的构造函数:第一种方法
        # threading.Thread.__init__(self)
        # 调用父类的构造函数:第二种方法
        super().__init__()
        self.n1 = n1
        self.n2 = n2
    # 重写 run() 方法
    def run(self):
        print('线程的名称:' + self.name)
        print(self.n1 + self.n2)

def main():
    # 实例化对象的过程,就是在创建线程
    t1 = MyThread(1, 1)
    # 设置线程的名称
    t1.setName('t1')
    # 开启线程
    t1.start()

if __name__ == '__main__':
    main()

(4) 锁的使用

  • 一定要保证相关的线程使用的是同一把锁,否则加锁操作无意义
# 加锁之前
# ----------------------------------------------------------
from threading import Thread

num = 0 # 声明共享资源

def Jia():
    # 标注使用共享的资源
    global num
    # 主逻辑代码
    for i in range(10000000):
        num+=1
    print(num)

def main():
    # 创建线程
    t1 = Thread(target=Jia)
    t2 = Thread(target=Jia)
    t3 = Thread(target=Jia)
    # 开启线程
    t1.start()
    t2.start()
    t3.start()

if __name__ == '__main__':
    main()

# 加锁之后
# ----------------------------------------------------------
from threading import Thread
from threading import Lock

lock = Lock() # 声明锁,要保证相关的线程使用的是同一把锁
num = 0 # 声明共享资源

def Jia(lock):
    # 加锁
    lock.acquire()
    # 标注使用共享的资源
    global num
    # 主逻辑代码
    for i in range(10000000):
        num+=1
    print(num)
    # 释放锁
    lock.release()

def main():
    # 创建线程
    t1 = Thread(target=Jia, args=(lock,))
    t2 = Thread(target=Jia, args=(lock,))
    t3 = Thread(target=Jia, args=(lock,))
    # 开启线程
    t1.start()
    t2.start()
    t3.start()

if __name__ == '__main__':
    main()

3. 进阶

(1) Thread.join()

  • 作用:阻塞当前所在的线程,只有当执行 join() 的线程结束之后,才会解除阻塞
  • 分析下面的代码:
    • 阻塞前:在主线程中有一句print('结束了'),本意是想要在fn函数执行完之后,再输出结束了,但是因为主线程和t1线程是同步的,他们在同时执行,所以print('结束了')的输出位置不一定是最后面,可能是在fn执行一半的时候就输出结束了
    • 阻塞后:t1线程调用了join(),阻塞了当前所在线程,即阻塞了主线程,所以主线程需要等t1线程结束后才可以继续执行主线程的内容,故实现了print('结束了')fn执行完后在输出内容的需求
# 阻塞前:也就是不调用 join()
# ----------------------------------------------------------
import time
from threading import Thread

def fn():
    for i in range(10):
        print(i)
    time.sleep(1.5)

def main():
    t1 = Thread(target=fn)
    t1.start()
    print('结束了')

if __name__ == '__main__':
    main()
    
# 阻塞后:调用了 join()
# ----------------------------------------------------------
import time
from threading import Thread

def fn():
    for i in range(10):
        print(i)
    time.sleep(1.5)

def main():
    t1 = Thread(target=fn)
    t1.start()
    t1.join()
    print('结束了')

if __name__ == '__main__':
    main()

(2) 守护进程

'''
    1. 进程分为主进程、守护进程、非守护进程

    2. 守护、非守护是相对于主进程 而言的

    3. 守护进程,可以理解为不重要的进程,当主进程结束后,守护进程会强制结束

    4. 非守护进程,是比守护进程重要的进程,当主进程结束后,守护进程不会被强制结束
'''
# t1进程是非守护进程:t1进程会陷入死循环
# ----------------------------------------------------------
from threading import Thread

def fn():
    while True:
        print(1)

def main():
    t1 = Thread(target=fn)
    t1.start()
    print('结束了')

if __name__ == '__main__':
    main()
# t1进程是守护进程:t1进程会因为主进程的结束,被强制结束
# ----------------------------------------------------------
from threading import Thread

def fn():
    while True:
        print(1)

def main():
    t1 = Thread(target=fn)
    t1.start()
    t1.setDaemon(True) # 设置为True时,说明此进程是"守护进程"【默认是False】
    print('结束了')

if __name__ == '__main__':
    main()

(3) 队列在线程之间的通信

  • 一定要在后面的代码中仔细思考一下,尤其是阶段5的代码
# Queue.join()
'''
	当生产者生产结束时,先阻塞生产者线程,只有当消费者发出已经消费完队中产品时,才解除阻塞
'''

# Queue.task_done()
'''
	消费者消费一个队中的产品,就向生产者发送一次信息
	当消费完队中信息之后,也向生产者发送信息,并发出已经消费完的提示,提示生产者可以解除生产者线程的阻塞了
'''

二、生产者与消费者模式

  • 该模式有两个对象,分别是生产者、消费者,两者同时进行操作
  • 下面分为5个阶段,慢慢讲解

阶段1:消费者线程的阻塞

from queue import Queue
from threading import Thread

# 生产者
def produce(q):
    for i in range(1, 11):
        q.put(i)
        print(f'生产产品——{i}')

# 消费者
def consumer(q):
    while True:
        tmp = q.get()
        print(f'消费产品——{tmp}')

# 主进程
def main():
    q = Queue()
    pro = Thread(target=produce,  args=(q,))
    con = Thread(target=consumer, args=(q,))
    pro.start()
    con.start()

if __name__ == '__main__':
    main()
  • 分析
    • 在主线程中创建并开启生产者线程和消费者线程,生产者共生产10个产品
    • 生产者生产产品的同时,消费者在调用q.get()方法消费产品,当生产者把产品全部生产完之后,生产者线程结束,消费者继续调用q.get()方法消费产品,当没有产品可以消费时,消费者再调用q.get()时,会导致消费者线程进入阻塞状态,直到再往里面加数据为止,但是生产者已经把产品生产完,不会再生产了,所以消费者线程会一直处于阻塞状态
    • image-20220113183844119

阶段2:产品消费不完

from queue import Queue
from threading import Thread

# 生产者
def produce(q):
    for i in range(1, 11):
        q.put(i)
        print(f'生产产品——{i}')

# 消费者
def consumer(q):
    while True:
        tmp = q.get()
        print(f'消费产品——{tmp}')

# 主进程
def main():
    q = Queue()
    pro = Thread(target=produce,  args=(q,))
    con = Thread(target=consumer, args=(q,))
    con.setDaemon(True) # 设置守护线程
    pro.start()
    con.start()

if __name__ == '__main__':
    main()
  • 针对阶段1的代码,只添加了一行代码,将消费者线程为 "守护线程"即可
  • 分析
    • 当生产者将产品全部生产完,生产者线程结束,然后主线程也结束了,接着消费者线程作为守护线程被强制退出,解决了消费者线程阻塞的问题
    • 但是,由下图可看到,虽然解决了消费者线程阻塞的问题,但是消费者本次只消费了5个产品,生产者所生产的产品没有被消费完,这个问题请看阶段3
    • image-20220113152402321

阶段3:小完美的代码

from queue import Queue
from threading import Thread

# 生产者
def produce(q):
    for i in range(1, 11):
        q.put(i)
        print(f'生产产品——{i}')
    q.join() # 阻塞生产者线程,只有接收到消费者发送来的已经消费了最后一个产品的时候,才解除阻塞

# 消费者
def consumer(q):
    while True:
        tmp = q.get()
        print(f'消费产品——{tmp}')
        q.task_done() # 向生产者发送消息,告诉生产者我已经消费了一个产品

# 主进程
def main():
    q = Queue()
    pro = Thread(target=produce,  args=(q,))
    con = Thread(target=consumer, args=(q,))
    con.setDaemon(True)
    pro.start()
    con.start()

if __name__ == '__main__':
    main()
  • 针对阶段2仅添加了两行代码

    • q.join()
    • q.task_done()
  • 分析:

    • 当生产者将产品全部生产完,生产者线程因为执行了q.join()而被阻塞,只有接收到消费者发送来的已经消费了最后一个产品的时候,才解除阻塞
    • 而消费者线程会边消费产品,边执行q.task_done()给生产者线程发送消息,直到消费完全部的产品时,在给生产者发送消息时,会通知生产者已经消费完全部的产品
    • 此时生产者接收到消费完全部产品的信息,阻塞被解除,生产者线程结束
    • 然后主线程结束
    • 再接着,由于消费者线程的守护线程,被强制关闭
    • image-20220113185921120

阶段4:有关线程执行顺序的问题

from queue import Queue
from threading import Thread

# 生产者
def produce(q):
    for i in range(1, 11):
        q.put(i)
        print(f'生产产品——{i}')
    q.join() # 阻塞生产者线程,只有接收到消费者发送来的已经消费了最后一个产品的时候,才解除阻塞

# 消费者
def consumer(q):
    while True:
        tmp = q.get()
        print(f'消费产品——{tmp}')
        q.task_done() # 向生产者发送消息,告诉生产者我已经消费了一个产品

# 主进程
def main():
    q = Queue()
    pro = Thread(target=produce,  args=(q,))
    con = Thread(target=consumer, args=(q,))
    con.setDaemon(True)
    pro.start()
    con.start()
    print('结束了')

if __name__ == '__main__':
    main()
  • 与阶段3相比,仅在主线程中添加一行输出语句
  • 分析
    • 我们想要的是两个子线程结束之后,再打印输出生产者和消费者全部结束了呀!!!,但是很明显,结果不是这样的,下面开始分析
    • 程序中有1个主线程、2个子线程,三者会同时执行,所以主线程中的输出语句的执行时间是随机的,故输出的位置也是随机的
    • 解决方法:阻塞当前线程,也就是阻塞主线程,见阶段5
    • image-20220113190227763

阶段5:线程执行顺序问题的解决

from queue import Queue
from threading import Thread

# 生产者
def produce(q):
    for i in range(1, 11):
        q.put(i)
        print(f'生产产品——{i}')
    q.join() # 阻塞生产者线程,只有接收到消费者发送来的已经消费了最后一个产品的时候,才解除阻塞

# 消费者
def consumer(q):
    while True:
        tmp = q.get()
        print(f'消费产品——{tmp}')
        q.task_done() # 向生产者发送消息,告诉生产者我已经消费了一个产品

# 主进程
def main():
    q = Queue()
    pro = Thread(target=produce,  args=(q,))
    con = Thread(target=consumer, args=(q,))
    con.setDaemon(True)
    pro.start()
    con.start()
    pro.join() # 阻塞当前所在的线程
    print('结束了')

if __name__ == '__main__':
    main()
  • 与阶段4相比,仅添加一句代码,以达到阻塞主线程的需求
  • 分析:
    • 程序中有1个主线程、2个子线程,三者会同时执行
    • 主线程中执行到pro.join()时,当前线程被阻塞,也即主线程被阻塞,知道生产完全部产品,消费完全部产品,生产者线程结束
    • 主线程才被解除阻塞
    • 然后主线程结束,消费者线程被强制结束
    • image-20220113190744707

三、参考

相关文章: