线程理论

一台计算机相当于一个工厂，工厂里有很多车间（进程），车间里面有很多工人（线程）。 真正干活儿的是工人（线程）。

1.什么是线程

进程是资源分配的最小单位，而线程是进程中的一部分，每个进程中至少有一个线程，是CPU调度的最小单位。

进程是资源分配的最小单位，线程是CPU调度的最小单位线程开销更小，更轻量级。

2.线程的数据是共享的

在操作系统中，每一个进程的内存空间是独立的，数据不共享。

但是同一个进程中的各个线程，数据是共享的

二：开启线程的2种方式

1.第一种 - 普通方式

  1. from threading import Thread
  2. import time
  5. def task(t_name, s_time):
  6.     print(f'子线程[{t_name}]开始')
  7.     time.sleep(s_time)
  8.     print(f'子线程[{t_name}]结束')
  11. if __name__ == '__main__':
  12.     t1 = Thread(target=task, args=('t1', 1))  # 实例化得到一个线程对象
  13.     t1.start()  # 对象.start()启动线程
  14.     print('主线程')
  17. # 子线程[t1]开始
  18. # 主线程
  19. # 子线程[t1]结束

2.第二种 - 继承类的方式

  1. from threading import Thread
  2. import time
  5. class MyThread(Thread):
  6.     def run(self):
  7.         print('子线程开始')
  8.         time.sleep(1)
  9.         print('子线程结束')
  12. if __name__ == '__main__':
  13.     t = MyThread()
  14.     t.start()
  15.     print('主线程')
  18. # 子线程开始
  19. # 主线程
  20. # 子线程结束

三：用多线程实现TCP服务端并发效果

服务端：

  1. from socket import *
  2. from threading import Thread
  4. server = socket(AF_INET, SOCK_STREAM)
  5. server.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
  6. server.bind(('127.0.0.1', 8080))
  7. server.listen(5)
  10. def talk(conn, client_addr):
  11.     while True:
  12.         try:
  13.             print(f'{client_addr} connected.')
  14.             msg = conn.recv(1024)
  15.             if not msg:
  16.                 break
  17.             conn.send(msg.upper())
  18.         except Exception:
  19.             break
  22. if __name__ == '__main__':
  23.     while True:
  24.         conn, client_addr = server.accept()
  25.         t = Thread(target=talk, args=(conn, client_addr))
  26.         t.start()

客户端：

  1. from socket import *
  3. client = socket(AF_INET, SOCK_STREAM)
  4. client.connect(('127.0.0.1', 8080))
  6. while True:
  7.     msg = input('Input your message: ').strip()
  8.     if not msg:
  9.         continue
  11.     client.send(msg.encode('UTF-8'))
  12.     msg = client.recv(1024)
  13.     print(msg.decode('UTF-8'))

四：线程对象join方法 - 等待子线程执行结束

  1. from threading import Thread
  2. import time
  5. def task(t_name, s_time):
  6.     print(f'{t_name} 开始')
  7.     time.sleep(s_time)
  8.     print(f'{t_name} 结束')
  11. if __name__ == '__main__':
  12.     t1 = Thread(target=task, args=('t1', 1))
  13.     t2 = Thread(target=task, args=('t2', 2))
  15.     t1.start()
  16.     t2.start()
  18.     t1.join()  # 等待子线程执行结束
  19.     t2.join()
  21.     print('主线程')
  24. # t1 开始
  25. # t2 开始
  26. # t1 结束
  27. # t2 结束
  28. # 主线程

五：同一个进程下的多个线程数据共享

  1. from threading import Thread
  2. import time
  4. money = 99
  7. def task(t_name, n):
  8.     global money
  9.     print(f'{t_name} 开始')
  10.     money = n
  11.     print(f'{t_name} 结束')
  14. if __name__ == '__main__':
  15.     t1 = Thread(target=task, args=('t1', 10))
  16.     t2 = Thread(target=task, args=('t2', 100))
  18.     t1.start()
  19.     t2.start()
  21.     t1.join()
  22.     t2.join()
  24.     print(money)
  25.     print('主线程')
  28. # t1 开始
  29. # t1 结束
  30. # t2 开始
  31. # t2 结束
  32. # 100
  33. # 主线程

