ToB企服应用市场:ToB评测及商务社交产业平台

标题: 多使命之进程与线程 [打印本页]

作者: 前进之路 时间: 2024-5-15 09:22
标题: 多使命之进程与线程
多使命进程与线程

一、多使命介绍

我们生存中有很多事情是同时举行的，好比开车的时间手和脚共同来驾驶汽车，再好比唱歌跳舞也是同时举行的；用程序来模拟：

from time import sleep
def sing():
for i in range(3):
print("正在唱歌...%d"%i)
sleep(1)
def dance():
for i in range(3):
print("正在跳舞...%d"%i)
sleep(1)
if __name__ == '__main__':
sing()
dance()

复制代码

总结：

很显然刚刚的程序并没有完成唱歌和跳舞同时举行的要求,我们称之为单进程
如果想要实现“唱歌跳舞”同时举行，那么就必要一个新的方法，叫做：多使命
那什么是多使命

简单地说，就是操作体系可以同时运行多个使命。打个比方，你一边在用浏览器上网，一边在听MP3，一边在用Word赶作业，这就是多使命
现在，多核CPU已经非常遍及了，但是，即使已往的单核CPU，也可以执行多使命。由于CPU执行代码都是顺序执行的，那么，单核CPU是怎么执行多使命的呢？
答案就是操作体系轮流让各个使命交替执行，使命1执行0.01秒，切换到使命2，使命2执行0.01秒，再切换到使命3，执行0.01秒……这样反复执行下去。表面上看，每个使命都是交替执行的，但是，由于CPU的执行速度实在是太快了，我们感觉就像所有使命都在同时执行一样。
真正的并行执行多使命只能在多核CPU上实现，但是，由于使命数量远远多于CPU的核心数量，所以，操作体系也会自动把很多使命轮流调度到每个核心上执行。所以以上只能算是并发。
并发与并行
- 并发：指的是使命数多余CPU核数，通过操作体系的各种使命调度算法，实现用多个使命“一起”执行（实际上总有一些使命不在执行，因为切换使命的速度相当快，看上去一起执行而已）
- 并行：指的是使命数小于等于CPU核数，即使命真的是一起执行的

二、进程

1> 基本概念

程序：例如xxx.py这是程序，是一个静态的
进程：一个程序运行起来后，代码+用到的资源称之为进程，它是操作体系分配资源的基本单位。
工作中，使命数每每大于CPU的核数，即一定有一些使命正在执行，而另外一些使命在等候CPU举行执行，因此导致了有了不同的状态
- 停当态：运行的条件都已经慢去，正在等在CPU执行
- 执行态：CPU正在执行其功能
- 等候态：等候某些条件满足，例如一个程序sleep了，此时就处于等候态

2> 进程使用和特性

multiprocessing模块就是跨平台版本的多进程模块，提供了一个Process类来代表一个进程对象，这个对象可以理解为是一个独立的进程，可以执行另外的事情

进程的简单实现
1. from multiprocessing import Process
2. import time
5. def func():
6. while True:
7. print("【子进程】")
8. time.sleep(3)
11. if __name__ == '__main__':
12. p = Process(target=func)
13. p.start()
14. while True:
15. print("【主进程】")
16. time.sleep(3)
复制代码
总结：创建子进程时，只必要传入一个执行函数和函数的参数，创建一个Process实例，用start()方法启动
进程通报参数与进程对象和进程ID
1. import multiprocessing
2. import os
3. from time import sleep
6. def func(name, age, **kwargs):
7. print(f"【子进程】({multiprocessing.current_process()}) 的进程号为:{os.getpid()}")
8. print(f"【子进程】 name={name}, age={age}, kwargs={kwargs}")
9. for i in range(10):
10. print(f"【子进程】 ---{i}---")
11. sleep(0.2)
14. if __name__ == '__main__':
15. p = multiprocessing.Process(target=func, args=('test', 18), kwargs={"m": 20})
16. p.start()
17. sleep(1)
18. # 主进程会等待所有的子进程执行结束再结束
19. print("【主进程】--- 结束 ---")
复制代码
使用多进程实现UDP通信同时收发数据
1. import socket
2. from multiprocessing import Process
5. def recv_data(udp_socket):
6. while True:
7. recv_msg = udp_socket.recvfrom(1024)
8. data = recv_msg[0].decode("gbk")
9. source = recv_msg[1]
10. print(f"{source}: {data}")
13. def main():
14. udp_client = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
15. udp_client.bind(("", 9000))
17. # 子进程接收数据
18. Process(target=recv_data, args=(udp_client,)).start()
19. # 主进程发送数据 (子进程中无法使用input)
20. while True:
21. msg = input("请输入发送的数据： ").encode("gbk")
22. dest_addr = ("127.0.0.1", 8000)
23. udp_client.sendto(msg, dest_addr)
26. if __name__ == "__main__":
27. main()
复制代码
进程间不共享全局变量
1. from multiprocessing import Process
2. import os
3. import time
5. nums = [11, 22]
8. def work1():
9. print(f"【子进程work1】 pid={os.getpid()} ,nums={nums}")
10. for i in range(3):
11. nums.append(i)
12. time.sleep(1)
13. print(f"【子进程work1】 pid={os.getpid()} ,nums={nums}")
14. print("【子进程work1】", id(nums), nums)
17. def work2():
18. print(f"【子进程work2】 pid={os.getpid()} ,nums={nums}")
19. print("【子进程work2】", id(nums), nums)
22. print(f"当前进程为{os.getpid()}, 进程中nums的id为{id(nums)}")
23. if __name__ == '__main__':
24. p1 = Process(target=work1)
25. p1.start()
26. time.sleep(5)
28. p2 = Process(target=work2)
29. p2.start()
30. print(f"【主进程】 pid={os.getpid()} ,nums={nums}")
复制代码
进程间通信

