一、多进程实现TCP服务端并发
import socket from multiprocessing import Process def get_server(): server = socket.socket() server.bind(('127.0.0.1', 8080)) server.listen(5) return server def get_talk(sock): while True: data = sock.recv(1024) print(data.decode('utf8')) sock.send(data.upper()) if __name__ == '__main__': server = get_server() while True: sock, addr = server.accept() # 开设多进程去聊天 p = Process(target=get_talk, args=(sock,)) p.start()
二、互斥锁代码实操
1、互斥锁的概念
互斥锁: 对共享数据进行锁定，保证同一时刻只能有一个线程去操作。
注意:
互斥锁是多个线程一起去抢，抢到锁的线程先执行，没有抢到锁的线程需要等待，等互斥锁使用完释放后，其它等待的线程再去抢这个锁。
2、互斥锁的使用
threading模块中定义了Lock变量，这个变量本质上是一个函数，通过调用这个函数可以获取一把互斥锁。
ps:建议只加载操作数据的部分 否则整个程序的效率会极低
from multiprocessing import Process, Lock import time import json import random def search(name): with open(r'data.json', 'r', encoding='utf8') as f: data = json.load(f) print('%s查看票 目前剩余:%s' % (name, data.get('ticket_num'))) def buy(name): # 先查询票数 with open(r'data.json', 'r', encoding='utf8') as f: data = json.load(f) # 模拟网络延迟 time.sleep(random.randint(1, 3)) # 买票 if data.get('ticket_num') > 0: with open(r'data.json', 'w', encoding='utf8') as f: data['ticket_num'] -= 1 json.dump(data, f) print('%s 买票成功' % name) else: print('%s 买票失败 非常可怜 没车回去了!!!' % name) def run(name, mutex): search(name) mutex.acquire() # 抢锁 buy(name) mutex.release() # 释放锁 if __name__ == '__main__': mutex = Lock() # 产生一把锁 for i in range(10): p = Process(target=run, args=('用户%s号' % i, mutex)) p.start() """ 锁有很多种 但是作用都一样 行锁 表锁 ... """
- acquire和release方法之间的代码同一时刻只能有一个线程去操作
- 如果在调用acquire方法的时候 其他线程已经使用了这个互斥锁，那么此时acquire方法会堵塞，直到这个互斥锁释放后才能再次上锁。
3、死锁现象
这里讲的就是前一个线程拿了a的锁然后想去拿b的锁，同时后一个 进程拿了b的锁，想去拿a的锁，双方都拿不到，代码就卡在那里不会动了，这个现象叫做死锁现象。
acquire() release() from threading import Thread,Lock import time mutexA = Lock() # 产生一把锁 mutexB = Lock() # 产生一把锁 class MyThread(Thread): def run(self): self.func1() self.func2() def func1(self): mutexA.acquire() print(f'{self.name}抢到了A锁') mutexB.acquire() print(f'{self.name}抢到了B锁') mutexB.release() print(f'{self.name}释放了B锁') mutexA.release() print(f'{self.name}释放了A锁') def func2(self): mutexB.acquire() print(f'{self.name}抢到了B锁') time.sleep(1) mutexA.acquire() print(f'{self.name}抢到了A锁') mutexA.release() print(f'{self.name}释放了A锁') mutexB.release() print(f'{self.name}释放了B锁') for i in range(10): obj = MyThread() obj.start()
4、 小结
- 互斥锁的作用就是保证同一时刻只能有一个线程去操作共享数据，保证共享数据不会出现错误问题
- 使用互斥锁的好处确保某段关键代码只能由一个线程从头到尾完整地去执行
- 使用互斥锁会影响代码的执行效率，多任务改成了单任务执行
- 互斥锁如果没有使用好容易出现死锁的情况
三、线程理论
进程
- 进程其实是资源单位，表示一块内存空间，就像运行py文件的时候一个py文件就是一个进程。
线程
- 线程才是执行单位，表示真正的代码指令，相当于一个py文件内部的代码，每个进程内部最少有一个线程。
线程简介
- 线程，有时被称为轻量进程(Lightweight Process,LWP)，是程序执行流的最小单元。
- 一个标准的线程由线程ID，当前指令指针(PC），寄存器集合和堆栈组成。
- 另外，线程是进程中的一个实体，是被系统独立调度和分派的基本单位，线程自己不拥有系统资源，只拥有一点儿在运行中必不可少的资源，但它可与同属一个进程的其它线程共享进程所拥有的全部资源。
- 一个线程可以创建和撤消另一个线程，同一进程中的多个线程之间可以并发执行。
- 由于线程之间的相互制约，致使线程在运行中呈现出间断性。线程也有就绪、阻塞和运行三种基本状态。
- 就绪状态是指线程具备运行的所有条件，逻辑上可以运行，在等待处理机;运行状态是指线程占有处理机正在运行;阻塞状态是指线程在等待一个事件（如某个信号量)，逻辑上不可执行。
- 每一个程序都至少有一个线程，若程序只有一个线程，那就是程序本身。
- 线程是程序中一个单一的顺序控制流程。进程内有一个相对独立的、可调度的执行单元，是系统独立调度和分派CPU的基本单位指令运行时的程序的调度单位。
- 在单个程序中同时运行多个线程完成不同的工作，称为多线程。
- 线程是程序中一个单一的顺序控制流程。进程内有一个相对独立的、可调度的执行单元，是系统独立调度和分派CPU的基本单位指令运行时的程序的调度单位。在单个程序中同时运行多个线程完成不同的工作，称为多线程。
我们可以将进程比喻是车间 线程是车间里面的流水线
为什么要使用多线程？
1.线程在程序中是独立的、并发的执行流。与分隔的进程相比，进程中线程之间的隔离程度要小，它们共享内存、文件句柄和其他进程应有的状态。
2.因为线程的划分尺度小于进程，使得多线程程序的并发性高。进程在执行过程之中拥有独立的内存单元，而多个线程共享内存，从而极大的提升了程序的运行效率。
3.线程比进程具有更高的性能，这是由于同一个进程中的线程都有共性，多个线程共享一个进程的虚拟空间。线程的共享环境包括进程代码段、进程的共有数据等，利用这些共享的数据，线程之间很容易实现通信。
4.操作系统在创建进程时，必须为改进程分配独立的内存空间，并分配大量的相关资源，但创建线程则简单得多。因此，使用多线程来实现并发比使用多进程的性能高得要多。
多线程概念
- 多线程:多线程( 英语: multithreading) ，是指从软件或者硬件上实现多个线程并发执行的技术。
- 具有多线程能力的计算机因有硬件支持而能够在同一时间执行多于一个线程，进而提升整体处理性能。
- 具有这种能力的系统包括对称多处理机、多核心处理器以及芯片级多处理( Chip-level multithreading)或同时多线程( Simultaneous multithreading)处理器。
- 在一个程序中，这些独立运行的程序片段叫作“线程”(Thread)，利用它编程的概念就叫作“多线程处理(Multithreading)”。具有多线程能力的计算机因有硬件支持而能够在同一时间执行多于一个线程(台湾译作“执行绪”) ，进而提升整体处理性能
多进程的优点：
- 进程之间不能共享内存，但线程之间共享内存非常容易。
- 操作系统在创建进程时，需要为该进程重新分配系统资源，但创建线程的代价则小得多。因此，使用多线程来实现多任务并发执行比使用多进程的效率高。
- Python 语言内置了多线程功能支持，而不是单纯地作为底层操作系统的调度方式，从而简化了 Python 的多线程编程。
线程与进程的区别
1）地址空间和其它资源（如打开文件）：进程间相互独立，同一进程的各线程间共享。某进程内的线程在其它进程不可见。
2）通信：进程间通信IPC，线程间可以直接读写进程数据段（如全局变量）来进行通信——需要进程同步和互斥手段的辅助，以保证数据的一致性。
3）调度和切换：线程上下文切换比进程上下文切换要快得多。
4）在多线程操作系统中，进程不是一个可执行的实体。
线程的特点
1.一个进程内可以开设多个线程
2.同一个进程下的多个线程数据是共享的
3.创建进程与线程的区别：创建进程的消耗要远远大于线程
四、创建线程的两种方式
跟进程一样，也是函数和面向对象两种方式
创建线程需要用到Threading.Thread类，他和Process模块使用的方法很像
from threading import Thread from multiprocessing import Process import time def task(name): print(f'{name} is running') time.sleep(0.1) print(f'{name} is over') if __name__ == '__main__': start_time = time.time() p_list = [] for i in range(100): p = Process(target=task, args=('用户%s'%i,)) p.start() p_list.append(p) for p in p_list: p.join() print(time.time() - start_time) t_list = [] for i in range(100): t = Thread(target=task, args=('用户%s'%i,)) t.start() t_list.append(t) for t in t_list: t.join() print(time.time() - start_time) t = Thread(target=task, args=('jason',)) t.start() print('主线程') """ 创建线程无需考虑反复执行的问题 """ class MyThread(Thread): def run(self): print('run is running') time.sleep(1) print('run is over') obj = MyThread() obj.start() print('主线程')
五、线程的诸多特性
1.join方法（让线程从异步变成同步）
from threading import Thread import time def task(name): print(f'{name} is running') time.sleep(1) print(f'{name} is over') t = Thread(target=task, args=('jason', )) t.start() t.join() print('主线程')
2.同进程内多个线程数据共享（可以看成一个py文件内的变量是通用的）
3.current_thread().name查看线程的名称
4.active_count()查看主线程的名称
六、GIL全局解释器锁
一、全局解释器锁GIL
GIL：又称全局解释器锁。作用就是限制多线程同时执行，保证同一时间内只有一个线程在执行。线程非独立的，所以同一进程里线程是数据共享，当各个线程访问数据资源时会出现“竞争”状态，即数据可能会同时被多个线程占用，造成数据混乱，这就是线程的不安全。所以引进了互斥锁，确保某段关键代码、共享数据只能由一个线程从头到尾完整地执行。
二、为什么会有GIL
Python为了利用多核CPU，开始支持多线程。而解决多线程之间数据完整性和状态同步的最简单方法自然就是加锁，于是有了GIL这把超级大锁。因为有了GIL，所以我们的Python可以实现多进程，但是这是一个假的多进程，虽然它会利用多个CPU共同协作，但实则是利用一个CPU的资源。
但是这种GIL导致我们的多进程并不是真正的多进程，所以它的效率很低。但当大家试图去拆分和去除GIL的时候，发现大量库代码开发者已经重度依赖GIL而非常难以去除了。如果推到重来，多线程的问题依然还是要面对，但是至少会比目前GIL这种方式会更优雅。所以简单的说：GIL的存在更多的是历史原因。
三、GIL的副作用
Python的多线程在多核CPU上，只对于IO密集型计算产生正面效果；而当有至少有一个CPU密集型线程存在，那么多线程效率会由于GIL而大幅下降。正因为有了GIL的存在，我们Python的多线程效率才会比较低，毕竟它不是真正的多线程。那么此时，我们就可以考虑使用多进程去实现，因为多进程是可以利用多核的CPU资源的。但是又有一个问题？多进程需要的资源较大，明显不是最好的解决办法，那么如何高效的解决这一问题呢？
我们都知道Python它其实是一个“胶水”语言，它除了可以调用自己的模块。类库之外，还可以调用C、C++等语言的很多模块、类库。此时，我们只需加载动态库，把多进程这块，换成利用C语言去实现就可以了
四、GIL的总结
因为GIL的存在，只有IO Bound场景下的多线程会得到较好的性能。
如果对并行计算性能较高的程序可以考虑把核心部分也成C模块，或者索性用其他语言实现。
GIL在较长一段时间内将会继续存在，但是会不断对其进行改进。
官方文档对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.
1.在CPython解释器中存在全局解释器锁简称GIL
python解释器有很多类型
CPython JPython PyPython (常用的是CPython解释器)
2.GIL本质也是一把互斥锁 用来阻止同一个进程内多个线程同时执行(重要)
3.GIL的存在是因为CPython解释器中内存管理不是线程安全的(垃圾回收机制)
垃圾回收机制
引用计数、标记清除、分代回收
当我们在运行python解释器的时候，之所以不能弄成多线程，是因为处于多线程状态下的时候，垃圾回收机制会误删数据值，在数据值还没有和变量名绑定的时候，就会提前把数据值删除，造成异常，因此python中使用的是GIL 来控制线程的运行先后顺序。
七、验证GIL的存在
根据GIL的作用我们得知，线程都是串行运行的，因此这里我们调用time模块让每个线程运行时睡0.1秒，来验证代码运行一百次是否需要10秒以上。
import time from threading import Thread,Lock num = 100 def task(mutex): global num mutex.acquire() count = num time.sleep(0.1) num = count - 1 mutex.release() mutex = Lock() t_list = [] for i in range(100): t = Thread(target=task,args=(mutex,)) t.start() t_list.append(t) for t in t_list: t.join() print(num)
八、GIL与普通互斥锁
既然CPython解释器中有GIL 那么我们以后写代码是不是就不需要操作锁了!!!
回答：GIL只能够确保同进程内多线程数据不会被垃圾回收机制弄乱。并不能确保程序里面的数据是否安全。
import time from threading import Thread,Lock num = 100 def task(mutex): global num mutex.acquire() count = num time.sleep(0.1) num = count - 1 mutex.release() mutex = Lock() t_list = [] for i in range(100): t = Thread(target=task,args=(mutex,)) t.start() t_list.append(t) for t in t_list: t.join() print(num)
九、python多线程是否有用
需要分情况
情况1
单个CPU
多个CPU
情况2
IO密集型(代码有IO操作)
计算密集型(代码没有IO)
1.单个CPU
IO密集型
多进程
申请额外的空间 消耗更多的资源
多线程
消耗资源相对较少 通过多道技术
ps:多线程有优势!!!
计算密集型
多进程
申请额外的空间 消耗更多的资源(总耗时+申请空间+拷贝代码+切换)
多线程
消耗资源相对较少 通过多道技术(总耗时+切换)
ps:多线程有优势!!!
2.多个CPU
IO密集型
多进程
总耗时(单个进程的耗时+IO+申请空间+拷贝代码)
多线程
总耗时(单个进程的耗时+IO)
ps:多线程有优势!!!
计算密集型
多进程
总耗时(单个进程的耗时)
多线程
总耗时(多个进程的综合)
ps:多进程完胜!!!
from threading import Thread from multiprocessing import Process import os import time def work(): # 计算密集型 res = 1 for i in range(1, 100000): res *= i if __name__ == '__main__': # print(os.cpu_count()) # 12 查看当前计算机CPU个数 start_time = time.time() # p_list = [] # for i in range(12): # 一次性创建12个进程 # p = Process(target=work) # p.start() # p_list.append(p) # for p in p_list: # 确保所有的进程全部运行完毕 # p.join() t_list = [] for i in range(12): t = Thread(target=work) t.start() t_list.append(t) for t in t_list: t.join() print('总耗时:%s' % (time.time() - start_time)) # 获取总的耗时 """ 计算密集型 多进程:5.665567398071289 多线程:30.233906745910645 """ def work(): time.sleep(2) # 模拟纯IO操作 if __name__ == '__main__': start_time = time.time() # t_list = [] # for i in range(100): # t = Thread(target=work) # t.start() # for t in t_list: # t.join() p_list = [] for i in range(100): p = Process(target=work) p.start() for p in p_list: p.join() print('总耗时:%s' % (time.time() - start_time)) """ IO密集型 多线程:0.0149583816528320 多进程:0.6402878761291504 """
十、信号量
在python并发编程中信号量相当于多把互斥锁（设置多少信号量，同一时间最多可以运行多少把互斥锁）
概念
信号量（英语：semaphore）又称为信号标，
是一个同步对象，用于保持在0至指定最大值之间的一个计数值。
当线程完成一次对该semaphore对象的等待（wait）时，该计数值减一；
当线程完成一次对semaphore对象的释放（release）时，计数值加一。
当计数值为0，则线程等待该semaphore对象不再能成功直至该semaphore对象变成signaled状态。
semaphore对象的计数值大于0，为signaled状态；计数值等于0，为nonsignaled状态.
举例
以停车场的运作为例。
假设停车场只有三个车位，开始三个车位都是空的。
这时同时来了五辆车，看门人开闸允许其中三辆直接进入，
剩下的车则必须在入口等待，后续来的车也在入口处等待。
这时一辆车想离开停车场，告知看门人，打开闸门放他出去，
看门人看了看空车位数量，然后看门人才让外面的一辆车进去。
如果又离开两辆，则又可以放入两辆，如此往复。
在这个停车场系统中，车位是公共资源，每辆车好比一个线程，
看门人起的就是信号量的作用。
代码
from threading import Thread, Lock, Semaphore import time import random sp = Semaphore(5) # 一次性产生五把锁 class MyThread(Thread): def run(self): sp.acquire() print(self.name) time.sleep(random.randint(1, 3)) sp.release() for i in range(20): t = MyThread() t.start()
十一、event事件
在初始情况下,Event对象中的信号标志被设置为假。如果有线程等待一个Event对象, 而这个Event对象的标志为假,那么这个线程将会被一直阻塞直至该标志为真。一个线程如果将一个Event对象的信号标志设置为真,它将唤醒所有等待这个Event对象的线程。如果一个线程等待一个已经被设置为真的Event对象,那么它将忽略这个事件, 继续执行。
from threading import Thread, Event import time event = Event() # 类似于造了一个红绿灯 def light(): print('红灯亮着的 所有人都不能动') time.sleep(3) print('绿灯亮了 油门踩到底 给我冲!!!') event.set() def car(name): print('%s正在等红灯' % name) event.wait() print('%s加油门 飙车了' % name) t = Thread(target=light) t.start() for i in range(20): t = Thread(target=car, args=('熊猫PRO%s' % i,)) t.start()
十二、进程池与线程池
概念介绍
在程序实际处理问题过程中，忙时会有成千上万的任务需要被执行，闲时可能只有零星任务。那么在成千上万个任务需要被执行的时候，我们就需要去创建成千上万个进程么？首先，创建进程需要消耗时间，销毁进程也需要消耗时间。第二即便开启了成千上万的进程，操作系统也不能让他们同时执行，这样反而会影响程序的效率。因此我们不能无限制的根据任务开启或者结束进程。那么我们要怎么做呢？
在这里，要给大家介绍一个进程池的概念，定义一个池子，在里面放上固定数量的进程，有需求来了，就拿一个池中的进程来处理任务，等到处理完毕，进程并不关闭，而是将进程再放回进程池中继续等待任务。如果有很多任务需要执行，池中的进程数量不够，任务就要等待之前的进程执行任务完毕归来，拿到空闲进程才能继续执行。也就是说，池中进程的数量是固定的，那么同一时间最多有固定数量的进程在运行。这样不会增加操作系统的调度难度，还节省了开闭进程的时间，也一定程度上能够实现并发效果。
进程和线程能否无限制的创建？
不可以
因为硬件的发展赶不上软件，有物理极限。如果我们在编写代码的过程中无限制的创建进程或者线程可能会导致计算机崩溃。
作用
池
降低程序的执行效率 但是保证了计算机硬件的安全
进程池
提前创建好固定数量的进程供后续程序的调用 超出则等待
线程池
提前创建好固定数量的线程供后续程序的调用 超出则等待
concurrent.fututres 模块
ProcessPoolExecutor 类–进程池开启
进程池类的导入：from concurrent.fututres import ProcessPoolExecutor
实例化：pool_p = ProcessPoolExecutor( 整数 ): 实例化获得一个进程池, 参数传入一个整数，代表进程池的大小
不传的话会默认开设当前计算机CPU 个数的进程
方法及属性介绍
异步提交任务–submit 方法
pool_p.submit(task,n=i) : task–提交的任务，逗号之后可以按照位置参数或者关键字参数传入task所需的参数;
submit 方法 ：会有一个返回值，返回一个Future对象：
Future对象会有一个result 方法
submit 方法演示
from concurrent.futures import ProcessPoolExecutor import time pool1 = ProcessPoolExecutor(5) def task(n): print(n) time.sleep(2) if __name__ == '__main__': for i in range(5): pool1.submit(task, i) print('main process')
由结果不难发现submit 方法是异步提交任务
from concurrent.futures import ProcessPoolExecutor import time pool1 = ProcessPoolExecutor(5) def task(n): print(n, end=" ") time.sleep(1) if __name__ == '__main__': for i in range(20): pool1.submit(task, i) print('main process')
result方法
submit 方法返回的Future对象会有一个result 方法
result 方法 会返回**提交的任务最终返回的结果 **
from concurrent.futures import ProcessPoolExecutor import time pool1 = ProcessPoolExecutor(5) def task(n): print(n,end=' ') time.sleep(1) if __name__ == '__main__': for i in range(5): res=pool1.submit(task, i) print(res.result()) print('main process')
修改提交任务的返回值–验证
from concurrent.futures import ProcessPoolExecutor import time pool1 = ProcessPoolExecutor(5) def task(n): print(n,end=' ') time.sleep(1) return n**2 if __name__ == '__main__': for i in range(5): res=pool1.submit(task, i) print(res.result()) print('main process')
shutdown方法
关闭线程池，等待线程池中所有的任务全部运行结束
if __name__ == '__main__': l = [] for i in range(10): res=pool1.submit(task, i) l.append(res) pool1.shutdown() for res in l: print('返回值：',res.result()) print('main process')
如何将所有的任务全部提交运行结束之后，再统一获得结果？
做法参考创建线程的第一种方式。
add_done_callback–异步回调机制
from concurrent.futures import ProcessPoolExecutor import time pool1 = ProcessPoolExecutor(5) def task(n): print('--调用函数：',n) time.sleep(0.5) return n ** 2 def call_back(feature): print('！！获得函数结果：',feature.result()) if __name__ == '__main__': for i in range(10): res = pool1.submit(task, i) time.sleep(0.2) res.add_done_callback(call_back)
注意：对于add_done_callback( callback )传入的函数名callback,在定义callback函数的时候一定要写一个位置参数
这个位置参数会通过add_done_callback( callback )方法自动传入, 而且这个参数就是 调用add_done_callback方法的feature对象
ThreadPoolExecutor类–线程池的开启
方法与属性 与进程池完全一致！！！
上课使用的简单例子
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor import os import time import random from threading import current_thread # 1.产生含有固定数量线程的线程池 # pool = ThreadPoolExecutor(10) pool = ProcessPoolExecutor(5) def task(n): print('task is running') # time.sleep(random.randint(1, 3)) # print('task is over', n, current_thread().name) # print('task is over', os.getpid()) return '我是task函数的返回值' def func(*args, **kwargs): print('from func') if __name__ == '__main__': # 2.将任务提交给线程池即可 for i in range(20): # res = pool.submit(task, 123) # 朝线程池提交任务 # print(res.result()) # 不能直接获取 # pool.submit(task, 123).add_done_callback(func)
十三、协程
""" 进程:资源单位 线程:执行单位 协程:单线程下实现并发(效率极高) 在代码层面欺骗CPU 让CPU觉得我们的代码里面没有IO操作 实际上IO操作被我们自己写的代码检测 一旦有 立刻让代码执行别的 (该技术完全是程序员自己弄出来的 名字也是程序员自己起的) 核心:自己写代码完成切换+保存状态 """ import time from gevent import monkey; monkey.patch_all() # 固定编写 用于检测所有的IO操作(猴子补丁) from gevent import spawn def func1(): print('func1 running') time.sleep(3) print('func1 over') def func2(): print('func2 running') time.sleep(5) print('func2 over') if __name__ == '__main__': start_time = time.time() # func1() # func2() s1 = spawn(func1) # 检测代码 一旦有IO自动切换(执行没有io的操作 变向的等待io结束) s2 = spawn(func2) s1.join() s2.join() print(time.time() - start_time) # 8.01237154006958 协程 5.015487432479858
十四、协程实现并发
同时运行两个线程，然后用代码检测I/O操作，然后让两者在I/O操作的时候不中断非阻塞态（不适用协程的时候，应该是串行方式运行代码，也就是同步的方式）
import socket from gevent import monkey;monkey.patch_all() # 固定编写 用于检测所有的IO操作(猴子补丁) from gevent import spawn def communication(sock): while True: data = sock.recv(1024) print(data.decode('utf8')) sock.send(data.upper()) def get_server(): server = socket.socket() server.bind(('127.0.0.1', 8080)) server.listen(5) while True: sock, addr = server.accept() # IO操作 spawn(communication, sock) s1 = spawn(get_server) s1.join() 如何不断的提升程序的运行效率 多进程下开多线程 多线程下开协程