六：线程对象 及 其他方法

  1. from threading import Thread, current_thread, active_count
  2. import time
  5. def task():
  6.     print('子线程开始')
  7.     print(current_thread().name)  # 打印出当前线程名字
  8.     time.sleep(1)
  9.     print('子线程结束')
  12. if __name__ == '__main__':
  13.     t1 = Thread(target=task, name='I\'m Thread1')
  14.     t2 = Thread(target=task, name='I\'m Thread2')
  15.     t1.start()
  16.     t2.start()
  17.     print(f'存活线程数：{active_count()}')  # 打印出当前有多少存活的线程
  19. # 子线程开始
  20. # I'm Thread1
  21. # 子线程开始
  22. # I'm Thread2
  23. # 存活线程数：3
  24. # 子线程结束
  25. # 子线程结束
  26. from threading import Thread, current_thread, active_count
  27. import time
  28. import os
  31. def task(t_name, n):
  32.     print(f'{t_name}开始')
  33.     print(f'当前线程：{current_thread().name}')  # 线程名字
  34.     print(f'当前线程所在的进程ID：{os.getpid()}')
  35.     time.sleep(n)
  36.     print(f'{t_name}结束')
  39. if __name__ == '__main__':
  40.     t1 = Thread(target=task, name='Thread1', args=('t1', 2))
  41.     t2 = Thread(target=task, name='Thread2', args=('t2', 5))
  42.     t1.start()
  43.     t2.start()
  44.     t1.join()
  45.     print(f'线程t1是否存活：{t1.is_alive()}')
  46.     print(f'线程t2是否存活：{t2.is_alive()}')
  48.     # 当作线程id号
  49.     print(f'线程t1的ID：{t1.ident}')
  50.     print(f'线程t2的ID：{t2.ident}')
  52.     print(f'当前线程所在的进程ID：{os.getpid()}')
  53.     print(f'存活线程数：{active_count()}')  # 打印出2 ，存活的是t2和主线程
  56. # t1开始
  57. # 当前线程：Thread1
  58. # 当前线程所在的进程ID：37420
  59. # t2开始
  60. # 当前线程：Thread2
  61. # 当前线程所在的进程ID：37420
  62. # t1结束
  63. # 线程t1是否存活：False
  64. # 线程t2是否存活：True
  65. # 线程t1的ID：45360
  66. # 线程t2的ID：73396
  67. # 当前线程所在的进程ID：37420
  68. # 存活线程数：2
  69. # t2结束

知识点：

1.获取线程名：t.name、t.getName()，（name是人为设置的，有默认值：Thread - X）

2.查看当前进程下有几个线程存活：active_count()

3.查看当前线程是否存活：t1.is_alive()

4.查看线程ID：t1.ident

5.查看当前线程所在的进程ID：os.getpid()

七：守护线程

  1. from threading import Thread, current_thread, active_count
  2. import time
  5. def task(t_name, n):
  6.     print(f'{t_name}开始')
  7.     time.sleep(n)
  8.     print(f'{t_name}结束')
  11. if __name__ == '__main__':
  12.     t1 = Thread(target=task, name='Thread1', args=('Thread1', 10))
  13.     t2 = Thread(target=task, name='Thread2', args=('Thread2', 4))
  15.     t1.setDaemon(True)
  17.     t1.start()
  18.     t2.start()
  20.     print('主线程')
  23. # Thread1开始
  24. # Thread2开始
  25. # 主线程
  26. # Thread2结束

八：线程互斥锁

1.不加互斥锁

多个线程同时读写一个资源，会出现数据错乱

  1. from threading import Thread, Lock
  2. import time
  4. money = 99
  7. def task():
  8.     global money
  9.     temp = money
  10.     time.sleep(1)
  11.     money = temp - 1
  12.     print(money)
  15. if __name__ == '__main__':
  16.     ll = []
  17.     for i in range(10):
  18.         t = Thread(target=task)
  19.         t.start()
  20.         ll.append(t)
  22.     for i in ll:
  23.         i.join()
  25.     print(f'Final: {money}')
  28. # 98
  29. # 989898
  30. # 98
  31. # 98
  32. # 
  33. # 98
  34. # 98
  35. # 98
  36. # 
  37. # 98
  38. # Final: 98

2.加上互斥锁