在多进程编程中，不同的进程之间必要举行通信。multiprocessing模块提供了多种进程间通信的方式，例如使用队列、管道、共享内存等进程通信
- 生产者消费者模型
  
  生产者消费者模式就是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过壅闭队列来举行通讯，所以生产者生产完数据之后不消等候消费者处理，直接扔给壅闭队列，消费者不找生产者要数据，而是直接从壅闭队列里取，壅闭队列就相当于一个缓冲区，均衡了生产者和消费者的处理能力。这个壅闭队列就是用来给生产者和消费者解耦的。
- 队列
  - multiprocessing.Queue()和queue.Queue()的区别
    - queue.Queue是进程内非壅闭队列，multiprocess.Queue是跨进程通信队列。
    - queue.Queue是进程内的用的队列，也就是多线程,multiprocessing.Queue是跨进程通信队列，也就是多进程
  - multiprocessing.Queue() 和 queue.Queue()队列使用
    from multiprocessing import Queue
    # 初始化队列;若括号中没有指定最大可接收的消息数量，或数量为负值，那么就代表可接受的消息数量没有上限（直到内存的尽头）
    q=Queue()
    # 返回当前队列包含的消息数量
    q.qsize()
    # 如果队列为空，返回True，反之False
    q.empty()
    # 如果队列满了，返回True,反之False；
    q.full()
    # 获取队列中的一条消息,然后将其从列队中移除,block默认值为True
    # Queue.get([block[, timeout]])
    q.get()
    # 相当Queue.get(False)；
    q.get_nowait()
    # 将item消息写入队列，block默认值为True；
    # Queue.put(item,[block[, timeout]])
    q.put()
    # 相当Queue.put(item, False)；
    q.put_nowait(item)
    复制代码
  - queue.Queue() 其他功能
    # 与multiprocessing.Queue()基本一致
    # 增加了队列计数器来实现队列的阻塞控制
    from queue import Queue
    # 队列阻塞，直到队列中的【所有项目】都已经被获取并【处理】才会解堵塞，
    # 如果线程里每从队列里取一次，但没有执行task_done()，则join无法判断队列到底有没有结束
    q.join()
    # 队列解堵塞，指示以前已排队的任务已完成，一般搭配.join使用
    q.task_done()
    复制代码
    总结：
    队列内部有一个计数器来实现队列的阻塞控制; 当调用 q.join()会开启队列阻塞，直到计数器计数为0则解堵塞。往队列q.put一个数据计数器加一，每执行一次q.task_done()队列计数器减一(q.get不影响计数器计数值);q.size()=0或q.empty()=True,只能表示队列中没有任务了，不能保证任务已经执行完成。
    复制代码
