29. 多线程编程

一、什么是线程

线程（thread）它们是同一个进程下执行的，并共享相同的下上文。线程包括开始、执行顺序和结束三部分。它有一个指令指针，用于记录当前运行的上下文。当其它线程运行时，它可以被抢占（中断）和临时挂起（也称为睡眠）—— 这种做法叫做让步（yielding）。

当一个程序运行时，默认有一个线程，这个线程我们称之为 主线程。多任务也就可以理解为让你的代码在运行过程中额外创建一些线程，让这些线程去执行代码。

多线程的执行顺序是不确定的，这是因为执行代码的时候，当前的运行环境可能不同以及资源的分配可能不同，导致操作系统在计算接下来应该调用哪个程序的时候得到了不一样的答案，因此顺序不确定；

二、线程的生命周期

要想实现多线程，必须在主线程中创建新的线程对象。Python 中使用 threading 模块或者 Thread 子类来表示线程，在它的一个完整的生命周期中通常要经过如下的五种状态：

创建：当一个 Thread 类或及其子类的对象被声明并创建时，新生的线程就处于新建状态；
就绪：处于新建的线程被 start() 后，将进入线程队列等待 CPU 时间片，此时它已具备了运行的条件，只是没分配到 CPU 资源；
运行：当就绪的线程被调度并获得 CPU 资源时，便进入运行状态，run() 方法定义了线程的操作和功能；
阻塞：在某种特殊情况下，被人为挂起或执行输入输出操作时，让出 CPU 并临时中止自己的执行，进入阻塞状态；
退出：线程完成了它的全部或线程被提前强制性中止或出现异常导致结束；

三、线程的创建

【1】、使用 threading 模块

如果我们想要执行一个单独的任务，那么就需要创建一个新的线程。如果我们想在一个程序中有多个任务一起运行，那么就需要在程序中创建多个 Thread 对象即可。

在 Python 中，我们可以使用 threading 模块中的 Thread 类创建一个对象。这个对象表示一个线程，但它不会真正创建出来一个线程。而当我们调用 start() 方法时，才会真正创建一个新的子线程，并开始执行的。至于这个线程去执行哪里的代码，要看在用 Thread 创建对象的时候给 target 传递的是哪个函数的引用，即将来线程就会执行 target 参数指向的那个函数。target 指向的那个函数代码执行完之后，意味着这个子线程结束；

创建 Thread 对象时，target 参数指明线程将来去哪里执行代码，而 args 参数执行线程去执行代码时所携带的数据，并且 args 参数是一个元组。如果我们想给指定的参数传递数据，我们可以给 kwargs 参数传递一个字典。

代码执行到最后，虽然主线程没有了代码，但是它依然会等待所有的子线程结束之后，它才会真正的结束，原因是：主线程有个特殊的功能，用来对子线程产生的垃圾进行回收处理。当主线程结束之后，才意味着整个程序真正的结束；

import time

# 1.导入threading模块
from threading import Thread

def task(name):
    print(f"{name}开始执行")
    time.sleep(3)
    print(f"{name}执行结束")

# 2.使用threading模块中Thread创建一个对象
t1 = Thread(target=task, args=("线程1",))
t2 = Thread(target=task, kwargs={"name": "线程2"})
# 3.调用这个实例对象的start()方法让这个线程开始执行
t1.start()
t2.start()

print("主线程执行了！")

一个程序中，可以有多个线程，执行相同的代码。但是，每个线程执行功能每个线程的功能，互不影响，仅仅是做的事情相同一样而已；

【2】、自定义类继承 Thread

我们可以自定义一个类继承 Thread，然后一定要实现它的 run() 方法，即定义一个 run() 方法，并且在方法中实现要执行的代码。当我们调用自己编写的类创建出来的对象的 start() 方法时，会创建新的线程，并且线程会自动调用 run() 方法开始执行。

如果除了 run() 方法之外还定义了很多其它的方法，那么这些方法需要在 run() 方法中自己去第调用，线程它不会自动调用。

import time

from threading import Thread

class MyThread(Thread):

    def __init__(self, name):
        super().__init__()
        self.name = name

    def run(self):
        print(f"{self.name}开始执行")
        time.sleep(1)
        print(f"{self.name}执行结束")


t= MyThread("线程1")
t.start()
print("主线程执行")

四、线程的常用属性和方法