将并发变成串行，牺牲了效率，但是保证了数据的安全性

  1. from threading import Thread, Lock
  2. import time
  4. money = 99
  7. def task(n, mutex):
  8.     global money
  9.     mutex.acquire()     # 在修改数据的时候，加锁
  10.     temp = money
  11.     # time.sleep(random.randint(1, 2))
  12.     time.sleep(0.1)
  13.     money = temp - 1
  14.     mutex.release()     # 修改完之后，释放锁，其他线程就可以抢到锁
  15.     print(money)
  18. if __name__ == '__main__':
  19.     ll = []
  20.     mutex = Lock()
  21.     for i in range(10):
  22.         t = Thread(target=task, args=(i, mutex))
  23.         t.start()
  24.         # t.join()    # 不能在这里join，会变成串行
  25.         ll.append(t)
  27.     for i in ll:
  28.         i.join()
  30.     print(f'Final: {money}')
  32. # 98
  33. # 97
  34. # 96
  35. # 95
  36. # 94
  37. # 93
  38. # 92
  39. # 91
  40. # 90
  41. # 89
  42. # Final: 89

3.了解

悲观锁、乐观锁、分布式锁

线程锁和进程锁 都属于 悲观锁

九：GIL 全局解释器锁

1.什么是GIL？

官网解释：

In CPython, the global interpreter lock, or GIL, is a mutex that prevents multiple native threads from executing Python bytecodes at once. This lock is necessary mainly because CPython’s memory management is not thread-safe. (However, since the GIL exists, other features have grown to depend on the guarantees that it enforces.) 在CPython中，全局解释器锁（GIL）是一个互斥体，它阻止多个本机线程同时执行Python字节码。cpyth的线程管理不是必需的。（然而，由于GIL的存在，其他特性也越来越依赖于它所执行的保证。）

GIL（Global Interpreter Lock）全局解释器锁

首先需要明确的一点是GIL并不是Python的特性，它是在实现Python解析器(CPython)时所引入的一个概念。

同样一段代码可以通过CPython，PyPy（执行速度比CPython快），Psyco等不同的Python执行环境来执行，然而因为CPython是大部分环境下默认的Python执行环境。

所以在很多人的概念里CPython就是Python，也就想当然的把GIL归结为Python语言的缺陷，所有GIL并不是python的特性，仅仅是因为历史原因在Cpython解释器中难以移除。

2.GIL存在的原因

①CPython在执行多线程的时候并不是线程安全的，所以为了程序的稳定性，加一把全局解释锁，能够确保任何时候都只有一个Python线程执行。

②因为垃圾回收机制（线程）的存在，如果一个定义了一个x=2，如果在一种较为极端的情况下，x已经生成，但是2还没赋值给x，此时的x就会被垃圾回收机制当成垃圾 被回收，为了避免这种情况出现，加了一把全局解释器锁。

3.为什么只有Python有GIL？

因为Python是一门动态语言，是用C编写的解释型语言，解释一行，执行一行。

前面定义了一个变量，不能保证后面一定会用得到。

4.线程释放GIL锁的2种情况

①.遇到I/O操作

例如：发出了HTTP请求，需要等待响应、读写文件…

②Time Tick 到期

Time Tick规定了线程的最长执行时间，超过时间后自动释放GIL锁。

3.单核与多核情况下的多线程任务调度

单核：

Python并发编程-线程 - 图1

由此可以发现：由于GIL机制的存在，单核CPU在同一时刻，只有一个线程在运行。 当线程遇到了I/O操作或者Time Tick到期的时候，就会释放GIL锁release GIL。 此时，其他2个线程去抢这把锁，抢到之后，线程才能运行。

虽然出现I/O操作和Time Tick到期都会释放GIL锁，但是这2种情况是不一样的：

出现I/O：

如果Thread1出现了IO操作，释放了GIL锁，Thread1会主动放弃抢锁，不再参与这场竞争。

Time Tick到期：

如果Thread1出现了TimeTick到期，释放了GIL锁，Thread1会和其他线程竞争，同时抢这把GIL锁

  • 所以，在单核的情况下，CPU的利用率是很高的。

多核：

Python并发编程-线程 - 图2

由此可以发现：Thread1在CPU1上运行，Thread2在CPU2上运行。 GIL是全局的，CPU2上的Thread2需要等待CPU1上的Thread1让出GIL锁，才有可能执行。 如果在多次竞争中，Thread2都胜出，Thread1没有得到GIL锁，意味着CPU1一直是闲置的，无法发挥多核的优势。