- 使用队列和多进程实现生产者消费者模型
  1. from multiprocessing import Process, Queue
  2. import time
  3. import random
  6. def producer(q):
  7. for value in ['A', 'B', 'C', 'D', 'E', 'F']:
  8. print(f'【Producer】 put {value} to queue...')
  9. q.put(value)
  10. time.sleep(random.random())
  11. # 发送结束信号
  12. q.put(None)
  15. def consumer(q):
  16. while True:
  17. if not q.empty():
  18. data = q.get()
  19. # 接收到结束信号退出程序
  20. if data is None:
  21. return
  22. print(f"【Consumer】 get {data} from queue")
  23. time.sleep(random.random())
  26. q = Queue()
  27. if __name__ == "__main__":
  28. pw = Process(target=producer, args=(q,))
  29. pr = Process(target=consumer, args=(q,))
  31. pw.start()
  32. pr.start()
  复制代码
进程池Pool

当必要创建的子进程数量不多时，可以直接利用multiprocessing中的Process动态成生多个进程，但如果是上百甚至上千个目标，手动的去创建/销毁进程的工作量巨大，此时就可以用到multiprocessing模块提供的Pool方法
1. 初始化Pool时，可以指定一个最大进程数，当有新的请求提交到Pool中时，如果池还没有满，那么就会创建一个新的进程用来执行该请求；但如果池中的进程数已经达到指定的最大值，那么该请求就会等待，直到池中有进程结束，才会用之前的进程来执行新的任务
复制代码
1. from multiprocessing import Pool
2. import time
3. import random
4. import os
7. def work(msg):
8. start_time = time.time()
9. print("任务{msg} 开始执行,进程号 {os.getpid()}")
10. time.sleep(random.random() * 2)
11. end_time = time.time()
12. print("任务{msg} 结束执行,运行时间{end_time - start_time}")
15. if __name__ == "__main__":
16. p = Pool(3)
17. for i in range(10):
18. p.apply_async(work, args=(i,))
20. # 观察进程池任务什么时候开始执行
21. time.sleep(1)
22. print("------start--------")
23. # 关闭Pool，使其不再接受新的任务；
24. p.close()
25. # 主进程阻塞，等待子进程的退出，必须在close或terminate之后使用；
26. # p.join()
27. print("--------end--------")
复制代码
总结：
- 主进程不会自动等候进程池使命执行,如果主进程执行完毕,进程池使命立即结束
- 程池在定义的时间没有指定最大进程数,体系会按当前运行盘算机的CPU核心数决定进程池内运行的最大进程数,如盘算机为双核,则进程池内最大进程数为2
参数分析(multiprocessing.Pool)：
- apply_async(func[, args[, kwds]]) ：使用非壅闭方式调用func（并行执行，堵塞方式必须等候上一个进程退出才能执行下一个进程），args为通报给func的参数列表，kwds为通报给func的关键字参数列表；
- close()：关闭Pool，使其不再担当新的使命；
- terminate()：不管使命是否完成，立即终止；
- join()：主进程壅闭，等候子进程的退出，必须在close或terminate之后使用；
进程池中的Queue
1. 如果要使用Pool创建进程，就需要使用multiprocessing.Manager()中的Queue()，而不是multiprocessing.Queue()
复制代码
- 使用队列和进程池实现生产者消费者模型
  1. from multiprocessing import Manager, Pool
  2. import time
  3. import random
  4. import os
  7. def producer(q):
  8. print("producer启动({os.getpid()}),父进程为({os.getppid()})")
  9. for value in ['A', 'B', 'C', 'D', 'E', 'F']:
  10. print(f'Put {value} to queue...')
  11. q.put(value)
  12. time.sleep(random.random())
  13. # 发送结束信号
  14. q.put(None)
  17. def consumer(q):
  18. print("consumer启动({os.getpid()}),父进程为({os.getppid()})")
  19. while True:
  20. if not q.empty():
  21. data = q.get()
  22. # 接收到结束信号退出程序
  23. if data is None:
  24. return
  25. print(f"get {data} from queue")
  26. time.sleep(random.random())
  29. if __name__ == "__main__":
  30. q = Manager().Queue()
  31. p = Pool()
  32. p.apply_async(producer, (q,))
  33. p.apply_async(consumer, (q,))
  35. p.close()
  36. p.join()
  37. print(f"主进程{os.getpid()} 结束")
  复制代码

三、线程

1> 基本概念

线程：线程是进程的一个实体,是CPU调度和分派的基本单位,它是比进程更小的能独立运行的基本单位.线程本身基本上不拥有体系资源,只拥有一点在运行中必不可少的资源(如程序计数器,一组寄存器和栈),但是它可与同属一个进程的其他的线程共享进程所拥有的全部资源.

2> 线程使用和特性

