Python学习笔记 - day13 - 进程与线程
Python学习笔记 - day13 - 进程与线程
概述
我们都知道windows是支持多任务的操作系统。
什么叫“多任务”呢？简单地说，就是操作系统可以同时运行多个任务。打个比方，你一边在用浏览器上网，一边在听MP3，一边在用Word赶作业，这就是多任务，至少同时有3个任务正在运行。还有很多任务悄悄地在后台同时运行着，只是桌面上没有显示而已。
现在，多核CPU已经非常普及了，但是，即使过去的单核CPU，也可以执行多任务。由于CPU执行代码都是顺序执行的，那么，单核CPU是怎么执行多任务的呢？
答案就是操作系统轮流让各个任务交替执行，任务1执行0.01秒，切换到任务2，任务2执行0.01秒，再切换到任务3，执行0.01秒……这样反复执行下去。表面上看，每个任务都是交替执行的，但是，由于CPU的执行速度实在是太快了，我们感觉就像所有任务都在同时执行一样。
真正的并行执行多任务只能在多核CPU上实现，但是，由于任务数量远远多于CPU的核心数量，所以，操作系统也会自动把很多任务轮流调度到每个核心上执行。
对于操作系统来说，一个任务就是一个进程（Process），比如打开一个浏览器就是启动一个浏览器进程，打开一个记事本就启动了一个记事本进程，打开两个记事本就启动了两个记事本进程，打开一个Word就启动了一个Word进程。
有些进程还不止同时干一件事，比如Word，它可以同时进行打字、拼写检查、打印等事情。在一个进程内部，要同时干多件事，就需要同时运行多个“子任务”，我们把进程内的这些“子任务”称为线程（Thread）。
由于每个进程至少要干一件事，所以，一个进程至少有一个线程。当然，像Word这种复杂的进程可以有多个线程，多个线程可以同时执行，多线程的执行方式和多进程是一样的，也是由操作系统在多个线程之间快速切换，让每个线程都短暂地交替运行，看起来就像同时执行一样。当然，真正地同时执行多线程需要多核CPU才可能实现。
我们前面编写的所有的Python程序，都是执行单任务的进程，也就是只有一个线程。如果我们要同时执行多个任务怎么办？有两种解决方案：
- 一种是启动多个进程，每个进程虽然只有一个线程，但多个进程可以一块执行多个任务。
- 一种方法是启动一个进程，在一个进程内启动多个线程，这样，多个线程也可以一块执行多个任务。
当然还有第三种方法，就是启动多个进程，每个进程再启动多个线程，这样同时执行的任务就更多了，当然这种模型更复杂，实际很少采用。
总结一下就是，多任务的实现有3种方式：
-
- 多进程模式；
- 多线程模式；
- 多进程+多线程模式
同时执行多个任务通常各个任务之间并不是没有关联的，而是需要相互通信和协调，有时，任务1必须暂停等待任务2完成后才能继续执行，有时，任务3和任务4又不能同时执行，所以，多进程和多线程的程序的复杂度要远远高于我们前面写的单进程单线程的程序。
Python既支持多进程，又支持多线程。
进程
正在进行的一个过程或者说一个任务。而负责执行任务的则是CPU
由于现在计算计算机都是多任务同时进行的，比如：打开了QQ，然后听着音乐，后面下载者片儿，那么这些都是怎么完成的呢？答案是通过多进程。操作系统会对CPU的时间进行规划，每个进程执行一个任务（功能），CPU会快速的在这些进行之间进行切换已达到同时进行的目的（单核CPU的情况）
进程与程序
程序:一堆代码的集合体。
进程:指的是程序运行的过程。
注意的是：一个程序执行两次，那么会产生两个互相隔离的进程。并发与并行
并行:同时运行，只有具备多个CPU才能实现并行
并发:是伪并行，即看起来是同时运行。单个CPU+多道技术就可以实现并发。（并行也属于并发）同步与异步
同步指一个进程在执行某个请求的时候，若该请求需要一段时间才能返回信息，那么这个进程将会一直等待下去，直到返回西南喜才继续执行下去。
异步是指进程不需要一直等下去，而是继续执行下面的操作，不管其他进程的状态。当有消息返回时系统会通知进程处理，这样可以提高执行的效率。
例子：打电话就是同步，发短信就是异步 进程的创建
主要分为4种：
1、系统初始化：（查看进程Linux中用ps命令，windows中用任务管理器，前台进程负责与用户交互，后台运行的进程与用户无关，运行在后台并且只有在需要时才唤醒的进程，成为守护进程，如电子邮件，web页面，新闻，打印等）
2、一个进程在运行过程中开启了子进程（如nginx开启多线程，操作系统os.fork()，subprocess.Popen等）
3、用户的交互请求,而创建一个新的进程（如用户双击QQ）
4、一个批处理作业的开始（只在大型批处理系统中应用）
以上四种其实都是由一个已经存在了的进程执行了一个用于创建进程的系统调用而创建的。
- 在unix/Linux系统中该调用是：fork，它非常特殊。普通的函数调用，调用一次，返回一次，但是
fork()调用一次，返回两次，因为操作系统自动把当前进程（称为父进程）复制了一份（称为子进程），然后，分别在父进程和子进程内返回。子进程返回0，父进程返回子进程的PID。 - 在winodws中调用的是createProcess，CreateProcess既处理进程的创建，也负责把正确的程序装入新进程。
注意：
- 进程创建后父进程和子进程有各自不同的地址空间（多道技术要求物理层面实现进程之间内存的隔离），任何一个进程的在其地址空间中的修改都不会影响到另外的进程。
- 在Unix/linux，子进程的初始地址空间是父进程的一个副本，子进程和父进程是可以有只读的共享内存区的。但是对于Winodws系统来说，从一开始父进程与子进程的地址空间就是不同的。
进程之间共享终端，共享一个文件系统
进程的状态
进程的状态主要分为三种：进行、阻塞、就绪
线程
在传统的操作系统中，每个进程有一个地址空间，而且默认就有一个控制线程，多线程(及多个控制线程)的概念是，在一个进程中存在多个控制线程，多个控制线程共享该进程的地址空间，进程只是用来把资源集中到一起(进程只是一个资源单位，或者说资源集合)，而线程才是CPU的执行单位。
为何要用多线程
多线程指的是，在一个进程中开启多个线程，简单来说：如果多个任务公用一块地址空间，那么必须在一个进程内开启多个线程。
1、多线程共享一个进程的地址空间
2、线程比进程更轻量级，线程比进程更容易创建和撤销，在许多操作系统中，创建一个线程比创建一个进程要快10-100倍
3、对于CPU密集型的应用，多线程并不能提升性能，但对于I/O密集型，使用多线程会明显的提升速度(I/O密集型，根本用不上多核优势)
4、在多CPU系统中，为了最大限度的利用多核，可以开启多个线程(比开进程开销要小的多) --> 针对其他语言
注意：
Python中的线程比较特殊，其他语言，1个进程内4个线程，如果有4个CPU的时候，是可以同时运行的，而Python在同一时间1个进程内，只有一个线程可以工作。（就算你有再多的CPU，对Python来说用不上）线程与进程的区别
1、线程共享创建它的进程的地址空间，进程拥有自己的地址空间
2、线程可以直接访问进程的数据，进程拥有它父进程内存空间的拷贝
3、线程可以和同一进程内其他的线程直接通信，进程必须interprocess communicateion(IPC机制)进行通信
4、线程可以被很容易的创建，而进程依赖于父进程内存空间的拷贝
5、线程可以直接控制同一进程内的其他线程，进程只能控制自己的子进程
6、改变主线程(控制)可能会影响其他线程，改变主进程不会影响它的子进程
multiprocessing模块
python中的多线程无法利用多核优势，如果想要充分地使用多核CPU的资源（os.cpu_count()查看），在python中大部分情况需要使用多进程。Python提供了multiprocessing,该模块用来开启子进程，并在子进程中执行我们定制的任务（比如函数），该模块与多线程模块threading的编程接口类似。
multiprocessing模块的功能众多：支持子进程、通信和共享数据、执行不同形式的同步，提供了Process、Queue、Pipe、Lock等组件。
需要再次强调的一点是：与线程不同，进程没有任何共享状态，进程修改的数据，改动仅限于该进程内。
Process类和使用
注意：在windows中Process()必须放到# if __name__ == '__main__':下
利用Process创建进程的类：
Process([group [, target [, name [, args [, kwargs]]]]])，由该类实例化得到的对象，表示一个子进程中的任务（尚未启动）
强调：
1. 需要使用关键字的方式来指定参数
2. args指定的为传给target函数的位置参数，是一个元组形式，必须有逗号
参数：
- group参数未使用，值始终为None
- target表示调用对象，即子进程要执行的任务
- args表示调用对象的位置参数元组，args=(1,2,'egon',)
- kwargs表示调用对象的字典,kwargs={'name':'egon','age':18}
- name为子进程的名称
Process类的方法
p.start()：
# 启动进程，并调用该子进程中的p.run() --> 和直接调用run方法是不同的，因为它会初始化部分其他参数。
p.run()：
# 进程启动时运行的方法，正是它去调用target指定的函数，我们自定义类的类中一定要实现该方法
p.terminate()：
# 强制终止进程p，不会进行任何清理操作，如果p创建了子进程，该子进程就成了僵尸进程，使用该方法需要特别小心这种情况。如果p还保存了一个锁那么也将不会被释放，进而导致死锁
p.is_alive()：
# 如果p仍然运行，返回True
p.join([timeout])：
# 主线程等待p终止（强调：是主线程处于等的状态，而p是处于运行的状态）。timeout是可选的超时时间，需要强调的是，p.join只能join住start开启的进程，而不能join住run开启的进程
Process的其他属性
p.daemon：
# 默认值为False，如果设为True，代表p为后台运行的守护进程，当p的父进程终止时，p也随之终止，并且设定为True后，p不能创建自己的新进程，必须在p.start()之前设置
p.name:
# 进程的名称
p.pid：
# 进程的pid
p.exitcode:
# 进程在运行时为None、如果为–N，表示被信号N结束(了解即可)
p.authkey:
# 进程的身份验证键,默认是由os.urandom()随机生成的32字符的字符串。这个键的用途是为涉及网络连接的底层进程间通信提供安全性，这类连接只有在具有相同的身份验证键时才能成功
特别强调：设置 p.daemon=True 是会随着主进程执行完毕而被回收，不管子进程是否完成任务。
基本使用
使用Process创建进程的类有两种方法：
1、通过实例化Process类完成进程的创建
2、继承Process类，定制自己需要的功能后实例化创建进程类
# --------------------------- 方法1 ---------------------------
import random
import time
from multiprocessing import Process
def hello(name):
print('Welcome to my Home')
time.sleep(random.randint(1,3))
print('Bye Bye')
p = Process(target=hello,args=('daxin',)) # 创建子进程p
p.start() # 启动子进程
print('主进程结束')
# --------------------------- 方法2 ---------------------------
import random
import time
from multiprocessing import Process
class MyProcess(Process):
def __init__(self,name):
super(MyProcess, self).__init__() # 必须继承父类的构造函数
self.name = name
def run(self): # 必须叫run方法，因为start,就是执行的run方法。
print('Welcome to {0} Home'.format(self.name))
time.sleep(random.randint(1,3))
print('Bye Bye')
p = MyProcess('daxin')
p.start()
print('主进程结束')
利用多进程完成修改socket server
上一节我们利用socket完成了socket server的编写，这里我们使用multiprocessing对server端进行改写，完成并发接受请求的功能。
如果服务端接受上万个请求，那么岂不是要创建1万个进程去分别对应？这样是不行的，那么我们可以使用进程池的概念来解决这个问题，进程池的问题，在后续小节中详细说明
进程同步锁
进程之间数据不共享,但是共享同一套文件系统,所以访问同一个文件,或同一个打印终端,是没有问题的,竞争带来的结果就是错乱，如何控制，就是加锁处理。
锁的目的就是：当程序1在使用的时候，申请锁，并且锁住共享资源，待使用完毕后，释放锁资源，其他程序获取锁后，重复这个过程。
Multiprocessing模块提供了Lock对象用来完成进程同步锁的功能
from multiprocessing import Lock
lock = Lock()
# 对象没有参数
# 通过使用lock对象的acquire/release方法来进行 锁/释放 的需求。
利用进程同步锁模拟抢票软件的需求：
- 创建票文件，内容为json，设置余票数量
- 并发100个进程抢票
- 利用random + time 模块模拟网络延迟
import random
import time
import json
from multiprocessing import Process,Lock
def gettickles(filename,str,lock):
lock.acquire() # 对要修改的部分加锁
with open(filename,encoding='utf-8') as f:
dic = json.loads(f.read())
if dic['count'] > 0 :
dic['count'] -= 1
time.sleep(random.random())
with open(filename,'w',encoding='utf-8') as f:
f.write(json.dumps(dic))
print('\033[33m{0}抢票成功\033[0m'.format(str))
else:
print('\033[35m{0}抢票失败\033[0m'.format(str))
lock.release() # 修改完毕后解锁
if __name__ == '__main__':
lock = Lock() # 创建一个锁文件
p_l = []
for i in range(1000):
p = Process(target=gettickles,args=('a.txt','用户%s' % i,lock))
p_l.append(p)
p.start()
加锁可以保证多个进程修改同一块数据时，同一时间只能有一个任务可以进行修改，即串行的修改，没错，速度是慢了，但牺牲了速度却保证了数据安全。
进程池
在利用Python进行系统管理的时候，特别是同时操作多个文件目录，或者远程控制多台主机，并行操作可以节约大量的时间。多进程是实现并发的手段之一，需要注意的问题是：
- 很明显需要并发执行的任务通常要远大于核数
- 一个操作系统不可能无限开启进程，通常有几个核就开几个进程
- 进程开启过多，效率反而会下降（开启进程是需要占用系统资源的，而且开启多余核数目的进程也无法做到并行）
例如当被操作对象数目不大时，可以直接利用multiprocessing中的Process动态成生多个进程，十几个还好，但如果是上百个，上千个。。。手动的去限制进程数量却又太过繁琐，此时可以发挥进程池的功效。
我们就可以通过维护一个进程池来控制进程数目，比如httpd的进程模式，规定最小进程数和最大进程数...
ps：对于远程过程调用的高级应用程序而言，应该使用进程池，Pool可以提供指定数量的进程，供用户调用，当有新的请求提交到pool中时，如果池还没有满，那么就会创建一个新的进程用来执行该请求；但如果池中的进程数已经达到规定最大值，那么该请求就会等待，直到池中有进程结束，就重用进程池中的进程。
创建进程池的类：如果指定numprocess为3，则进程池会从无到有创建三个进程，然后自始至终使用这三个进程去执行所有任务，不会开启其他进程
from multiprocessing import Pool
pool = Pool(processes=None, initializer=None, initargs=())
参数：
- processes：进程池的最大进程数量
- initiallizer：初始化完毕后要执行的函数
- initargs：要传递给函数的参数
常用方法
p.apply(func [, args [, kwargs]])
# 调用进程池中的一个进程执行函数func，args/kwargs为传递的参数，注意apply是阻塞式的，既串行执行。
p.apply_async(func [, args [, kwargs]])
# 功能同apply，区别是非阻塞的，既异步执行。 ———> 常用
p.close()
# 关闭进程池，防止进一步操作。如果所有操作持续挂起，它们将在工作进程终止前完成
P.join()
# 等待所有工作进程退出。此方法只能在close（）或teminate()之后调用注意：
apply_async 会返回AsyncResul对象，这个AsyncResul对象有有一下方法：
利用进程池改写socket server：
import os
import socket
import multiprocessing
server = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)
server.bind(('127.0.0.1',8100))
server.listen(5)
def talk(conn):
print('我的进程号是： %s' % os.getpid() )
while True:
msg = conn.recv(1024)
if not msg:break
data = msg.decode('utf-8')
msg = data.upper()
conn.send(msg.encode('utf-8'))
if __name__ == '__main__':
pool = multiprocessing.Pool(1)
while True:
conn,addr = server.accept()
print(addr)
pool.apply_async(talk,args=(conn,))
pool.close()
pool.join()这里指定了进程池的数量为1，那么并发两个连接的话，第二个会hold住，只有第一个断开后，才会连接，注意：进程的Pid号，还是相同的。
回调函数
需要回调函数的场景：进程池中任何一个任务一旦处理完了，就立即告知主进程：我好了额，你可以处理我的结果了。主进程则调用一个函数去处理该结果，该函数即回调函数。我们可以把耗时间（阻塞）的任务放到进程池中，然后指定回调函数（主进程负责执行），这样主进程在执行回调函数时就省去了I/O的过程，直接拿到的是任务的结果。
apply_async(self, func, args=(), kwds={}, callback=None）
# func的结果会交给指定的callback函数处理
一个爬虫的小例子：
from multiprocessing import Pool
import requests
import os
def geturl(url):
print('我的进程号为： %s' % os.getpid())
print('我处理的url为： %s ' % url )
response = requests.get(url) # 请求网页
return response.text # 返回网页源码
def urlparser(htmlcode):
print('我的进程号是: %s ' % os.getpid())
datalength = len(htmlcode) # 计算源码的长度
print('解析到的html大小为： %s' % datalength)
if __name__ == '__main__':
pool = Pool()
url = [
'http://www.baidu.com',
'http://www.sina.com',
'http://www.qq.com',
'http://www.163.com'
]
res_l = []
for i in url:
res = pool.apply_async(geturl,args=(i,),callback=urlparser) # res 是 geturl执行的结果，因为已经交给urlparser处理了，所以这里不用拿
res_l.append(res)
pool.close()
pool.join()
for res in res_l:
print(res.get()) # 这里拿到的就是网页的源码进程间通讯
进程彼此之间互相隔离，要实现进程间通信（IPC），multiprocessing模块提供的两种形式：队列和管道，这两种方式都是使用消息传递的。但是还有一种基于共享数据的方式，现在已经不推荐使用，建议使用队列的方式进行进程间通讯。
展望未来，基于消息传递的并发编程是大势所趋，即便是使用线程，推荐做法也是将程序设计为大量独立的线程集合，通过消息队列交换数据。这样极大地减少了对使用锁定和其他同步手段的需求，还可以扩展到分布式系统中。
队列
底层就是以管道和锁定的方式实现。
创建队列的类：
Queue([maxsize]):创建共享的进程队列，Queue是多进程安全的队列，可以使用Queue实现多进程之间的数据传递。
# 参数
maxsize: 队列能承载的最大数量，省略的话则不限制队列大小
基本使用：
from multiprocessing import Queue
q = Queue(3)
q.put('a') # 数据存入Queue
print(q.get()) # 从Queue中取出数据
注意：队列（Queue）是FIFO模式，既先进先出。
队列的方法
q.put() 用于插入数据到队列中。
q.put(obj, block=True, timeout=None)
# 参数：
# blocked，timeout：如果blocked为True（默认值），并且timeout为正值，该方法会阻塞timeout指定的时间，直到该队列有剩余的空间。如果超时，会抛出Queue.Full异常。如果blocked为False，但该Queue已满，会立即抛出Queue.Full异常。
PS：q.put_nowait() 等同于 q.put(block=False)
q.get() 用于从队列中获取数据。
q.get(block=True,timeout=None)
# 参数：
# blocked和timeout。如果blocked为True（默认值），并且timeout为正值，那么在等待时间内没有取到任何元素，会抛出Queue.Empty异常。如果blocked为False，有两种情况存在，如果Queue有一个值可用，则立即返回该值，否则，如果队列为空，则立即抛出Queue.Empty异常.
PS：q.get_nowait() 等同于 q.get(block=False)
生产者消费者模型
在并发编程中使用生产者和消费者模式能够解决绝大多数并发问题。该模式通过平衡生产线程和消费线程的工作能力来提高程序的整体处理数据的速度。
为什么要使用生产者和消费者模式
在线程世界里，生产者就是生产数据的线程，消费者就是消费数据的线程。在多线程开发当中，如果生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。为了解决这个问题于是引入了生产者和消费者模式。
什么是生产者消费者模式
生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。
基于队列实现生产者消费者模型：
- 生产者只负责生产蛋糕，生产完毕的蛋糕放在队列中
- 消费者只负责消费蛋糕，每次从队列中拿取蛋糕
上面的例子很完美，但是生产者生产完毕，消费者也消费完毕了，那么我们的主程序就应该退出了，可是并没有，因为消费者还在等待从队列中获取(q.get)，这里我们考虑可以发送一个做完/吃完的信号，抓取到信号后退出即可。
- 在队列中放固定的值来做信号
- 利用JoinableQueue对象 + daemon属性 来对消费者进程进行回收
其中：
- 利用JoinableQueue对象的join，task_done方法，完成确认/通知的目的。
- 如果生产者生产完毕，消费者必然也会给生产者确认消费完毕，那么只要等待生产者执行完毕后进行就可以退出主进程了。
- 主进程退出但是消费者进程还未回收，那么就可以设置消费者daemon属性为true，跟随主进程被回收即可。
共享数据
进程间数据是独立的，可以借助于队列或管道实现通信，二者都是基于消息传递的，虽然进程间数据独立，但也可以通过Manager实现数据共享，事实上Manager的功能远不止于此。
Manager()
# 没有参数
# 使用Manager对象创建共享数据类型
利用Manager创建数据，完成进程共享
import os
from multiprocessing import Manager,Process
def worker(d,l):
d[os.getpid()]=os.getpid() # 对共享数据进行修改
l.append(os.getpid())
if __name__ == '__main__':
m = Manager()
d = m.dict() # 创建共享字典
l = m.list() # 创建共享列表
p_l = []
for i in range(10):
p= Process(target=worker,args=(d,l))
p_l.append(p)
p.start()
for p in p_l:
p.join()
print(d)
print(l)
Threading模块
Python 标准库提供了 thread 和 threading 两个模块来对多线程进行支持。其中， thread 模块以低级、原始的方式来处理和控制线程，而 threading 模块通过对 thread 进行二次封装，提供了更方便的 api 来处理线程。
PS：multiprocessing完全模仿了threading模块的接口，二者在使用层面，有很大的相似性，所以很多用法都是相同的，所以可能看起来会比较眼熟。
Thread类和使用
Thread 是threading模块中最重要的类之一，可以使用它来创建线程。
有两种方式来创建线程：
- 通过继承Thread类，重写它的run方法；
- 创建一个threading.Thread对象，在它的初始化函数（__init__）中将可调用对象作为参数传入；
# -----------------------实例化对象--------------------------
import threading
def work(name):
print('hello,{0}'.format(name))
if __name__ == '__main__':
t = threading.Thread(target=work,args=('daxin',))
t.start()
print('主进程')
# -----------------------自己创建类--------------------------
import threading
class Work(threading.Thread):
def __init__(self,name):
super(Work, self).__init__()
self.name = name
def run(self):
print('hello,{0}'.format(self.name))
if __name__ == '__main__':
t = Work(name='daxin')
t.start()
print('主进程')
PS：执行的时候，我们可以看到会先打印"hello,daxin"，然后才会打印"主进程"，所以这也同时说明了，创建线程比创建进程消耗资源少的多，线程会被很快的创建出来并执行。如果我们在target执行的函数和主函数中，同时打印os.getpid，你会发现，进程号是相同的，这也说明了这里开启的是自线程。