threading.Thread.name                   # 当前线程实例别名，默认为Thread-N，N从1开始递增的整数
threading.enumerate()                   # 当前程序正在运行的线程
threading.current_thread()              # 获取当前线程
threading.Thread.start()                # 启动线程实例
threading.Thread.run()                  # 如果没有给定target参数，对这个对象调用start()方法时，就会执行对象中的run()方法
threading.Thread.is_alive()             # 判断线程实例是否还存活
threading.Thread.join([timeout])        # 在线程a中调用线程b的join()，此时线程a进入阻塞状态，直到线程b完全执行以后，线程a才结束阻塞状态

import time
import threading

money = 100

def task(n):
    print(f"{threading.current_thread().name}开始执行")
    global money
    money *= n
    time.sleep(n)
    print(f"{threading.current_thread().name}的money: {money}")
    print(f"{threading.current_thread().name}执行结束")


## 1、实例化对象
t1 = threading.Thread(target=task, args=(1,))
t2 = threading.Thread(target=task, args=(2,))
t3 = threading.Thread(target=task, args=(3,))

print(f"当前程序中正在运行的线程：{threading.enumerate()}")

start_time = time.time()

# 2、开启线程
t1.start()                      # 告诉操作系统帮你创建一个进程
t2.start()
t3.start()

print(f"当前程序中正在运行的线程：{threading.enumerate()}")

print(t2.is_alive())            # 获取线程状态
print(threading.active_count())           # 统计当前活跃的线程数

# 主线程等待子线程运行结束之后在继续往后执行
t3.join()

print(f"{threading.current_thread().name} {time.time() - start_time}")
print(f"{threading.current_thread().name} money: {money}")

五、守护线程

5.1、什么是守护线程

守护线程，专门用于服务其他的线程。当所有非守护线程结束时，没有了被守护者，守护线程也就没有工作可做，当然也就没有继续执行的必要了，程序就会终止，同时会杀死所有的"守护线程"，也就是说只要有任何非守护线程还在运行，程序就不会终止

在一个含有线程的python程序中，当主线程的代码运行完之后，如果还有其他子线程还未执行完毕，那么主线程会等待子线程执行完毕之后，再结束；如果有一个线程必须设置为无限循环，那么该线程不结束，意味着整个python程序就不能结束，那为了能够让python程序正常退出，将这类无限循环的线程设置为 守护线程 ，当程序当中仅仅剩下守护线程时，python程序就能够正常退出，不必关心这类线程是否执行完毕，这就是守护线程的意义。

import time

from threading import Thread

def task(name,n):
    print(f"{name}开始执行")
    time.sleep(n)
    print(f"{name}执行结束")

if __name__ == "__main__":
    t = Thread(target=task, args=("线程1", 3))
    t.start()
    print("主线程执行")

5.2、设置守护线程的方式

如果要设置守护线程，必须在线程启动前（调用start（））之前进行设置。

【1】、创建线程对象时，将 daemon=True 作为关键字参数传入

import time

from threading import Thread

def task(name,n):
    print(f"{name}开始执行")
    time.sleep(n)
    print(f"{name}执行结束")

if __name__ == "__main__":
    t = Thread(target=task, args=("守护线程",3), daemon=True)    # 1、实例化对象
    t.start()                                                   # 2、开启线程，告诉操作系统帮你创建一个进程
    print("主线程执行")

【2】、将线程对象的 daemon 属性设置为 True

import time

from threading import Thread

def task(name,n):
    print(f"{name}开始执行")
    time.sleep(n)
    print(f"{name}执行结束")

if __name__ == "__main__":
    t = Thread(target=task, args=("守护线程",3))     # 1、实例化对象
    t.daemon = True                                 # 2、将进程设置为守护进程
    t.start()                                       # 3、开启线程，告诉操作系统帮你创建一个进程
    print("主线程执行")

线程会继承当前线程的 daemon 的值，如果当前线程为守护线程，那么在该线程中新建的线程默认为守护线程；

六、线程间通信

如果我们想让多个线程间共享数据，可以通过队列来实现。队列（Queue）是具有一定约束的线性表，它只能在 一端插入 （入队，AddQ）而在 另一端删除 （出队，DeleteQ）。它具有 先进先出 （FIFO）的特性。，它的常用方法如下：