线程的简单实现与线程标识符
1. import threading
2. from time import sleep, ctime
5. def sing():
6. print(f"\n【子线程sing】({threading.current_thread()})的标识符为: {threading.get_ident()}")
7. for i in range(3):
8. print(f"正在唱歌...{i}")
9. sleep(3)
10. print(f"【子线程sing】--- 结束 ---")
13. def dance():
14. print(f"\n【子线程dance】({threading.current_thread()})的标识符为: {threading.get_ident()}")
15. for i in range(3):
16. print(f"正在跳舞...{i}")
17. sleep(3)
18. print(f"【子线程dance】--- 结束 ---")
21. if __name__ == '__main__':
22. print(f"【主线程】---开始---:{ctime()}")
24. t1 = threading.Thread(target=sing)
25. t2 = threading.Thread(target=dance)
27. t1.start()
28. t2.start()
29. print(f"【主线程】子线程sing的标识符: {t1.ident}")
30. print(f"【主线程】子线程dance的标识符: {t2.ident}")
32. length = len(threading.enumerate())
33. print(f'【主线程】当前运行的线程数为：{length}')
35. print(f"【主线程】--- 结束 ---:{ctime()}")
复制代码
总结：
- 多线程并发的操作比单线程效率更高
线程间共享全局变量
1. from threading import Thread
2. import time
4. nums = [11, 22]
7. def work1():
8. for i in range(3):
9. nums.append(i)
11. print(f"【子线程work1】 g_num={nums}")
14. def work2():
15. print(f"【子线程work2】 g_num={nums}")
18. print(f"【主线程】子线程创建之前g_num={nums}")
19. t1 = Thread(target=work1)
20. t1.start()
22. # 延时一会，保证t1线程中的任务做完
23. time.sleep(1)
25. t2 = Thread(target=work2)
26. t2.start()
复制代码
总结：
- 一个进程内的所有线程共享全局变量，很方便在多个线程间共享数据
- 缺点就是，线程是对全局变量随意遂改大概造成多线程之间对全局变量的混乱
多线程资源竞争问题
1. import threading
2. import time
4. g_num = 0
7. def work1(num):
8. global g_num
9. for i in range(num):
10. g_num += 1
11. print(f"【子线程work1】 g_num={g_num}")
14. def work2(num):
15. global g_num
16. for i in range(num):
17. g_num += 1
18. print(f"【子线程work2】 g_num={g_num}")
21. print(f"【主线程】子线程创建之前g_num={g_num}")
23. t1 = threading.Thread(target=work1, args=(1000000,))
24. t2 = threading.Thread(target=work2, args=(1000000,))
25. t1.start()
26. t2.start()
28. while len(threading.enumerate()) != 1:
29. time.sleep(1)
31. print(f"【主线程】2个线程对同一个全局变量操作之后的最终结果是:{g_num}")
复制代码
- 结果分析：
  1. 假设两个线程t1和t2都要对全局变量g_num(默认是0)进行加1运算，t1和t2都各对`g_num`加10次，g_num的最终的结果应该为20。但是由于是多线程同时操作，有可能出现下面情况：
  3. 1> 在g_num=0时，t1取得g_num=0。此时系统把t1调度为”sleeping”状态，把t2转换为”running”状态，t2也获得g_num=0
  4. 2> 然后t2对得到的值进行加1并赋给g_num，使得g_num=1
  5. 3> 然后系统又把t2调度为”sleeping”，把t1转为”running”。线程t1又把它之前得到的0加1后赋值给g_num。
  6. 4> 这样导致虽然t1和t2都对g_num加1，但结果仍然是g_num=1
  复制代码
解决多线程资源竞争问题 - 互斥锁
- 互斥锁
  
  解决多线程资源竞争问题，最简单的机制就是引入互斥锁，互斥锁为资源引入一个状态：锁定/非锁定
- 互斥锁工作原理
  
  某个线程要更改共享数据时，先将其锁定，此时资源的状态为“锁定”，其他线程不能更改；直到该线程释放资源，将资源的状态变成“非锁定”，其他的线程才能再次锁定该资源。互斥锁保证了每次只有一个线程举行写入操作，从而保证了多线程环境下数据的准确性。
- 互斥锁使用
  1. # 创建锁
  2. mutex = threading.Lock()
  4. # 锁定
  5. mutex.acquire()
  7. # 释放
  8. mutex.release()
  复制代码
  分析：
  - 如果这个锁之前是没有上锁的，那么acquire不会堵塞
  - 如果在调用acquire对这个锁上锁之前它已经被其他线程上了锁，那么此时acquire会堵塞，直到这个锁被解锁为止