所以，有了GIL的存在，CPython的的多线程 其实就是 单线程。

4.针对与cpython解释器，有哪些注意点？

单核：

不管做什么，建议开多线程。

多核：

①针对IO密集型：开多线程，CPU遇到IO就会切换到其他线程执行，

②针对计算密集型：开多进程，能被多个CPU调度执行。

5.既然GIL有缺陷，为什么还有这么多人用呢？

①.python的库多，库都是基于cpython写起来的，其他解释器没有那么多的第三方库（海纳百川，有容乃大）。

②.虽然有这个历史诟病，但是丝毫不影响我们敲代码（金无足赤，人无完人）。

③.代码是一成不变的，但是我们可以利用它的优势，优化它的劣势（取其精华，去其糟粕）

④.Python牛逼！（人生苦短，我用Python）

6.总结

cpython解释器中只有一个全局锁（GIL），线程必须获取到GIL才能执行，我们开的多线程，不管有几个CPU，同一时刻，只能有一个线程在执行（python的多线程，不能利用多核优势）

十：验证GIL锁的存在方式

1.多线程

  1. from threading import Thread
  4. def task():
  5.     while True:
  6.         pass
  9. if __name__ == '__main__':
  10.     for i in range(16):
  11.         t = Thread(target=task)  # 因为有GIL锁的存在，同一时刻，只有一条线程执行，所以CPU不会满
  12.         t.start()

Python并发编程-线程 - 图3

2.多进程

  1. from multiprocessing import Process
  4. def task():
  5.     while True:
  6.         pass
  9. if __name__ == '__main__':
  10.     for i in range(16):
  11.         t = Process(target=task)  # 由于是多进程，进程中的线程会被CPU调度，16个CPU全部在工作，就会跑满
  12.         t.start()

Python并发编程-线程 - 图4

十一：GIL锁与普通互斥锁的区别

GIL锁并不能保证我们自己的数据安全，自己使用互斥锁才能保证自己的数据安全

  1. from threading import Thread, Lock
  2. import time
  4. mutex = Lock()
  5. money = 100
  8. def task():
  9.     global money
  10.     temp = money
  11.     # time.sleep(1)
  12.     money = temp - 1
  13.     print(money)
  16. if __name__ == '__main__':
  17.     for i in range(10):
  18.         t = Thread(target=task)
  19.         t.start()
  20.         # t.join()
  22.     print(money)

十二：IO密集型 和 计算密集型

1.计算密集型

开线程

  1. from threading import Thread
  2. import time
  5. # 计算密集型
  6. def task():
  7.     count = 0
  8.     for i in range(100000000):
  9.         count += i
  12. if __name__ == '__main__':
  13.     c_time = time.time()
  14.     l1 = []
  15.     for i in range(10):
  16.         t = Thread(target=task)
  17.         t.start()
  18.         l1.append(t)
  20.     for j in l1:
  21.         j.join()
  23.     print(time.time() - c_time)
  25. # 39.65782141685486

开进程

  1. from multiprocessing import Process
  2. import time
  5. # 计算密集型
  6. def task():
  7.     count = 0
  8.     for i in range(100000000):
  9.         count += i
  12. if __name__ == '__main__':
  13.     c_time = time.time()
  14.     l1 = []
  15.     for i in range(10):
  16.         t = Process(target=task)
  17.         t.start()
  18.         l1.append(t)
  20.     for j in l1:
  21.         j.join()
  23.     print(time.time() - c_time)
  25. # 7.2425291538238525

2.IO密集型

开线程

  1. from threading import Thread
  2. import time
  5. # 计算密集型
  6. def task():
  7.     time.sleep(2)
  10. if __name__ == '__main__':
  11.     c_time = time.time()
  12.     l1 = []
  13.     for i in range(400):
  14.         t = Thread(target=task)
  15.         t.start()
  16.         l1.append(t)
  18.     for j in l1:
  19.         j.join()
  21.     print(time.time() - c_time)
  23. # 2.0086498260498047