queue.Queue([maxsize])                            # 生成队列，最大可以存放maxsize数据量，默认值为32767
queue.Queue.qsize()                               # 返回当前队列包含的消息数量
queue.Queue.put(item, block=True, timeout=None)   # 向队列中存取数据，默认情况下，如果队列已满，还要放数据，程序会阻塞，直到有位置让出来，不会报错
queue.Queue.put_nowait(obj)                       # 向队列中存取数据，如果队列已满，还要放数据，程序会抛出异常
queue.Queue.get(block=True, timeout=None)         # 取队列中的数据，默认情况下，如果队列中没有数据，还要取数据，程序会阻塞，直到有新的数据到来，不会报错
queue.Queue.get_nowait()                          # 取队列中的数据，如果队列中没有数据，还要取数据，程序会抛出异常
queue.Queue.empty()                               # 如果队列为空，返回True，反之返回False
queue.Queue.full()                                # 如果队列满了，返回True，反之返回False

from queue import Queue

names = ["Sakura","Mikoto","Shana","Akame","Kurome"]

q = Queue(3)

print("向队列中存储数据")
i = 0
while not q.full():
    q.put(names[i])
    i += 1

# 如果消息队列已满，如果还要向队列中存储数据，程序会阻塞或抛出异常
try:
    # 如果没有设置timeout，向已满队列存储数据会阻塞，直到有位置让出来
    # 如果设置timeout，则会等待timeout秒，如果在此期间还没有位置空出来，程序会抛出异常
    q.put(names[i],timeout=3)
except Exception:
    print("队列已满，现有消息数量：%s" % q.qsize())

try:
    # 向已满队列存储数据会抛出异常
    q.put_nowait(names[i+1])
except Exception:
    print("队列已满，现有消息数量：%s" % q.qsize())

print("从队列中读取数据")
while not q.empty():
    data = q.get()
    print(f"读取的数据为{data}")

# 如果消息队列已空，如果还要从队列中读取数据，程序会阻塞或抛出异常
try:
    # 如果没有设置timeout，向已满队列存储数据会阻塞，直到有位置让出来
    # 如果设置timeout，则会等待timeout秒，如果在此期间还没有位置空出来，程序会抛出异常
    q.get(timeout=3)
except Exception:
    print("队列已空，现有消息数量：%s" % q.qsize())

try:
    # 向已满队列存储数据会抛出异常
    q.get_nowait()
except Exception:
    print("队列已空，现有消息数量：%s" % q.qsize())

from threading  import Thread 
from queue import Queue

def produces(q):
    q.put("hello world!")

def consumer(q):
    print(q.get())

if __name__ == "__main__":
    q = Queue(3)

    p1 = Thread(target=produces,args=(q,))
    p1.start()
  
    p2 = Thread(target=consumer,args=(q,))
    p2.start()

七、互斥锁

多个线程操作同一份数据时，可能会出现数据错乱的问题。例如，有 3 个线程，其中线程 1 和线程 2 修改全局变量，线程 3 获取全局变量的值。可能会出现第线程 1 刚刚将数据存放到了全局变量中，本意是想让线程 3 获取它的数据，但是因为操作系统的调度原因导致线程 3 没有被调度，而线程 2 被调度了，恰巧线程 2 也对全局变量进行了修改。而当线程 3 去读取数据时，读取到的是线程 2 修改的数据，而不是线程修改的数据。

针对上述问题，解决方式就是加锁处理：将并发变成串行，牺牲效率但保证了数据的安全。

某个线程要更改共享数据时，先将其锁定，此时资源的状态为 “锁定” ，其它线程不能更改；直到该线程释放资源，将资源的状态变成 “非锁定”，其它的线程才能再次锁定该资源。互斥锁保证了每次只有一个线程进行写入操作，从而保证了多线程情况下数据的正确性。

import time

from threading import Thread,Lock

ticket = 100
mutex = Lock()                                  # 创建一个互斥锁对象

def task(name):
    while True:
        global ticket
        if ticket > 0:
            buy(name)
        else:
            break

def buy(name):
    mutex.acquire()                             # 加锁
    global ticket
    if ticket > 0:
        time.sleep(0.1)
        print(f"{name}卖票，票号为：{ticket}")
        ticket -= 1
    mutex.release()                             # 释放锁

if __name__ == "__main__":
    t1 = Thread(target=task,args=("窗口1",))
    t2 = Thread(target=task,args=("窗口2",))
    t3 = Thread(target=task,args=("窗口3",))

    t1.start()
    t2.start()
    t3.start()

不知道为什么大部分都是只有一个窗口卖票，但是多运行几次或把 ticket 改大一些会发现其它窗口也卖票；

八、死锁问题

不同的线程分别占用对方需要的同步资源不放弃，都在等待对方放弃自己需要的同步资源，就形成了线程的死锁；出现死锁后，不会出现异常，不会出现提示，只是所有的线程都处于阻塞状态，无法继续；