- 为全局变量加入互斥锁
  1. import threading
  2. import time
  4. g_num = 0
  7. def work1(num):
  8. global g_num
  9. for i in range(num):
  10. mutex.acquire() # 上锁
  11. g_num += 1
  12. mutex.release() # 解锁
  14. print(f"【子线程work1】 g_num={g_num}")
  17. def work2(num):
  18. global g_num
  19. for i in range(num):
  20. mutex.acquire() # 上锁
  21. g_num += 1
  22. mutex.release() # 解锁
  24. print(f"【子线程work2】 g_num={g_num}")
  27. # 创建一个互斥锁
  28. # 默认是未上锁的状态
  29. mutex = threading.Lock()
  31. # 创建2个线程，让他们各自对g_num加1000000次
  32. p1 = threading.Thread(target=work1, args=(1000000,))
  33. p2 = threading.Thread(target=work2, args=(1000000,))
  34. p1.start()
  35. p2.start()
  37. # 等待计算完成
  38. while len(threading.enumerate()) != 1:
  39. time.sleep(1)
  41. print("【主线程】 2个线程对同一个全局变量操作之后的最终结果是:%s" % g_num)
  复制代码
  总结：
  - 锁的长处：确保了某段尖键代码只能由一个线程从头到尾完整地执行
  - 锁的坏处：
    - 制止了多线程并发执行﹐包含锁的某段代码实际上只能以单线程模式执行﹐效率就大大地下降了
    - 由于可以存在多个锁﹐不同的线程持有不同的锁﹐并试图获取对方持有的锁时﹐大概会造成死锁
  - 死锁: 在线程间共享多个资源的时间，如果两个线程分别占据一部分资源并且同时等候对方的资源，就会造成死锁。
  - 避免死锁:
    - 程序计划时要只管避免（银行家算法）
    - 添加超时时间等
线程间通信
- 使用队列和多线程实现生产者消费者模型
  1. from queue import Queue
  2. import random
  3. import time
  4. import threading
  7. class Producer(threading.Thread):
  9. def run(self):
  10. for value in ['A', 'B', 'C', 'D', 'E', 'F']:
  11. print(f'【Producer】 put {value} to queue...')
  12. q.put(value)
  13. # 设置队列堵塞,只有消费者完成任务并执行task_done 计算器计数为0才会解堵塞
  14. # q.join()
  15. time.sleep(random.random())
  16. # 发送结束信号
  17. q.put(None)
  18. print("生产者任务结束！")
  21. class Consumer(threading.Thread):
  23. def run(self):
  24. while True:
  25. if not q.empty():
  26. data = q.get()
  27. # 任务完成队列计数器减一
  28. # q.task_done()
  29. # 接收到结束信号退出程序
  30. if data is None:
  31. return
  32. print(f"【Consumer】 get {data} from queue")
  33. time.sleep(random.random())
  36. q = Queue()
  37. Producer().start()
  38. Consumer().start()
  复制代码
线程池ThreadPool

线程池是一个线程管理技术，创建一个大概多个线程举行管理，避免线程的创建和销毁带来的开销线程过多会带来调度开销，进而影响缓存局部性和整体性能。而线程池维护着多个线程，等候着监视管理者分配可并发执行的使命。这避免了在处理短时间使命时创建与销毁线程的代价。线程池不仅能够保证内核的充实利用，还能防止过分调度.
- 线程池的长处
  - 降低资源消耗；通过重复利用已创建的线程降低创建和销毁造成的消耗。
  - 进步相应速度，不必等候线程的创建，正常环境下(没有使命进入队列的环境)不必要等候。
  - 线程管理，线程统一由线程池管理，随取随用。