开进程

  1. from multiprocessing import Process
  2. import time
  5. # 计算密集型
  6. def task():
  7.     time.sleep(2)
  10. if __name__ == '__main__':
  11.     c_time = time.time()
  12.     l1 = []
  13.     for i in range(400):
  14.         t = Process(target=task)
  15.         t.start()
  16.         l1.append(t)
  18.     for j in l1:
  19.         j.join()
  21.     print(time.time() - c_time)
  23. # 7.729241371154785

十三：死锁现象（哲学家就餐问题） 与 递归锁（可重入所）

1.什么是死锁现象

指两个或两个以上的进程或线程在执行过程中，因争夺资源而造成的一种互相等待的现象，若无外力作用，它们都将无法推进下去。此时称系统处于死锁状态或系统产生了死锁，这些永远在互相等待的进程称为死锁进程，如下就是死锁

①出现死锁的情况1

进程1拿到了A锁，等待B锁；

进程2拿到了B锁，等待A锁；

进程1和进程2相互等待，僵持不下。

②出现死锁的情况2

进程1拿到了A锁，进程1再次拿A锁；

此时，A锁已经被进程1拿到，没有释放，无法再次拿A锁，程序凝滞了。

2.死锁实例

  1. from threading import Thread, Lock
  2. import time
  4. mutexA = Lock()
  5. mutexB = Lock()
  8. # print(mutexA is mutexB)  # False  # 两把锁不同
  10. def eat_apple(name):
  11.     mutexA.acquire()
  12.     print(f'{name} 获取了A锁')
  13.     mutexB.acquire()
  14.     print(f'{name} 获取了B锁')
  15.     print('开始吃苹果')
  16.     print('苹果吃完了')
  17.     mutexB.release()
  18.     print(f'{name} 释放了B锁')
  19.     mutexA.release()
  20.     print(f'{name} 释放了A锁')
  23. def eat_egg(name):
  24.     mutexB.acquire()
  25.     print(f'{name} 获取了B锁')
  26.     time.sleep(2)
  27.     mutexA.acquire()
  28.     print(f'{name} 获取了A锁')
  29.     print('开始吃鸡蛋')
  30.     print('鸡蛋吃完了')
  31.     mutexA.release()
  32.     print(f'{name} 释放了A锁')
  33.     mutexB.release()
  34.     print(f'{name} 释放了B锁')
  37. if __name__ == '__main__':
  38.     l1 = ['Alan', 'Ben', 'Cindy']
  39.     for name in l1:
  40.         t1 = Thread(target=eat_apple, args=(name,))
  41.         t2 = Thread(target=eat_egg, args=(name,))
  42.         t1.start()
  43.         t2.start()
  46. # Alan 获取了A锁
  47. # Alan 获取了B锁
  48. # 开始吃苹果
  49. # 苹果吃完了
  50. # Alan 释放了B锁
  51. # Alan 释放了A锁
  52. # Alan 获取了B锁
  53. # Ben 获取了A锁
  55. # Alan正在第二轮获取到B锁之后，sleep了2秒，A锁被Ben抢走，两者相互等待，出现了死锁现象

3.什么是递归锁？

递归锁（可重入），可以多次acquire，每acquire一次，内部计数器+1，每release一次，内部计数器-1。

可重入：可以被同一个人acquire多次

只有计数器为0，其他人才可以获取这把锁。