import time

from threading import Thread,Lock

mutexA = Lock()
mutexB = Lock()

class MyThread(Thread):
  
    def run(self):
        self.fun1()
        self.fun2()

    def fun1(self):
        mutexA.acquire()
        print(f"{self.name}抢到A锁")
        time.sleep(3)
        mutexB.acquire()
        print(f"{self.name}抢到B锁")
        mutexB.release()
        mutexA.release()
  
    def fun2(self):
        mutexB.acquire()
        print(f"{self.name}抢到B锁")
        time.sleep(3)
        mutexA.acquire()
        print(f"{self.name}抢到A锁")
        mutexA.release()
        mutexB.release()

if __name__ == "__main__":
    for i in range(10):
        t  = MyThread()
        t.start()

九、什么是递归锁

递归锁可以被连续的 acquire 和 realease，但是只能被第一个抢到这把锁的对象执行上述操作。递归锁的内部有一个计数器，每 acquire 一次计数加 1，每 realease 一次计数减 1，只要计数不为 0，那么其它人都无法抢到这个锁。

import time

from threading import Thread,RLock

mutexA = mutexB = RLock()

class MyThread(Thread):
  
    def run(self):
        self.fun1()
        self.fun2()

    def fun1(self):
        mutexA.acquire()
        print(f"{self.name}抢到A锁")
        mutexB.acquire()
        print(f"{self.name}抢到B锁")
        mutexB.release()
        mutexA.release()
  
    def fun2(self):
        mutexB.acquire()
        print(f"{self.name}抢到B锁")
        time.sleep(3)
        mutexA.acquire()
        print(f"{self.name}抢到A锁")
        mutexA.release()
        mutexB.release()

if __name__ == "__main__":
    for i in range(10):
        t  = MyThread()
        t.start()

十、线程池

池是用来保证计算机硬件安全的情况下最大限度的利用计算机，它降低了程序的运行效率，但是保证了计算机硬件的安全，从而让你写的程序能够正常运行。

初始化 Pool 时，可以指定一个最大线程数，当有新的请求提交到 Pool 时，如果池还没有满，那么就会创建一个新的线程用来执行该请求。但是如果池中的线程数已经达到指定的最大值，那么该请求就会等待，直到池中有线程结束，才会用之前的线程来执行新的任务。

import time

from concurrent.futures import ThreadPoolExecutor

# 括号内可以传数字指定线程数，不传的话，默认会开设当前计算机CPU个数5倍的线程
# 池子造出来后，会存在一定数量的线程，这些线程不会出现重复创建和销毁的过程
pool = ThreadPoolExecutor(5)

def task(n):
    print(n)
    time.sleep(1)
    return n*100

def call_back(n):
    print(f"call_back:{n.result()}")

t_list= []

# 池子的使用非常简单，只需要将需要做的任务往池子中提交即可
for i in range(20):
    # 异步提交任务的返回结果，应该通过回调机制来获取
    pool.submit(task,i).add_done_callback(call_back)      # 朝池子中提交任务，异步提交

print("主线程执行了")

十一、Event事件

一些进程/线程需要等待另外一些进程/线程运行完毕之后才能运行，类似于发射信号一样。这时，我们可以使用 Event 事件。

事件 Event 中有一个全局内置标志 flag，值为 True 或者 False。使用 wait() 函数的线程会处于阻塞状态，此时 flag 值为 False，直到有其他线程调用 set() 函数让全局标志 flag 置为 True，其阻塞的线程立刻恢复运行，还可以用 is_set() 函数检查当前的 flag 状态。

threading.Event.set()                   # 将标志设为True，并通知所有处于等待阻塞状态的线程恢复运行状态
threading.Event.clear()                 # 将标志设为False
threading.Event.wait(timeout=None)      # 如果标志为True将立即返回，否则阻塞线程至等待阻塞状态，等待其他线程调用set()
threading.Event.is_set()                # 获取内置标志状态，返回True或False。

import time

from threading import Thread, Event

event = Event()

def light():
    print("红灯亮着呢")
    time.sleep(3)
    print("绿灯亮了")
    # 告诉等待红灯的人可以走了
    event.set()

def car(name):
    print(f"{name}正在等红灯")
    # 别人通知不要等了
    event.wait()            # 等待别人给你发信号
    print(f"{name}开走了")

if __name__ == "__main__":
    t = Thread(target=light)
    t.start()

    for i in range(20):
        t = Thread(target=car, args=(f"小车{i}",))
        t.start()