- 自定义线程池实现
  1. from threading import Thread
  2. from queue import Queue
  3. import time
  4. import random
  5. import threading
  8. # 自定义线程池
  9. class MyThreadPool:
  11. def __init__(self, thread_num):
  12. self.thread_num = thread_num
  13. self.task_queue = Queue()
  15. # 初始化的时候启动线程池
  16. self.__start()
  18. # 依次创建并启动线程
  19. def __start(self):
  20. for _ in range(self.thread_num):
  21. # 每个线程的执行相同的方法，在方法中读取队列的任务
  22. # 这里daemon=True 表示当主线程运行结束时不对这个子线程进行检查而直接退出(实现主线程退出时,关闭线程池中子线程的任务执行)
  23. # 参考daemon在多线程中的作用: https://blog.51cto.com/u_9653244/6450770
  24. Thread(target=self._target, daemon=True).start()
  26. # 为线程分配任务
  27. def _target(self):
  28. while True:
  29. target, args, kwargs = self.task_queue.get()
  30. target(*args, **kwargs)
  31. # 队列计数器减一
  32. self.task_queue.task_done()
  34. # 任务队列执行完成之前一直堵塞
  35. def join(self):
  36. self.task_queue.join()
  38. # 往队列中添加任务
  39. def submit_task(self, target, args=(), kwargs=None):
  40. if kwargs is None:
  41. kwargs = {}
  42. self.task_queue.put((target, args, kwargs))
  45. def work(name, no, **kwargs):
  46. thread_id = threading.get_ident()
  47. print(f"【子线程{thread_id}-开始】({threading.current_thread()}) ")
  48. print(f"【子线程{thread_id}-data】 name={name}, no={no}, kwargs={kwargs}")
  49. time.sleep(random.randint(1, 10))
  50. print(f"【子线程{thread_id}-结束】")
  53. t_pool = MyThreadPool(3)
  54. for i in range(5):
  55. t_pool.submit_task(work, (f"任务{i}", i), {"test": 1})
  57. t_pool.join()
  复制代码
- 内置线程池模块 ThreadPool
  - 模块分析
    from multiprocessing.pool import ThreadPool
    # multiprocessing.dummy.Pool为一个函数，本质是使用ThreadPool创建线程池
    from multiprocessing.dummy import Pool as TPool
    print(ThreadPool, TPool)
    print(ThreadPool().__class__, TPool().__class__)
    print(ThreadPool().__class__ is TPool().__class__)
    复制代码
  - 使用内置线程池ThreadPool 模块实现线程复用
    from multiprocessing.pool import ThreadPool
    import time
    import random
    import threading
    def work(name, no, **kwargs):
    thread_id = threading.get_ident()
    print(f"【子线程{thread_id}-开始】({threading.current_thread()}) ")
    print(f"【子线程{thread_id}-data】 name={name}, no={no}, kwargs={kwargs}")
    time.sleep(random.randint(1, 10))
    print(f"【子线程{thread_id}-结束】")
    t_pool = ThreadPool(3)
    for i in range(5):
    t_pool.apply_async(work, (f"任务{i}", i), {"test": 1})
    # 关闭ThreadPool，使其不再接受新的任务；
    t_pool.close()
    t_pool.join()
    复制代码

四、进程线程的等候/终止和保卫模式

进程的等候和终止
1. from multiprocessing import Process
2. import time
5. def func():
6. while True:
7. print("【子进程】")
8. time.sleep(3)
11. if __name__ == '__main__':
12. print("【主进程】")
13. p = Process(target=func)
14. p.start()
16. # 主进程等待子进程完成
17. # p.join()
18. # 立即结束子进程,不推荐使用，会导致子进程的资源无法被释放
19. # p.terminate()
20. print("【主进程】结束")
复制代码
线程的等候和终止
- 线程的等候
  1. from threading import Thread
  2. import time
  5. def func():
  6. while True:
  7. print("【子线程】")
  8. time.sleep(3)
  11. if __name__ == '__main__':
  12. print("【主线程】")
  13. t = Thread(target=func)
  14. t.start()
  16. # 主线程等待子线程完成再继续往后执行
  17. # t.join()
  19. print("【主线程】结束")
  复制代码
  总结：
  - 主线程代码执行完成以后默认等候子线程；所有子线程结束以后程序才会结束
  - t.join() 主线程在某个位置等候子线程执行完成, 再继承执行