4.递归锁 实例

  1. from threading import Thread, RLock
  2. import time
  4. mutexA = mutexB = RLock()  # 链式复制
  7. # print(mutexA is mutexB)  # True  # 同一把锁
  9. def eat_apple(name):
  10.     mutexA.acquire()
  11.     print(f'{name} 获取了A锁')
  12.     mutexB.acquire()
  13.     print(f'{name} 获取了B锁')
  14.     print('开始吃苹果')
  15.     print('苹果吃完了')
  16.     mutexB.release()
  17.     print(f'{name} 释放了B锁')
  18.     mutexA.release()
  19.     print(f'{name} 释放了A锁')
  22. def eat_egg(name):
  23.     mutexB.acquire()
  24.     print(f'{name} 获取了B锁')
  25.     time.sleep(2)
  26.     mutexA.acquire()
  27.     print(f'{name} 获取了A锁')
  28.     print('开始吃苹果')
  29.     print('鸡蛋吃完了')
  30.     mutexA.release()
  31.     print(f'{name} 释放了A锁')
  32.     mutexB.release()
  33.     print(f'{name} 释放了B锁')
  36. if __name__ == '__main__':
  37.     l1 = ['Alan', 'Ben', 'Cindy']
  38.     for name in l1:
  39.         t1 = Thread(target=eat_apple, args=(name,))
  40.         t2 = Thread(target=eat_egg, args=(name,))
  41.         t1.start()
  42.         t2.start()
  44. # Alan 获取了A锁
  45. # Alan 获取了B锁
  46. # 开始吃苹果
  47. # 苹果吃完了
  48. # Alan 释放了B锁
  49. # Alan 释放了A锁
  50. # Alan 获取了B锁
  51. # Alan 获取了A锁
  52. # 开始吃苹果
  53. # 鸡蛋吃完了
  54. # Alan 释放了A锁
  55. # Alan 释放了B锁
  56. # Ben 获取了A锁
  57. # Ben 获取了B锁
  58. # 开始吃苹果
  59. # 苹果吃完了
  60. # Ben 释放了B锁
  61. # Ben 释放了A锁
  62. # Ben 获取了B锁
  63. # Ben 获取了A锁
  64. # 开始吃苹果
  65. # 鸡蛋吃完了
  66. # Ben 释放了A锁
  67. # Ben 释放了B锁
  68. # Cindy 获取了A锁
  69. # Cindy 获取了B锁
  70. # 开始吃苹果
  71. # 苹果吃完了
  72. # Cindy 释放了B锁
  73. # Cindy 释放了A锁
  74. # Cindy 获取了B锁
  75. # Cindy 获取了A锁
  76. # 开始吃苹果
  77. # 鸡蛋吃完了
  78. # Cindy 释放了A锁
  79. # Cindy 释放了B锁

十四：Semaphore信号量

1.什么是信号量？

Semaphore信号量，可以理解为多把锁，同时允许多个线程来更改数据。

2.应用场景

多个人抢占固定的公共场所资源

3.实例

  1. from threading import Thread, Semaphore
  2. import time
  3. import random
  5. sm = Semaphore(3)  # 数字表示可以同时有多少个线程操作
  8. def task(name):
  9.     sm.acquire()
  10.     print(f'{name} 正在占用')
  11.     time.sleep(random.randint(1, 5))
  12.     sm.release()
  13.     print(f'{name} 出去了')
  16. if __name__ == '__main__':
  17.     for i in range(10):
  18.         t = Thread(target=task, args=(f'路人{i}号',))
  19.         t.start()
  22. # 路人0号 正在占用
  23. # 路人1号 正在占用
  24. # 路人2号 正在占用
  26. # 路人1号 出去了
  27. # 路人3号 正在占用
  29. # 路人3号 出去了
  30. # 路人4号 正在占用
  31. # 路人0号 出去了
  32. # 路人5号 正在占用
  33. # 路人2号 出去了
  34. # 路人6号 正在占用
  35. # 路人5号 出去了
  36. # 路人4号 出去了
  37. # 路人7号 正在占用
  38. # 路人8号 正在占用
  39. # 
  40. # 路人8号 出去了
  41. # 路人9号 正在占用
  42. # 路人7号 出去了
  43. # 路人6号 出去了
  44. # 路人9号 出去了

十五：Event事件

一些线程需要等到其他线程执行完成之后才能执行，类似于发射信号 比如一个线程等待另一个线程执行结束再继续执行

场景：1个进程读取一个文件的前半部分，读完了之后，发1个信号，另1个进程读取这个文件的后半部分

1.实例1 - 教师抢教室

  1. from threading import Thread, Event
  2. import time
  3. import random
  5. event = Event()
  8. def room(name):
  9.     print(f'教室 {name} 正在被使用')
  10.     time.sleep(random.randint(1, 3))
  11.     print(f'教室 {name} 使用完毕，门已打开')
  12.     event.set()
  15. def teacher(name):
  16.     print(f'{name} 正在等待教室开门')
  17.     event.wait()  # 只要教室门没开，老师就只能在外面等着
  18.     print(f'门开了 {name} 开始冲刺了')
  21. if __name__ == '__main__':
  22.     r1 = Thread(target=room, args=('Room1',))
  23.     r1.start()
  25.     for i in range(5):
  26.         t = Thread(target=teacher, args=(f'老师{i}',))
  27.         t.start()
  30. # 教室 Room1 正在被使用
  31. # 老师0 正在等待教室开门
  32. # 老师1 正在等待教室开门
  33. # 老师2 正在等待教室开门
  34. # 老师3 正在等待教室开门
  35. # 老师4 正在等待教室开门
  36. # 教室 Room1 使用完毕，门已打开
  37. # 门开了 老师0 开始冲刺了
  38. # 门开了 老师1 开始冲刺了
  39. # 门开了 老师3 开始冲刺了
  40. # 门开了 老师2 开始冲刺了
  41. # 门开了 老师4 开始冲刺了

