多任务进程与线程
一、多任务介绍
我们生活中有很多事情是同时进行的，比如开车的时候 手和脚共同来驾驶汽车，再比如唱歌跳舞也是同时进行的；用程序来模拟：
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类来代表一个进程对象，这个对象可以理解为是一个独立的进程，可以执行另外的事情
-
进程的简单实现
from multiprocessing import Process import time def func(): while True: print("【子进程】") time.sleep(3) if __name__ == '__main__': p = Process(target=func) p.start() while True: print("【主进程】") time.sleep(3)总结：创建子进程时，只需要传入一个执行函数和函数的参数，创建一个Process实例，用start()方法启动
-
进程传递参数与进程对象和进程ID
import multiprocessing import os from time import sleep def func(name, age, **kwargs): print(f"【子进程】({multiprocessing.current_process()}) 的进程号为:{os.getpid()}") print(f"【子进程】 name={name}, age={age}, kwargs={kwargs}") for i in range(10): print(f"【子进程】 ---{i}---") sleep(0.2) if __name__ == '__main__': p = multiprocessing.Process(target=func, args=('test', 18), kwargs={"m": 20}) p.start() sleep(1) # 主进程会等待所有的子进程执行结束再结束 print("【主进程】--- 结束 ---") -
使用多进程实现
UDP通信同时收发数据import socket from multiprocessing import Process def recv_data(udp_socket): while True: recv_msg = udp_socket.recvfrom(1024) data = recv_msg[0].decode("gbk") source = recv_msg[1] print(f"{source}: {data}") def main(): udp_client = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) udp_client.bind(("", 9000)) # 子进程接收数据 Process(target=recv_data, args=(udp_client,)).start() # 主进程发送数据 (子进程中无法使用input) while True: msg = input("请输入发送的数据： ").encode("gbk") dest_addr = ("127.0.0.1", 8000) udp_client.sendto(msg, dest_addr) if __name__ == "__main__": main() -
进程间不共享全局变量
from multiprocessing import Process import os import time nums = [11, 22] def work1(): print(f"【子进程work1】 pid={os.getpid()} ,nums={nums}") for i in range(3): nums.append(i) time.sleep(1) print(f"【子进程work1】 pid={os.getpid()} ,nums={nums}") print("【子进程work1】", id(nums), nums) def work2(): print(f"【子进程work2】 pid={os.getpid()} ,nums={nums}") print("【子进程work2】", id(nums), nums) print(f"当前进程为{os.getpid()}, 进程中nums的id为{id(nums)}") if __name__ == '__main__': p1 = Process(target=work1) p1.start() time.sleep(5) p2 = Process(target=work2) p2.start() 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,只能表示队列中没有任务了，不能保证任务已经执行完成 。
-
-
使用队列和多进程实现生产者消费者模型
from multiprocessing import Process, Queue import time import random def producer(q): for value in ['A', 'B', 'C', 'D', 'E', 'F']: print(f'【Producer】 put {value} to queue...') q.put(value) time.sleep(random.random()) # 发送结束信号 q.put(None) def consumer(q): while True: if not q.empty(): data = q.get() # 接收到结束信号退出程序 if data is None: return print(f"【Consumer】 get {data} from queue") time.sleep(random.random()) q = Queue() if __name__ == "__main__": pw = Process(target=producer, args=(q,)) pr = Process(target=consumer, args=(q,)) pw.start() pr.start()
-
-
进程池Pool
当需要创建的子进程数量不多时，可以直接利用multiprocessing中的Process动态成生多个进程，但如果是上百甚至上千个目标，手动的去创建/销毁进程的工作量巨大，此时就可以用到multiprocessing模块提供的Pool方法
初始化Pool时，可以指定一个最大进程数，当有新的请求提交到Pool中时，如果池还没有满，那么就会创建一个新的进程用来执行该请求；但如果池中的进程数已经达到指定的最大值，那么该请求就会等待，直到池中有进程结束，才会用之前的进程来执行新的任务from multiprocessing import Pool import time import random import os def work(msg): start_time = time.time() print(f"任务{msg} 开始执行,进程号 {os.getpid()}") time.sleep(random.random() * 2) end_time = time.time() print(f"任务{msg} 结束执行,运行时间{end_time - start_time}") if __name__ == "__main__": p = Pool(3) for i in range(10): p.apply_async(work, args=(i,)) # 观察进程池任务什么时候开始执行 time.sleep(1) print("------start--------") # 关闭Pool，使其不再接受新的任务； p.close() # 主进程阻塞，等待子进程的退出， 必须在close或terminate之后使用； # p.join() print("--------end--------")总结：
- 主进程不会主动等待进程池任务执行,如果主进程执行完毕,进程池任务立即结束
- 程池在定义的时候没有指定最大进程数,系统会按当前运行计算机的CPU核心数决定进程池内运行的最大进程数,如计算机为双核,则进程池内最大进程数为2
参数说明(
multiprocessing.Pool)：apply_async(func[, args[, kwds]])：使用非阻塞方式调用func（并行执行，堵塞方式必须等待上一个进程退出才能执行下一个进程），args为传递给func的参数列表，kwds为传递给func的关键字参数列表；- close()：关闭Pool，使其不再接受新的任务；
- terminate()：不管任务是否完成，立即终止；
- join()：主进程阻塞，等待子进程的退出， 必须在close或terminate之后使用；
-
进程池中的Queue
如果要使用Pool创建进程，就需要使用multiprocessing.Manager()中的Queue()，而不是multiprocessing.Queue()-
使用队列和进程池实现生产者消费者模型
from multiprocessing import Manager, Pool import time import random import os def producer(q): print("producer启动({os.getpid()}),父进程为({os.getppid()})") for value in ['A', 'B', 'C', 'D', 'E', 'F']: print(f'Put {value} to queue...') q.put(value) time.sleep(random.random()) # 发送结束信号 q.put(None) def consumer(q): print("consumer启动({os.getpid()}),父进程为({os.getppid()})") while True: if not q.empty(): data = q.get() # 接收到结束信号退出程序 if data is None: return print(f"get {data} from queue") time.sleep(random.random()) if __name__ == "__main__": q = Manager().Queue() p = Pool() p.apply_async(producer, (q,)) p.apply_async(consumer, (q,)) p.close() p.join() print(f"主进程{os.getpid()} 结束")
-
3> 进程的理解
-
进程执行过程分析
""" 说明： 这里 if __name__ == "__main__" 的执行过程 1. 主进程从上到下执行 2. pw.start() 启动子进程时，操作系统会为该子进程pw重新加载所有资源，此时__name__ 值为 __mp_main__ (我们可以通过判断__name__的值来给所有子进程传参) 3. 如果资源定义在 __name__ == "__main__" 内部，则子进程无法获取到 """ from multiprocessing import Process, Queue import time import os import random print(f"进程 {os.getpid()} 开始加载资源...") def write(): for value in ['A', 'B', 'C']: print('Put %s to queue...' % value) q.put(value) time.sleep(random.random()) def read(): while True: if not q.empty(): data = q.get() print("get {} from queue".format(data)) time.sleep(random.randint(0, 5)) else: continue print(111, __name__) if __name__ == "__main__": print(222, __name__) q = Queue() pw = Process(target=write) pr = Process(target=read) print(333, __name__) pw.start() pr.start() pw.join() pr.join() # 运行结果: """ 进程 7244 开始加载资源... 111 __main__ 222 __main__ 333 __main__ 进程 8848 开始加载资源... 111 __mp_main__ 进程 4596 开始加载资源... 111 __mp_main__ Put A to queue... Process Process-1: Process Process-2: NameError: name 'q' is not defined 结果分析: 队列是主进程资源,子进程无法访问 队列必须在进程函数定义之前定义，函数只能访问定义前或则传递过来的参数 通过传参的方式接收的参数是变量的引用，而全局参数在进程内部无法修改变量值，在进程内部修改的是一个新的变量 """ -
总结
-
多进程执行过程-在创建进程后,操作系统会给新创建的这个进程启动后(调用start)拷贝一份运行代码(实际上是写时拷贝,只有运行过程中修改了运行代码才会真正的进行拷贝,没有修改运行代码实际是公用一份运行代码),之后每个进程(包括主进程)内数据都是独立的,即进程间数据是不共享的.
-
拷贝的代码只是在进程创建的时候,由主进程分配给子进程的任务代码和传递的参数,而主进程所拥有的资源子进程并不是全部拥有
-
进程内出现异常会报错,但进程池中的进程异常不会产生异常
-
进程通信:在多进程编程中，不同的进程之间需要进行通信。multiprocessing模块提供了多种进程间通信的方式，例如使用队列、管道、共享内存等。Python编程之多进程(multiprocessing)详解
-
三、线程
1> 基本概念
- 线程：线程是进程的一个实体,是CPU调度和分派的基本单位,它是比进程更小的能独立运行的基本单位.线程自己基本上不拥有系统资源,只拥有一点在运行中必不可少的资源(如程序计数器,一组寄存器和栈),但是它可与同属一个进程的其他的线程共享进程所拥有的全部资源.
2> 线程使用和特性
-
线程的简单实现与线程标识符
import threading from time import sleep, ctime def sing(): print(f"\n【子线程sing】({threading.current_thread()})的标识符为: {threading.get_ident()}") for i in range(3): print(f"正在唱歌...{i}") sleep(3) print(f"【子线程sing】--- 结束 ---") def dance(): print(f"\n【子线程dance】({threading.current_thread()})的标识符为: {threading.get_ident()}") for i in range(3): print(f"正在跳舞...{i}") sleep(3) print(f"【子线程dance】--- 结束 ---") if __name__ == '__main__': print(f"【主线程】---开始---:{ctime()}") t1 = threading.Thread(target=sing) t2 = threading.Thread(target=dance) t1.start() t2.start() print(f"【主线程】子线程sing的标识符: {t1.ident}") print(f"【主线程】子线程dance的标识符: {t2.ident}") length = len(threading.enumerate()) print(f'【主线程】当前运行的线程数为：{length}') print(f"【主线程】--- 结束 ---:{ctime()}")总结：
- 多线程并发的操作比单线程效率更高
-
线程间共享全局变量
from threading import Thread import time nums = [11, 22] def work1(): for i in range(3): nums.append(i) print(f"【子线程work1】 g_num={nums}") def work2(): print(f"【子线程work2】 g_num={nums}") print(f"【主线程】 子线程创建之前g_num={nums}") t1 = Thread(target=work1) t1.start() # 延时一会，保证t1线程中的任务做完 time.sleep(1) t2 = Thread(target=work2) t2.start()总结：
- 一个进程内的所有线程共享全局变量，很方便在多个线程间共享数据
- 缺点就是，线程是对全局变量随意遂改可能造成多线程之间对全局变量的混乱
-
多线程资源竞争问题
import threading import time g_num = 0 def work1(num): global g_num for i in range(num): g_num += 1 print(f"【子线程work1】 g_num={g_num}") def work2(num): global g_num for i in range(num): g_num += 1 print(f"【子线程work2】 g_num={g_num}") print(f"【主线程】 子线程创建之前g_num={g_num}") t1 = threading.Thread(target=work1, args=(1000000,)) t2 = threading.Thread(target=work2, args=(1000000,)) t1.start() t2.start() while len(threading.enumerate()) != 1: time.sleep(1) print(f"【主线程】2个线程对同一个全局变量操作之后的最终结果是:{g_num}")-
结果分析：
假设两个线程t1和t2都要对全局变量g_num(默认是0)进行加1运算，t1和t2都各对`g_num`加10次，g_num的最终的结果应该为20。 但是由于是多线程同时操作，有可能出现下面情况： 1> 在g_num=0时，t1取得g_num=0。此时系统把t1调度为”sleeping”状态，把t2转换为”running”状态，t2也获得g_num=0 2> 然后t2对得到的值进行加1并赋给g_num，使得g_num=1 3> 然后系统又把t2调度为”sleeping”，把t1转为”running”。线程t1又把它之前得到的0加1后赋值给g_num。 4> 这样导致虽然t1和t2都对g_num加1，但结果仍然是g_num=1
-
-
解决多线程资源竞争问题 - 互斥锁
-
互斥锁
解决多线程资源竞争问题，最简单的机制就是引入互斥锁，互斥锁为资源引入一个状态：锁定/非锁定
-
互斥锁工作原理
某个线程要更改共享数据时，先将其锁定，此时资源的状态为“锁定”，其他线程不能更改；直到该线程释放资源，将资源的状态变成“非锁定”，其他的线程才能再次锁定该资源。互斥锁保证了每次只有一个线程进行写入操作，从而保证了多线程情况下数据的正确性。
-
互斥锁使用
# 创建锁 mutex = threading.Lock() # 锁定 mutex.acquire() # 释放 mutex.release()说明：
- 如果这个锁之前是没有上锁的，那么acquire不会堵塞
- 如果在调用acquire对这个锁上锁之前 它已经被 其他线程上了锁，那么此时acquire会堵塞，直到这个锁被解锁为止
-
为全局变量加入互斥锁
import threading import time g_num = 0 def work1(num): global g_num for i in range(num): mutex.acquire() # 上锁 g_num += 1 mutex.release() # 解锁 print(f"【子线程work1】 g_num={g_num}") def work2(num): global g_num for i in range(num): mutex.acquire() # 上锁 g_num += 1 mutex.release() # 解锁 print(f"【子线程work2】 g_num={g_num}") # 创建一个互斥锁 # 默认是未上锁的状态 mutex = threading.Lock() # 创建2个线程，让他们各自对g_num加1000000次 p1 = threading.Thread(target=work1, args=(1000000,)) p2 = threading.Thread(target=work2, args=(1000000,)) p1.start() p2.start() # 等待计算完成 while len(threading.enumerate()) != 1: time.sleep(1) print("【主线程】 2个线程对同一个全局变量操作之后的最终结果是:%s" % g_num)总结：
- 锁的好处：确保了某段关键代码只能由一个线程从头到尾完整地执行
- 锁的坏处：
- 阻止了多线程并发执行﹐包含锁的某段代码实际上只能以单线程模式执行﹐效率就大大地下降了
- 由于可以存在多个锁﹐不同的线程持有不同的锁﹐并试图获取对方持有的锁时﹐可能会造成死锁
- 死锁: 在线程间共享多个资源的时候，如果两个线程分别占有一部分资源并且同时等待对方的资源，就会造成死锁。
- 避免死锁:
- 程序设计时要尽量避免（银行家算法）
- 添加超时时间等
-
-
线程间通信
-
使用队列和多线程实现生产者消费者模型
from queue import Queue import random import time import threading class Producer(threading.Thread): def run(self): for value in ['A', 'B', 'C', 'D', 'E', 'F']: print(f'【Producer】 put {value} to queue...') q.put(value) # 设置队列堵塞,只有消费者完成任务并执行task_done 计算器计数为0才会解堵塞 # 可以监控消费者任务执行情况 # q.join() time.sleep(random.random()) # 发送结束信号 q.put(None) print("生产者任务结束！") class Consumer(threading.Thread): def run(self): while True: if not q.empty(): data = q.get() # 任务完成 队列计数器减一 # q.task_done() # 接收到结束信号退出程序 if data is None: break print(f"【Consumer】 get {data} from queue") time.sleep(random.random()) print("消费者任务结束！") q = Queue() Producer().start() Consumer().start()
-
-
线程池
ThreadPool 线程池是一个线程管理技术，创建一个或者多个线程进行管理，避免线程的创建和销毁带来的开销线程过多会带来调度开销，进而影响缓存局部性和整体性能。而线程池维护着多个线程，等待着监督管理者分配可并发执行的任务。这避免了在处理短时间任务时创建与销毁线程的代价。线程池不仅能够保证内核的充分利用，还能防止过分调度.
-
线程池的优点
- 降低资源消耗；通过重复利用已创建的线程降低创建和销毁造成的消耗。
- 提高响应速度，不必等待线程的创建，正常情况下(没有任务进入队列的情况)不需要等待。
- 线程管理，线程统一由线程池管理，随取随用。
-
自定义线程池实现
from threading import Thread from queue import Queue import time import random import threading # 自定义线程池 class MyThreadPool: def __init__(self, thread_num): self.thread_num = thread_num self.task_queue = Queue() # 初始化的时候启动线程池 self.__start() # 依次创建并启动线程 def __start(self): for _ in range(self.thread_num): # 每个线程的执行相同的方法，在方法中读取队列的任务 # 这里daemon=True 表示当主线程运行结束时不对这个子线程进行检查而直接退出(实现主线程退出时,关闭线程池中子线程的任务执行) # 参考daemon在多线程中的作用: https://blog.51cto.com/u_9653244/6450770 Thread(target=self._target, daemon=True).start() # 为线程分配任务 def _target(self): while True: target, args, kwargs = self.task_queue.get() target(*args, **kwargs) # 队列计数器减一 self.task_queue.task_done() # 任务队列执行完成之前一直堵塞 def join(self): self.task_queue.join() # 往队列中添加任务 def submit_task(self, target, args=(), kwargs=None): if kwargs is None: kwargs = {} self.task_queue.put((target, args, kwargs)) 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 = MyThreadPool(3) for i in range(5): t_pool.submit_task(work, (f"任务{i}", i), {"test": 1}) 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()
-
-
四、进程线程的等待/终止 和 守护模式
-
进程的等待和终止
from multiprocessing import Process import time def func(): while True: print("【子进程】") time.sleep(3) if __name__ == '__main__': print("【主进程】") p = Process(target=func) p.start() # 主进程等待子进程完成 # p.join() # 立即结束子进程,不推荐使用，会导致子进程的资源无法被释放 # p.terminate() print("【主进程】 结束") -
线程的等待和终止
-
线程的等待
from threading import Thread import time def func(): while True: print("【子线程】") time.sleep(3) if __name__ == '__main__': print("【主线程】") t = Thread(target=func) t.start() # 主线程等待子线程完成再继续往后执行 # t.join() 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("【主线程】 结束")
-
-
四. 将子线程设置为守护线程
不能指定子线程终止时间, 接下来我们就说下守护模式
-
-
-
守护模式
-
守护进程概念
随着主进程代码执行结束，守护进程结束, 可以理解为子进程开启守护进程以后，主进程为子进程的运行保驾护航；当主进程结束，没有守护以后，子进程立刻就会结束；
-
守护进程代码示例
from multiprocessing import Process import time def func(): while True: print("【子进程】") time.sleep(3) if __name__ == '__main__': print("【主进程】") # 设置p为守护进程, 创建时设置 # p = Process(target=func, daemon=True) p = Process(target=func) print(f"【主进程-创建进程后】子进程是否正在运行: {p.is_alive()}") # 设置p为守护进程, 启动进程前设置(默认p.daemon = False) p.daemon = True p.start() print(f"【主进程-启动进程后】子进程是否正在运行: {p.is_alive()}") print("【主进程】 结束")
-
-
守护线程概念
在主线程代码执行结束后，等待其它非守护子线程执行结束，守护线程立即结束；即主线程只会等待非守护线程结束，不会等待守护线程执行完毕，只要主线程代码执行结束，守护线程就会结束。
-
守护线程代码示例
from threading import Thread import time def func(): while True: print("【子线程】") time.sleep(3) if __name__ == '__main__': print("【主线程】") # 设置t为守护线程, 创建时设置 # t = Thread(target=func, daemon=True) t = Thread(target=func) print(f"【主线程-创建线程后】子线程是否正在运行: {t.is_alive()}") # 设置t为守护线程, 启动线程前设置(默认t.daemon = False) t.daemon = True # 与 t.daemon = True 等效 # t.setDaemon(True) t.start() print(f"【主线程-创建线程后】子线程是否正在运行: {t.is_alive()}") print("【主线程】 结束")
-
五、多进程多线程队列综合演练
-
进程池线程池高性能并发通信
# 【本机环境运行】 # 导入进程池 from multiprocessing import Pool, cpu_count # 导入线程池 from multiprocessing.pool import ThreadPool from socket import * from queue import Queue import os # 从队列读取数据并返回给客户端 def send_data(client, addr, q): # 子进程中无法直接使用input,而且子进程错误不会展示 print(f"【send_data】准备向客户{addr}发送数据...") while True: msg = q.get() if not msg: print(f"【send_data】收到关闭通知, 发送功能关闭！") return client.send(f"您的消息 【{msg}】 已收到, over ！".encode("gbk")) # 收到客户发来的数据存储到队列 def recv_data(client, addr, q): print(f"【recv_data】准备接收客户{addr}的数据...") while True: data = client.recv(1024).decode('gbk') q.put(data) # 客户端调用close; data为 '' (网络调试助手需要关闭通讯窗口才会调用close) if not data: # 往队列写入None,通知发送消息的子线程关闭,并关闭服务套接字 q.put('') client.close() print(f"【recv_data】客户{addr}关闭连接, 接收功能关闭！") return print(f"【recv_data】 {addr} 发来消息 : {data}\n") # 进程负责处理连接请求(一个进程跟进一个客户) def process_connect(client, addr): print(f"由进程 {os.getpid()} 为新客户 {addr} 服务！") # 线程负责处理数据请求(一个线程处理客户的一个需求) t_pool = ThreadPool(2) # 创建一个队列，为接收和发送之间传递消息 q = Queue() t_pool.apply_async(send_data, (client, addr, q)) t_pool.apply_async(recv_data, (client, addr, q)) def main(): # 创建tcp监听套接字 tcp_server_socket = socket(AF_INET, SOCK_STREAM) tcp_server_socket.bind(("127.0.0.1", 9000)) tcp_server_socket.listen(128) # 进程池负责接收连接请求(进程池数与cpu处理器数量一致) pool = Pool(cpu_count()) while True: # 等待连接请求,获取服务套接字 client_socket, client_addr = tcp_server_socket.accept() pool.apply_async(process_connect, (client_socket, client_addr)) if __name__ == '__main__': 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的声明
- Python语言和GIL没有半毛钱关系。仅仅是由于历史原因在
七、进程线程对比
-
多进程和多线程的关系 进程与线程的一个简单解释
-
定义不同
- 进程是系统进行资源分配和调度的一个独立单位.
- 线程CPU调度和分派的基本单位
-
功能对比
- 进程，能够完成多任务，比如 在一台电脑上能够同时运行多个QQ
- 线程，能够完成多任务，比如 一个QQ中的多个聊天窗口
-
区别
- 一个程序至少有一个进程,一个进程至少有一个线程.
- 线程的划分尺度小于进程(资源比进程少)，使得多线程程序的并发性高。
- 进程在执行过程中拥有独立的内存单元，而多个线程共享内存，从而极大地提高了程序的运行效率
- 线线程不能够独立执行，必须依存在进程中
- 可以将进程理解为工厂中的一条流水线，而其中的线程就是这个流水线上的工人
-
优缺点
- 线程和进程在使用上各有优缺点：线程执行开销小，但不利于资源的管理和保护；而进程正相反。
-
进程线程选择
- 计算密集型: 多进程；
- IO密集型: 多线程、协程
- 计算密集型: 多进程；
-
其他问题
-
IO密集型中多线程与协程的执行速度
IO密集型执行时间主要在IO读写,python中由于GIL锁的原因,多线程其实还是使用的单核在进行cpu计算,如果计算任务加锁了，cpu时间片调度机制会在一个cpu时间片（python默认是处理完1000个字节码）结束后，去释放GIL锁，并查看其他线程是否可以执行，由于任务被加锁，会在第二个cpu时间片继续把时间片分给第一个线程，这会让cpu调度时间白白浪费，反而导致多线程比协程(遇到耗时操作自动切换任务)耗时更久 -
计算密集型中多线程与单线程的执行速度
计算量小的情况下单线程快,因为多线程切换需要时间 计算量大的情况下多线程快,多线程会获得更多的CPU执行时间
-