- 线程的终止
  - 一. 使用 t1.stop()方法强行终止线程
    跟进程的terminate一样,不推荐使用;目前该方法已被弃用，会导致被终止的线程所拥有的资源如打开的文件、数据库事务等不能被正确释放；造成数据泄漏或死锁，除非可以肯定数据安全，否则不建议强行杀死线程
    复制代码
  - 二. 通过抛出异常来终止线程
    比较复杂一般不用
    复制代码
  - 三.通过一个终止标志来终止线程
    - 实现方式1：
      # 方式一: 使用全局变量stop_threads控制线程终止;
      # 这里子线程直接访问主线程中的全局变量会导致数据混乱，不推荐
      from threading import Thread
      import time
      def func():
      while True:
            print("【子线程】")
            time.sleep(3)
            if stop_threads:
               return
      if __name__ == '__main__':
      print("【主线程】")
      stop_threads = False
      t = Thread(target=func)
      t.start()
      time.sleep(10)
      stop_threads = True
      print("【主线程】结束")
      复制代码
    - 实现方式2：
      # 方式二：通过函数间接访问局部变量stop_threads，推荐使用
      # 好处：通过在子线程中指定位置设置检测点，可以在主线程中任何时候终止子线程，并且子线程所拥有的资源也能被正确释放
      from threading import Thread
      import time
      def func(get_stop_flag):
      while True:
            print("【子线程】")
            time.sleep(3)
            if get_stop_flag():
               return
      if __name__ == '__main__':
      print("【主线程】")
      stop_threads = False
      t = Thread(target=func, args=(lambda: stop_threads,))
      t.start()
      time.sleep(10)
      stop_threads = True
      print("【主线程】结束")
      复制代码
  - 四. 将子线程设置为保卫线程
    不能指定子线程终止时间, 接下来我们就说下守护模式
    复制代码
保卫模式
- 保卫进程概念
  
  随着主进程代码执行结束，保卫进程结束, 可以理解为子进程开启保卫进程以后，主进程为子进程的运行保驾护航；当主进程结束，没有保卫以后，子进程立刻就会结束；
- 保卫进程代码示例
  1. from multiprocessing import Process
  2. import time
  5. def func():
  6. while True:
  7. print("【子进程】")
  8. time.sleep(3)
  11. if __name__ == '__main__':
  12. print("【主进程】")
  13. # 设置p为守护进程, 创建时设置
  14. # p = Process(target=func, daemon=True)
  15. p = Process(target=func)
  16. print(f"【主进程-创建进程后】子进程是否正在运行: {p.is_alive()}")
  18. # 设置p为守护进程, 启动进程前设置(默认p.daemon = False)
  19. p.daemon = True
  20. p.start()
  21. print(f"【主进程-启动进程后】子进程是否正在运行: {p.is_alive()}")
  23. print("【主进程】结束")
  复制代码
- 保卫线程概念
  
  在主线程代码执行结束后，等候其它非保卫子线程执行结束，保卫线程立即结束；即主线程只会等候非保卫线程结束，不会等候保卫线程执行完毕，只要主线程代码执行结束，保卫线程就会结束。
- 保卫线程代码示例
  1. from threading import Thread
  2. import time
  5. def func():
  6. while True:
  7. print("【子线程】")
  8. time.sleep(3)
  11. if __name__ == '__main__':
  12. print("【主线程】")
  13. # 设置t为守护线程, 创建时设置
  14. # t = Thread(target=func, daemon=True)
  15. t = Thread(target=func)
  16. print(f"【主线程-创建线程后】子线程是否正在运行: {t.is_alive()}")
  18. # 设置t为守护线程, 启动线程前设置(默认t.daemon = False)
  19. t.daemon = True
  20. # 等效于 t.daemon = True
  21. # t.setDaemon(True)
  23. t.start()
  24. print(f"【主线程-创建线程后】子线程是否正在运行: {t.is_alive()}")
  26. print("【主线程】结束")
  复制代码

五、多进程多线程队列综合演练

进程池线程池高性能并发通信
1. # 【本机环境运行】
2. # 导入进程池
3. from multiprocessing import Pool, cpu_count
4. # 导入线程池
5. from multiprocessing.pool import ThreadPool
6. from socket import *
7. from queue import Queue
8. import os
11. # 从队列读取数据并返回给客户端
12. def send_data(client, addr, q):
13. # 子进程中无法使用input,而且子进程错误不会展示
14. print(f"【send_data】准备向客户{addr}发送数据...")
15. while True:
16. msg = q.get()
17. if not msg:
18. print(f"【send_data】收到关闭通知, 发送功能关闭！")
19. return
21. client.send(f"您的消息【{msg}】已收到, over ！".encode("gbk"))
24. # 收到客户发来的数据存储到队列
25. def recv_data(client, addr, q):
26. print(f"【recv_data】准备接收客户{addr}的数据...")
27. while True:
28. data = client.recv(1024).decode('gbk')
29. q.put(data)
30. # 客户端调用close; data为 '' (网络调试助手需要关闭通讯窗口才会调用close)
31. if not data:
32. # 往队列写入None,通知发送消息的子线程关闭,并关闭服务套接字
33. q.put('')
34. client.close()
35. print(f"【recv_data】客户{addr}关闭连接, 接收功能关闭！")
36. return
38. print(f"【recv_data】 {addr} 发来消息 : {data}\n")
41. # 进程负责处理连接请求(一个进程跟进一个客户)
42. def process_connect(client, addr):
43. print(f"由进程 {os.getpid()} 为新客户 {addr} 服务！")
44. # 线程负责处理数据请求(一个线程处理客户的一个需求)
45. t_pool = ThreadPool(2)
46. # 创建一个队列，为接收和发送之间传递消息
47. q = Queue()
48. t_pool.apply_async(send_data, (client, addr, q))
49. t_pool.apply_async(recv_data, (client, addr, q))
52. def main():
53. # 创建tcp监听套接字
54. tcp_server_socket = socket(AF_INET, SOCK_STREAM)
55. tcp_server_socket.bind(("127.0.0.1", 9000))
56. tcp_server_socket.listen(128)
58. # 进程池负责接收连接请求(进程池数与cpu处理器数量一致)
59. pool = Pool(cpu_count())
60. while True:
61. # 等待连接请求,获取服务套接字
62. client_socket, client_addr = tcp_server_socket.accept()
63. pool.apply_async(process_connect, (client_socket, client_addr))
65. if __name__ == '__main__':
66. main()
复制代码