2.实例2 - 文件分批读取

a.txt

  1. 1234567890

py

  1. from threading import Thread, Event
  2. import os
  4. event = Event()
  5. size = os.path.getsize('a.txt')  # 获取文件总大小
  8. def read_first():
  9.     with open('a.txt', mode='r', encoding='UTF-8') as f:
  10.         n = size // 2  # 整除 取文件的一半
  11.         data = f.read(n)
  12.         print(data)
  13.         print('我已读完一半了，发出信号')
  14.         event.set()  # 文件的一半读取完毕，发送信号
  17. def read_last():
  18.     event.wait()  # 等待信号
  19.     with open('a.txt', mode='r', encoding='UTF-8') as f:
  20.         n = size // 2  # 整除 取文件的一半
  21.         f.seek(n, 0)  # 光标从文件开头开始，移动了n个字节，移动到了文件的一半
  22.         data = f.read()
  23.         print(data)
  24.         event.set()  # 文件的一半读取完毕，发送信号
  27. if __name__ == '__main__':
  28.     t1 = Thread(target=read_first)
  29.     t2 = Thread(target=read_last)
  31.     t1.start()
  32.     t2.start()
  35. # 12345
  36. # 我已读完一半了，发出信号
  37. # 67890

十六：线程Queue

1.进程queue 和 线程queue 不是同一个queue

进程queue

  1. from multiprocessing import Queue

线程queue

  1. from queue import Queue

线程间通信，因为共享变量会出现数据不安全问题，用线程queue通信，不需要加锁，内部自带。

queue是线程安全的

2.三种Queue

  • Queue：队列 FIFO 先进先出
  • PriorityQueue：优先级队列，谁小 谁先出
  • LifoQueue：栈 LIFO 后进先出

3.如何使用

Queue

  1. from queue import Queue
  3. q = Queue(5)
  4. q.put('Alan')
  5. q.put('Ben')
  6. q.put('Cindy')
  7. q.put('Dad')
  8. q.put('Edge')
  10. print(q.get())      # Alan
  11. print(q.get())      # Ben
  12. print(q.get())      # Cindy
  13. print(q.get())      # Dad
  14. print(q.get())      # Edge
  16. # print(q.get())    # 卡住
  17. # print(q.get_nowait())   # 不会卡，但是会报错
  19. # print(q.full())     # 是否满
  20. # print(q.empty())    # 是否空

LifoQueue

  1. from queue import LifoQueue
  3. q = LifoQueue(5)
  4. q.put('Alan')
  5. q.put('Ben')
  6. q.put('Cindy')
  7. q.put('Dad')
  8. q.put('Edge')
  10. print(q.get())      # Edge
  11. print(q.get())      # Dad
  12. print(q.get())      # Cindy
  13. print(q.get())      # Ben
  14. print(q.get())      # Alan

PriorityQueue

  1. from queue import PriorityQueue
  3. q = PriorityQueue(5)
  4. q.put((10, 'Alan'))
  5. q.put((100, 'Ben'))
  6. q.put((99, 'Cindy'))
  7. q.put((1002, 'Dad'))
  8. q.put((999, 'Edge'))
  10. print(q.get())      # (10, 'Alan')
  11. print(q.get())      # (99, 'Cindy')
  12. print(q.get())      # (100, 'Ben')
  13. print(q.get())      # (999, 'Edge')
  14. print(q.get())      # (1002, 'Dad')

十七：线程池

1.为什么会出现 池？

不管是开进程 还是 开线程，都不能无限制地开，通过 池，假设池子里有10个，不管之后再怎么开，永远都是这10个。

2.实例

无池 - 持续不断地输出

  1. from threading import Thread
  2. import time
  3. pool = ThreadPoolExecutor(5)    # 填写的数字，就是池的大小
  5. def task():
  6.     print('Task Start')
  7.     time.sleep(2)
  8.     print('Task Done')
  11. if __name__ == '__main__':
  12.     for i in range(100):    # 启动了100个线程
  13.         t = Thread(target=task)
  14.         t.start()
  16. # Task Start
  17. # Task Start
  18. # Task Start
  19. # Task Start
  20. # Task Start
  21. # Task Start
  22. # Task Start
  23. # Task Start
  24. # Task Start
  25. # Task Start
  26. # Task Start
  27. # Task Start
  28. # Task Start

有池 - 最多只能5个5个输出

  1. from concurrent.futures import ThreadPoolExecutor
  2. import time
  3. pool = ThreadPoolExecutor(5)    # 填写的数字，就是池的大小
  5. def task():
  6.     print('Task Start')
  7.     time.sleep(2)
  8.     print('Task Done')
  11. if __name__ == '__main__':
  12.     for i in range(100):    # 启动了100个线程
  13.         pool.submit(task,)
  15. # Task Start
  16. # Task Start
  17. # Task Start
  18. # Task Start
  19. # Task Start
  21. # Task Done
  22. # Task Done
  24. # Task Start
  26. # Task Done
  27. # Task Done
  29. # Task Start
  31. # Task Done
  33. # Task Start
  34. # Task Start
  36. # Task Start

3.爬虫小案例

  1. from concurrent.futures import ThreadPoolExecutor
  3. import requests  # 爬虫会学到的模块（需要自行安装：pip3 install requests）
  5. pool = ThreadPoolExecutor(2)
  8. def get_pages(url):
  9.     # https://www.baidu.com
  10.     res = requests.get(url)  # 向这个地址发送请求
  12.     name = url.rsplit('/')[-1] + '.html'
  13.     print(name)  # www.baidu.com.html
  14.     # res.content拿到页面的二进制
  15.     return {'name': name, 'text': res.content}
  18. def call_back(f):
  19.     dic = f.result()
  20.     with open(dic['name'], 'wb') as f:
  21.         f.write(dic['text'])
  24. if __name__ == '__main__':
  25.     l1 = ['https://www.baidu.com', 'https://www.bilibili.com', 'https://www.cnblogs.com']
  26.     for url in l1:
  27.         pool.submit(get_pages, url).add_done_callback(call_back)

十八：不同线程之间 数据交互的2种方式

1.共享变量

不同线程修改同一份数据需要加锁（互斥锁）

2.通过Queue

不需要考虑数据安全问题（线程安全了）

十九：线程池的shutdown

主线程等待所有任务执行完成之后，再运行

  1. from concurrent.futures import ThreadPoolExecutor
  2. import time
  4. pool = ThreadPoolExecutor(3)
  7. def task(name):
  8.     print(f'{name} 开始')
  9.     time.sleep(1)
  10.     print(f'{name} 结束')
  13. if __name__ == '__main__':
  14.     for i in range(20):
  15.         pool.submit(task, f'Use {i}')
  17.     # 放到for循环外面，等待所有任务执行完成，主线程再继续走
  18.     pool.shutdown(wait=True)    # 等待所有任务完成，并且把池关闭（默认的wait=True）
  19.     # 问题：关闭之后还能提交任务吗？ 不能！！！
  20.     # pool.submit(task, 'emmmm')  # RuntimeError: cannot schedule new futures after shutdown
  21.     print('Main')  # 立马执行，等20个线程都执行完了，再执行

二十：定时器（线程）

多长时间之后 执行一个任务

Timer(时间, 函数)

  1. from threading import Timer
  4. def task():
  5.     print('I\'m Darker')
  8. if __name__ == '__main__':
  9.     t = Timer(1, task)  # 本质就是开2个线程，延迟1秒执行
  10.     t.start()

args传值

  1. from threading import Timer
  4. def task(name):
  5.     print(f'I\'m {name}')
  8. if __name__ == '__main__':
  9.     t = Timer(1, task, args=('Darker',))
  10.     t.start()

kwargs传值

  1. from threading import Timer
  4. def task(name):
  5.     print(f'I\'m {name}')
  8. if __name__ == '__main__':
  9.     t = Timer(1, task, kwargs={'name': 'Darker'})
  10.     t.start()