六、GIL全局解释器锁

基本概念

GIL 是python的全局解释器锁，同一进程中假如有多个线程运行，一个线程在运行python程序的时候会霸占python解释器（加了一把锁即GIL），使该进程内的其他线程无法运行，等该线程运行完后其他线程才能运行。如果线程运行过程中遇到耗时操作，则解释器锁解开，使其他线程运行。所以在多线程中，线程的运行仍是有先后顺序的，并不是同时进行。
我们可以把GIL看作是“通行证”，并且在一个python进程中，GIL只有一个。拿不到通行证的线程，就不允许进入CPU执行。GIL只在cpython中才有，即同一个进程下的多个线程无法利用多核优势。

复制代码

互斥锁和GIL全局解释器锁的区别
- 互斥锁就是对共享数据举行锁定，保证同一时刻只有一个线程操作数据，是数据级别的锁。
- GIL锁是解释器级别的锁，保证同一时刻下同一个进程中只有一个线程拿到GIL锁，拥有执行
  权限。
关于GIL全局解释器锁的分析
- Python语言和GIL没有半毛钱关系。仅仅是由于历史原因在Cpython假造机(解释器)，难以移除GIL。更换其他解释器就不会存在GIL
- Python使用多进程是可以利用多核的CPU资源。
- 为什么不删除GIL-Guido的声明

七、进程线程对比

多进程和多线程的关系 进程与线程的一个简单解释
定义不同
- 进程是体系举行资源分配和调度的一个独立单位.
- 线程CPU调度和分派的基本单位
功能对比
- 进程，能够完成多使命，好比在一台电脑上能够同时运行多个QQ
- 线程，能够完成多使命，好比一个QQ中的多个聊天窗口
区别
- 一个程序至少有一个进程,一个进程至少有一个线程.
- 线程的分别尺度小于进程(资源比进程少)，使得多线程程序的并发性高。
  - 进程在执行过程中拥有独立的内存单位，而多个线程共享内存，从而极大地进步了程序的运行效率
  - 线线程不能够独立执行，必须依存在进程中
  - 可以将进程理解为工厂中的一条流水线，而其中的线程就是这个流水线上的工人
优缺点
- 线程和进程在使用上各有优缺点：线程执行开销小，但不利于资源的管理和保护；而进程正相反。
进程线程选择
- 盘算麋集型: 多进程；
  - IO麋集型: 多线程、协程
其他问题
- IO麋集型中多线程与协程的执行速度
  1. IO密集型执行时间主要在IO读写,python中由于GIL锁的原因,多线程其实还是使用的单核在进行cpu计算,如果计算任务加锁了，cpu时间片调度机制会在一个cpu时间片（python默认是处理完1000个字节码）结束后，去释放GIL锁，并查看其他线程是否可以执行，由于任务被加锁，会在第二个cpu时间片继续把时间片分给第一个线程，这会让cpu调度时间白白浪费，反而导致多线程比协程(遇到耗时操作自动切换任务)耗时更久
  复制代码
- 盘算麋集型中多线程与单线程的执行速度
  1. 计算量小的情况下单线程快,因为多线程切换需要时间
  2. 计算量大的情况下多线程快,多线程会获得更多的CPU执行时间
  复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/)