进程
相关概念
进程
进程（Process）是计算机中的程序关于某数据集合上的一次运行活动，是系统进行资源分配和调度的基本单位，是操作系统结构的基础。在早期面向进程设计的计算机结构中，进程是程序的基本执行实体；在当代面向线程设计的计算机结构中，进程是线程的容器。程序是指令、数据及其组织形式的描述，进程是程序的实体。
狭义定义：进程是正在运行的程序的实例（an instance of a computer program that is being executed）。
广义定义：进程是一个具有一定独立功能的程序关于某个数据集合的一次运行活动。它是操作系统动态执行的基本单元，在传统的操作系统中，进程既是基本的分配单元，也是基本的执行单元。
同步/异步
所谓同步就是一个任务的完成需要依赖另外一个任务时，只有等待被依赖的任务完成后，依赖的任务才能算完成，这是一种可靠的任务序列。要么成功都成功，失败都失败，两个任务的状态可以保持一致。
所谓异步是不需要等待被依赖的任务完成，只是通知被依赖的任务要完成什么工作，依赖的任务也立即执行，只要自己完成了整个任务就算完成了。至于被依赖的任务最终是否真正完成，依赖它的任务无法确定，所以它是不可靠的任务序列。
阻塞/非阻塞
阻塞和非阻塞这两个概念与程序等待消息通知(无所谓同步或者异步)时的状态有关。也就是说阻塞与非阻塞主要是程序（线程）等待消息通知时的状态角度来说的。
并发/并行
并行 : 并行是指多个任务同时执行，比如两个男人同时在给自己女朋友发微信。
并发 : 并发是多个任务交替轮流使用资源，比如一个男人在给他7个女朋友发微信，只要他发的够快，宏观上来说他在同时聊7个人。
进程状态与调度
在了解其他概念之前，我们首先要了解进程的几个状态。在程序运行的过程中，由于被操作系统的调度算法控制，程序会进入几个状态：就绪，运行和阻塞。
（1）就绪(Ready)状态
当进程已分配到除CPU以外的所有必要的资源，只要获得处理机便可立即执行，这时的进程状态称为就绪状态。
（2）执行/运行（Running）状态当进程已获得处理机，其程序正在处理机上执行，此时的进程状态称为执行状态。
（3）阻塞(Blocked)状态正在执行的进程，由于等待某个事件发生而无法执行时，便放弃处理机而处于阻塞状态。引起进程阻塞的事件可有多种，例如，等待I/O完成、申请缓冲区不能满足、等待信件(信号)等。
Python中使用多进程
之前我们已经了解了很多进程相关的理论知识，了解进程是什么应该不再困难了，刚刚我们已经了解了，运行中的程序就是一个进程。所有的进程都是通过它的父进程来创建的。因此，运行起来的python程序也是一个进程，那么我们也可以在程序中再创建进程。多个进程可以实现并发效果，也就是说，当我们的程序中存在多个进程的时候，在某些时候，就会让程序的执行速度变快。以我们之前所学的知识，并不能实现创建进程这个功能，所以我们就需要借助python中强大的模块。
multiprocessing模块
仔细说来，multiprocess不是一个模块而是python中一个操作、管理进程的包。 之所以叫multi是取自multiple的多功能的意思,在这个包中几乎包含了和进程有关的所有子模块。由于提供的子模块非常多，为了方便大家归类记忆，我将这部分大致分为四个部分：创建进程部分，进程同步部分，进程池部分，进程之间数据共享。
multiprocessing.Process介绍
process模块是一个创建进程的模块，借助这个模块，就可以完成进程的创建。
Process([group [, target [, name [, args [, kwargs]]]]])，由该类实例化得到的对象，表示一个子进程中的任务（尚未启动）强调：1. 需要使用关键字的方式来指定参数2. args指定的为传给target函数的位置参数，是一个元组形式，必须有逗号参数介绍：1 group参数未使用，值始终为None2 target表示调用对象，即子进程要执行的任务3 args表示调用对象的位置参数元组，args=(1,2,'egon',)4 kwargs表示调用对象的字典,kwargs={'name':'egon','age':18}5 name为子进程的名称
1 p.start()：启动进程，并调用该子进程中的p.run()2 p.run():进程启动时运行的方法，正是它去调用target指定的函数，我们自定义类的类中一定要实现该方法3 p.terminate():强制终止进程p，不会进行任何清理操作，如果p创建了子进程，该子进程就成了僵尸进程，使用该方法需要特别小心这种情况。如果p还保存了一个锁那么也将不会被释放，进而导致死锁4 p.is_alive():如果p仍然运行，返回True5 p.join([timeout]):主线程等待p终止（强调：是主线程处于等的状态，而p是处于运行的状态）。timeout是可选的超时时间，需要强调的是，p.join只能join住start开启的进程，而不能join住run开启的进程
方法介绍
1 p.daemon：默认值为False，如果设为True，代表p为后台运行的守护进程，当p的父进程终止时，p也随之终止，并且设定为True后，p不能创建自己的新进程，必须在p.start()之前设置2 p.name:进程的名称3 p.pid：进程的pid4 p.exitcode:进程在运行时为None、如果为–N，表示被信号N结束(了解即可)5 p.authkey:进程的身份验证键,默认是由os.urandom()随机生成的32字符的字符串。这个键的用途是为涉及网络连接的底层进程间通信提供安全性，这类连接只有在具有相同的身份验证键时才能成功（了解即可）
属性介绍
在Windows操作系统中由于没有fork(linux操作系统中创建进程的机制)，在创建子进程的时候会自动 import 启动它的这个文件，而在 import 的时候又执行了整个文件。因此如果将process()直接写在文件中就会无限递归创建子进程报错。所以必须把创建子进程的部分使用if __name__ ==‘__main__’ 判断保护起来，import 的时候 ，就不会递归运行了。
在windows中使用process模块的注意事项
使用process模块创建进程
在一个python进程中开启子进程，start方法和并发效果。
from multiprocessing import Processimport timedef task(name): print('{} is running!'.format(name)) time.sleep(3) print('{} is done!'.format(name))# Windows开子进程要写在__name__==__main__下# 因为开子进程会重新加载父进程的内容if __name__ == '__main__': # 创建一个Python中的进程对象 p = Process(target=task, args=('t1', )) # p = Process(target=task, kwargs={'name': 't1'}) p.start() # 调用操作系统接口启动一个进程执行命令 print('--- 主进程 ----')
查看主进程id和父进程id:
from multiprocessing import Processimport timeimport osdef task(name): print('{} is running!'.format(name)) print('子进程id ：', os.getpid(), '父进程id ：', os.getppid()) time.sleep(3) print('{} is done!'.format(name))# Windows开子进程要写在__name__==__main__下# 因为开子进程会重新加载父进程的内容if __name__ == '__main__': print('主进程id ：', os.getpid()) # 创建一个Python中的进程对象 p = Process(target=task, args=('t1', )) # p = Process(target=task, kwargs={'name': 't1'}) p.start() # 调用操作系统接口启动一个进程执行命令 print('--- 主进程 ----')
查看进程id
我们再启动多个进程看一下，运行的效果：(注意，子进程的执行顺序不是根据启动顺序决定的)
from multiprocessing import Processimport timedef task(name): print('{} is running!'.format(name)) time.sleep(3) print('{} is done!'.format(name))if __name__ == '__main__': for i in range(10): p = Process(target=task, args=(i, )) p.start() print('--- 主进程 ----')
进程间的数据隔离
进程的运行时数据是互相独立的，不会相互影响。
想个办法，验证一下：
from multiprocessing import Processimport timex = 100def change(): global x x = 10 print('子进程修改了x,子进程结束了！')if __name__ == '__main__': p = Process(target=change) p.start() time.sleep(5) # 想办法等子进程结束 此处应该有思考 print(x)
上面的例子中可以看到，父进程中的X变量并没有被子进程中的代码修改掉。
join
接下来我们看一下上面代码中的问题，也就是我们应该如何优雅的实现父进程等待子进程结束呢？
这里就用到了Python中进程对象的join()。
from multiprocessing import Processx = 100def change(): global x x = 10 print('子进程修改了x,子进程结束了！')if __name__ == '__main__': p = Process(target=change) p.start() p.join() # 优雅地实现主进程等待子进程结束 print(x)
再来几个多线程的例子，验证一下：
from multiprocessing import Processimport timedef task(n): print('这是子进程:{}'.format(n)) time.sleep(n) print('子进程:{}结束了！'.format(n))if __name__ == '__main__': start_time = time.time() p1 = Process(target=task, args=(1,)) p2 = Process(target=task, args=(2,)) p3 = Process(target=task, args=(3,)) p1.start() p2.start() p3.start() p1.join() p2.join() p3.join() print('我是主进程') print('共耗时：{}'.format(time.time()-start_time))
或者写成下面这样的简写方式：
from multiprocessing import Processimport timedef task(n): print('这是子进程:{}'.format(n)) time.sleep(n) print('子进程:{}结束了！'.format(n))if __name__ == '__main__': start_time = time.time() # p1 = Process(target=task, args=(1,)) # p2 = Process(target=task, args=(2,)) # p3 = Process(target=task, args=(3,)) # p1.start() # p2.start() # p3.start() # # p1.join() # p2.join() # p3.join() # 或者简写成下面的方式 p_list = [] for i in range(1, 4): p = Process(target=task, args=(i,)) p.start() p_list.append(p) for p in p_list: p.join() print('我是主进程') print('共耗时：{}'.format(time.time()-start_time))
for循环起进程
思考一下：
下面的代码有何不可？
from multiprocessing import Processimport timedef task(n): print('这是子进程:{}'.format(n)) time.sleep(n) print('子进程:{}结束了！'.format(n))if __name__ == '__main__': start_time = time.time() p1 = Process(target=task, args=(1,)) p2 = Process(target=task, args=(2,)) p3 = Process(target=task, args=(3,)) p1.start() p1.join() p2.start() p2.join() p3.start() p3.join() print('我是主进程') print('共耗时：{}'.format(time.time() - start_time))
这样写的问题是？
除了上面这些开启进程的方法，还有一种以继承Process类的形式开启进程的方式
import osfrom multiprocessing import Processclass MyProcess(Process): def __init__(self,name): super().__init__() self.name=name def run(self): print(os.getpid()) print('%s 正在和女主播聊天' %self.name)p1=MyProcess('wupeiqi')p2=MyProcess('yuanhao')p3=MyProcess('nezha')p1.start() #start会自动调用runp2.start()# p2.run()p3.start()p1.join()p2.join()p3.join()print('主线程')
通过继承Process类开启进程
守护进程
父进程中将一个子进程设置为守护进程，那么这个子进程会随着主进程的结束而结束。
主进程创建守护进程
其一：守护进程会在主进程代码执行结束后就终止
其二：守护进程内无法再开启子进程,否则抛出异常：AssertionError: daemonic processes are not allowed to have children
注意：进程之间是互相独立的，主进程代码运行结束，守护进程随即终止
import osimport timefrom multiprocessing import Processclass Myprocess(Process): def __init__(self,person): super().__init__() self.person = person def run(self): print(os.getpid(),self.name) print('%s正在和女主播聊天' %self.person)p=Myprocess('哪吒')p.daemon=True #一定要在p.start()前设置,设置p为守护进程,禁止p创建子进程,并且父进程代码执行结束,p即终止运行p.start()time.sleep(10) # 在sleep时查看进程id对应的进程ps -ef|grep idprint('主')
守护进程的启动
from multiprocessing import Processdef foo(): print(123) time.sleep(1) print("end123")def bar(): print(456) time.sleep(3) print("end456")p1=Process(target=foo)p2=Process(target=bar)p1.daemon=Truep1.start()p2.start()time.sleep(0.1)print("main-------")#打印该行则主进程代码结束,则守护进程p1应该被终止.#可能会有p1任务执行的打印信息123,因为主进程打印main----时,p1也执行了,但是随即被终止.
主进程代码执行结束守护进程立即结束
socket聊天并发实例
from socket import *from multiprocessing import Processserver=socket(AF_INET,SOCK_STREAM)server.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)server.bind(('127.0.0.1',8080))server.listen(5)def talk(conn,client_addr): while True: try: msg=conn.recv(1024) if not msg:break conn.send(msg.upper()) except Exception: breakif __name__ == '__main__': #windows下start进程一定要写到这下面 while True: conn,client_addr=server.accept() p=Process(target=talk,args=(conn,client_addr)) p.start()
使用多进程实现socket聊天并发-server
from socket import *client=socket(AF_INET,SOCK_STREAM)client.connect(('127.0.0.1',8080))while True: msg=input('>>: ').strip() if not msg:continue client.send(msg.encode('utf-8')) msg=client.recv(1024) print(msg.decode('utf-8'))
client端
多进程中的其他方法
from multiprocessing import Processimport timeimport randomclass Myprocess(Process): def __init__(self,person): self.name=person super().__init__() def run(self): print('%s正在和网红脸聊天' %self.name) time.sleep(random.randrange(1,5)) print('%s还在和网红脸聊天' %self.name)p1=Myprocess('哪吒')p1.start()p1.terminate()#关闭进程,不会立即关闭,所以is_alive立刻查看的结果可能还是存活print(p1.is_alive()) #结果为Trueprint('开始')print(p1.is_alive()) #结果为False
进程对象的其他方法:terminate,is_alive
1 class Myprocess(Process): 2 def __init__(self,person): 3 self.name=person # name属性是Process中的属性，标示进程的名字 4 super().__init__() # 执行父类的初始化方法会覆盖name属性 5 #self.name = person # 在这里设置就可以修改进程名字了 6 #self.person = person #如果不想覆盖进程名，就修改属性名称就可以了 7 def run(self): 8 print('%s正在和网红脸聊天' %self.name) 9 # print('%s正在和网红脸聊天' %self.person)10 time.sleep(random.randrange(1,5))11 print('%s正在和网红脸聊天' %self.name)12 # print('%s正在和网红脸聊天' %self.person)131415 p1=Myprocess('哪吒')16 p1.start()17 print(p1.pid) #可以查看子进程的进程id
进程对象的其他属性:pid和name
互斥锁
通过刚刚的学习，我们千方百计实现了程序的异步，让多个任务可以同时在几个进程中并发处理，他们之间的运行没有顺序，一旦开启也不受我们控制。尽管并发编程让我们能更加充分的利用IO资源，但是也给我们带来了新的问题。
当多个进程使用同一份数据资源的时候，就会因为竞争而引发数据安全或顺序混乱问题。
互斥锁介绍
下面的代码演示了不同的任务争抢一个资源（终端输出）的场景。
from multiprocessing import Processimport timeimport randomdef task1(): print('这是 task1 任务'.center(30, '-')) print('task1 进了洗手间') time.sleep(random.randint(1, 3)) print('task1 办事呢...') time.sleep(random.randint(1, 3)) print('task1 走出了洗手间')def task2(): print('这是 task2 任务'.center(30, '-')) print('task2 进了洗手间') time.sleep(random.randint(1, 3)) print('task2 办事呢...') time.sleep(random.randint(1, 3)) print('task2 走出了洗手间')def task3(): print('这是 task3 任务'.center(30, '-')) print('task3 进了洗手间') time.sleep(random.randint(1, 3)) print('task3 办事呢...') time.sleep(random.randint(1, 3)) print('task3 走出了洗手间')if __name__ == '__main__': p1 = Process(target=task1) p2 = Process(target=task2) p3 = Process(target=task3) p1.start() p2.start() p3.start()
进程之间竞争资源
通过加锁来控制。
from multiprocessing import Process, Lockimport timeimport random# 生成一个互斥锁mutex_lock = Lock()def task1(lock): # 锁门 lock.acquire() print('这是 task1 任务'.center(30, '-')) print('task1 进了洗手间') time.sleep(random.randint(1, 3)) print('task1 办事呢...') time.sleep(random.randint(1, 3)) print('task1 走出了洗手间') # 释放锁 lock.release()def task2(lock): # 锁门 lock.acquire() print('这是 task2 任务'.center(30, '-')) print('task2 进了洗手间') time.sleep(random.randint(1, 3)) print('task2 办事呢...') time.sleep(random.randint(1, 3)) print('task2 走出了洗手间') # 释放锁 lock.release()def task3(lock): # 锁门 lock.acquire() print('这是 task3 任务'.center(30, '-')) print('task3 进了洗手间') time.sleep(random.randint(1, 3)) print('task3 办事呢...') time.sleep(random.randint(1, 3)) print('task3 走出了洗手间') # 释放锁 lock.release()if __name__ == '__main__': p1 = Process(target=task1, args=(mutex_lock, )) p2 = Process(target=task2, args=(mutex_lock, )) p3 = Process(target=task3, args=(mutex_lock, )) # 释放新建进程的信号，具体谁先启动无法确定 p1.start() p2.start() p3.start()
使用互斥锁解决竞争
上面这种情况虽然使用加锁的形式实现了顺序的执行，但是程序又重新变成串行了，这样确实会浪费了时间，却保证了数据的安全。
互斥锁示例
接下来，我们以模拟抢票为例，来看看数据安全的重要性。
from multiprocessing import Process, Lockimport jsonimport timeimport randomimport osdef search(): time.sleep(0.5) with open('db.json', 'r', encoding='utf8') as f: data = json.load(f) print('剩余票数:{}'.format(data.get('count')))def buy(): with open('db.json', 'r', encoding='utf8') as f: data = json.load(f) if data.get('count', 0) > 0: data['count'] -= 1 time.sleep(random.randint(1, 3)) with open('db.json', 'w', encoding='utf8') as f2: json.dump(data, f2) print('{}购票成功！'.format(os.getpid())) else: print('购票失败')def task(): search() # 查票并发 buy() # 串行买票if __name__ == '__main__': for i in range(10): p = Process(target=task) p.start()
多进程同时抢票
使用互斥锁，保证数据安全。
from multiprocessing import Process, Lockimport jsonimport timeimport randomimport os# 设置互斥锁mutex_lock = Lock()def search(): time.sleep(0.5) with open('db.json', 'r', encoding='utf8') as f: data = json.load(f) print('剩余票数:{}'.format(data.get('count')))def buy(): with open('db.json', 'r', encoding='utf8') as f: data = json.load(f) if data.get('count', 0) > 0: data['count'] -= 1 time.sleep(random.randint(1, 3)) with open('db.json', 'w', encoding='utf8') as f2: json.dump(data, f2) print('{}购票成功！'.format(os.getpid())) else: print('购票失败')def task(lock): search() # 查票并发 lock.acquire() buy() # 串行买票 lock.release()if __name__ == '__main__': for i in range(10): p = Process(target=task, args=(mutex_lock, )) p.start()
互斥锁版抢票
一点思考
加锁可以保证多个进程修改同一块数据时，同一时间只能有一个任务可以进行修改，即串行的修改，在牺牲速度的前提下保证数据安全。
虽然可以用文件共享数据实现进程间通信，但问题是：
1. 效率低（共享数据基于文件，而文件是硬盘上的数据）
2. 需要自己加锁处理
因此我们最好找寻一种解决方案能够兼顾：
1. 效率高（多个进程共享一块内存的数据）
2. 帮我们处理好锁问题。
mutiprocessing模块中为我们提供了一个IPC通信机制：队列和管道。
队列和管道都是将数据存放于内存中，队列又是基于（管道+锁）实现的，可以让我们从复杂的锁问题中解脱出来，
我们应该尽量避免使用共享数据，尽可能使用消息传递和队列，避免处理复杂的同步和锁问题，而且在进程数目增多时，往往可以获得更好的可扩展性。
队列
Queue介绍
我们可以创建一个共享的进程队列，Queue是多进程安全的队列，可以使用Queue实现多进程之间的数据传递。
Queue([maxsize])创建共享的进程队列。参数 ：maxsize是队列中允许的最大项数。如果省略此参数，则无大小限制。底层队列使用管道和锁定实现。
Queue([maxsize])创建共享的进程队列。maxsize是队列中允许的最大项数。如果省略此参数，则无大小限制。底层队列使用管道和锁定实现。另外，还需要运行支持线程以便队列中的数据传输到底层管道中。Queue的实例q具有以下方法：q.get( [ block [ ,timeout ] ] )返回q中的一个项目。如果q为空，此方法将阻塞，直到队列中有项目可用为止。block用于控制阻塞行为，默认为True. 如果设置为False，将引发Queue.Empty异常（定义在Queue模块中）。timeout是可选超时时间，用在阻塞模式中。如果在制定的时间间隔内没有项目变为可用，将引发Queue.Empty异常。q.get_nowait( )同q.get(False)方法。q.put(item [, block [,timeout ] ] )将item放入队列。如果队列已满，此方法将阻塞至有空间可用为止。block控制阻塞行为，默认为True。如果设置为False，将引发Queue.Empty异常（定义在Queue库模块中）。timeout指定在阻塞模式中等待可用空间的时间长短。超时后将引发Queue.Full异常。q.qsize()返回队列中目前项目的正确数量。此函数的结果并不可靠，因为在返回结果和在稍后程序中使用结果之间，队列中可能添加或删除了项目。在某些系统上，此方法可能引发NotImplementedError异常。q.empty()如果调用此方法时 q为空，返回True。如果其他进程或线程正在往队列中添加项目，结果是不可靠的。也就是说，在返回和使用结果之间，队列中可能已经加入新的项目。q.full()如果q已满，返回为True. 由于线程的存在，结果也可能是不可靠的（参考q.empty（）方法）。。
方法介绍
q.close()关闭队列，防止队列中加入更多数据。调用此方法时，后台线程将继续写入那些已入队列但尚未写入的数据，但将在此方法完成时马上关闭。如果q被垃圾收集，将自动调用此方法。关闭队列不会在队列使用者中生成任何类型的数据结束信号或异常。例如，如果某个使用者正被阻塞在get（）操作上，关闭生产者中的队列不会导致get（）方法返回错误。q.cancel_join_thread()不会再进程退出时自动连接后台线程。这可以防止join_thread()方法阻塞。q.join_thread()连接队列的后台线程。此方法用于在调用q.close()方法后，等待所有队列项被消耗。默认情况下，此方法由不是q的原始创建者的所有进程调用。调用q.cancel_join_thread()方法可以禁止这种行为。
其他方法(了解)
队列简单示例
'''multiprocessing模块支持进程间通信的两种主要形式:管道和队列都是基于消息传递实现的,但是队列接口'''from multiprocessing import Queueq=Queue(3)#put ,get ,put_nowait,get_nowait,full,emptyq.put(3)q.put(3)q.put(3)# q.put(3) # 如果队列已经满了，程序就会停在这里，等待数据被别人取走，再将数据放入队列。 # 如果队列中的数据一直不被取走，程序就会永远停在这里。try: q.put_nowait(3) # 可以使用put_nowait，如果队列满了不会阻塞，但是会因为队列满了而报错。except: # 因此我们可以用一个try语句来处理这个错误。这样程序不会一直阻塞下去，但是会丢掉这个消息。 print('队列已经满了')# 因此，我们再放入数据之前，可以先看一下队列的状态，如果已经满了，就不继续put了。print(q.full()) #满了print(q.get())print(q.get())print(q.get())# print(q.get()) # 同put方法一样，如果队列已经空了，那么继续取就会出现阻塞。try: q.get_nowait(3) # 可以使用get_nowait，如果队列满了不会阻塞，但是会因为没取到值而报错。except: # 因此我们可以用一个try语句来处理这个错误。这样程序不会一直阻塞下去。 print('队列已经空了')print(q.empty()) #空了
单看队列用法
上面这个例子还没有加入进程通信，只是先来看看队列为我们提供的方法，以及这些方法的使用和现象。
import timefrom multiprocessing import Process, Queuedef f(q): q.put([time.asctime(), 'hi', 'hello']) #调用主函数中p进程传递过来的进程参数 put函数为向队列中添加一条数据。if __name__ == '__main__': q = Queue() #创建一个Queue对象 p = Process(target=f, args=(q,)) #创建一个进程 p.start() print(q.get()) p.join()
子进程发送数据给父进程
上面是一个queue的简单应用，使用队列q对象调用get函数来取得队列中最先进入的数据。 接下来看一个稍微复杂一些的例子：
import osimport timeimport multiprocessing# 向queue中输入数据的函数def inputQ(queue): info = str(os.getpid()) + '(put):' + str(time.asctime()) queue.put(info)# 向queue中输出数据的函数def outputQ(queue): info = queue.get() print ('%s%s\033[32m%s\033[0m'%(str(os.getpid()), '(get):',info))# Mainif __name__ == '__main__': multiprocessing.freeze_support() record1 = [] # store input processes record2 = [] # store output processes queue = multiprocessing.Queue(3) # 输入进程 for i in range(10): process = multiprocessing.Process(target=inputQ,args=(queue,)) process.start() record1.append(process) # 输出进程 for i in range(10): process = multiprocessing.Process(target=outputQ,args=(queue,)) process.start() record2.append(process) for p in record1: p.join() for p in record2: p.join()
批量生产数据放入队列再批量获取结果 x
生产者消费者模型
在并发编程中使用生产者和消费者模式能够解决绝大多数并发问题。该模式通过平衡生产数据和消费数据的工作能力来提高程序的整体处理数据的速度。
为什么要使用生产者和消费者模式
生产者就是生产数据的一方，消费者就是消费数据的一方。通常生产者和消费者的能力很难协调，例如：如果生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。为了解决这个问题于是引入了生产者和消费者模式。
什么是生产者消费者模式
生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。
基于队列实现生产者消费者模型
from multiprocessing import Process,Queueimport time,random,osdef consumer(q): while True: res=q.get() time.sleep(random.randint(1,3)) print('\033[45m%s 吃 %s\033[0m' %(os.getpid(),res))def producer(q): for i in range(10): time.sleep(random.randint(1,3)) res='包子%s' %i q.put(res) print('\033[44m%s 生产了 %s\033[0m' %(os.getpid(),res))if __name__ == '__main__': q=Queue() #生产者们:即厨师们 p1=Process(target=producer,args=(q,)) #消费者们:即吃货们 c1=Process(target=consumer,args=(q,)) #开始 p1.start() c1.start() print('主')
基于队列实现生产者消费者模型
此时的问题是主进程永远不会结束，原因是：生产者p在生产完后就结束了，但是消费者c在取空了q之后，则一直处于死循环中且卡在q.get()这一步。
解决方式无非是让生产者在生产完毕后，往队列中再发一个结束信号，这样消费者在接收到结束信号后就可以break出死循环。
from multiprocessing import Process,Queueimport time,random,osdef consumer(q): while True: res=q.get() if res is None:break #收到结束信号则结束 time.sleep(random.randint(1,3)) print('\033[45m%s 吃 %s\033[0m' %(os.getpid(),res))def producer(q): for i in range(10): time.sleep(random.randint(1,3)) res='包子%s' %i q.put(res) print('\033[44m%s 生产了 %s\033[0m' %(os.getpid(),res)) q.put(None) #发送结束信号if __name__ == '__main__': q=Queue() #生产者们:即厨师们 p1=Process(target=producer,args=(q,)) #消费者们:即吃货们 c1=Process(target=consumer,args=(q,)) #开始 p1.start() c1.start() print('主')
改良版——生产者消费者模型
注意：结束信号None，不一定要由生产者发，主进程里同样可以发，但主进程需要等生产者结束后才应该发送该信号
from multiprocessing import Process,Queueimport time,random,osdef consumer(q): while True: res=q.get() if res is None:break #收到结束信号则结束 time.sleep(random.randint(1,3)) print('\033[45m%s 吃 %s\033[0m' %(os.getpid(),res))def producer(q): for i in range(2): time.sleep(random.randint(1,3)) res='包子%s' %i q.put(res) print('\033[44m%s 生产了 %s\033[0m' %(os.getpid(),res))if __name__ == '__main__': q=Queue() #生产者们:即厨师们 p1=Process(target=producer,args=(q,)) #消费者们:即吃货们 c1=Process(target=consumer,args=(q,)) #开始 p1.start() c1.start() p1.join() q.put(None) #发送结束信号 print('主')
主进程在生产者生产完毕后发送结束信号None
但上述解决方式，在有多个生产者和多个消费者时，我们则需要用一个很low的方式去解决
from multiprocessing import Process,Queueimport time,random,osdef consumer(q): while True: res=q.get() if res is None:break #收到结束信号则结束 time.sleep(random.randint(1,3)) print('\033[45m%s 吃 %s\033[0m' %(os.getpid(),res))def producer(name,q): for i in range(2): time.sleep(random.randint(1,3)) res='%s%s' %(name,i) q.put(res) print('\033[44m%s 生产了 %s\033[0m' %(os.getpid(),res))if __name__ == '__main__': q=Queue() #生产者们:即厨师们 p1=Process(target=producer,args=('包子',q)) p2=Process(target=producer,args=('骨头',q)) p3=Process(target=producer,args=('泔水',q)) #消费者们:即吃货们 c1=Process(target=consumer,args=(q,)) c2=Process(target=consumer,args=(q,)) #开始 p1.start() p2.start() p3.start() c1.start() p1.join() #必须保证生产者全部生产完毕,才应该发送结束信号 p2.join() p3.join() q.put(None) #有几个消费者就应该发送几次结束信号None q.put(None) #发送结束信号 print('主')
多个消费者的例子：有几个消费者就需要发送几次结束信号
线程
线程介绍
有了进程为什么要有线程
进程有很多优点，它提供了多道编程，让我们感觉我们每个人都拥有自己的CPU和其他资源，可以提高计算机的利用率。很多人就不理解了，既然进程这么优秀，为什么还要线程呢？其实，仔细观察就会发现进程还是有很多缺陷的，主要体现在两点上：
-
进程只能在一个时间干一件事，如果想同时干两件事或多件事，进程就无能为力了。
-
进程在执行的过程中如果阻塞，例如等待输入，整个进程就会挂起，即使进程中有些工作不依赖于输入的数据，也将无法执行。
如果这两个缺点理解比较困难的话，举个现实的例子也许你就清楚了：如果把我们上课的过程看成一个进程的话，那么我们要做的是耳朵听老师讲课，手上还要记笔记，脑子还要思考问题，这样才能高效的完成听课的任务。而如果只提供进程这个机制的话，上面这三件事将不能同时执行，同一时间只能做一件事，听的时候就不能记笔记，也不能用脑子思考，这是其一；如果老师在黑板上写演算过程，我们开始记笔记，而老师突然有一步推不下去了，阻塞住了，他在那边思考着，而我们呢，也不能干其他事，即使你想趁此时思考一下刚才没听懂的一个问题都不行，这是其二。
现在你应该明白了进程的缺陷了，而解决的办法很简单，我们完全可以让听、写、思三个独立的过程，并行起来，这样很明显可以提高听课的效率。而实际的操作系统中，也同样引入了这种类似的机制——线程。
线程的出现
60年代，在OS中能拥有资源和独立运行的基本单位是进程，然而随着计算机技术的发展，进程出现了很多弊端，一是由于进程是资源拥有者，创建、撤消与切换存在较大的时空开销，因此需要引入 轻型进程 ；二是由于对称多处理机（SMP）出现， 可以满足多个运行单位 ，而多个进程并行开销过大。
因此在80年代，出现了 能独立运行的基本单位 ——线程（Threads） 。
注意：进程是资源分配的最小单位,线程是CPU调度的最小单位.
每一个进程中至少有一个线程。
进程和线程的关系
线程与进程的区别 可以归纳为以下4点：
1）地址空间和其它资源（如打开文件）：进程间相互独立，同一进程的各线程间共享。某进程内的线程在其它进程不可见。
3）调度和切换：线程上下文切换比进程上下文切换要快得多。
4）在多线程操作系统中，进程不是一个可执行的实体。
线程的特点
在多线程的操作系统中，通常是在一个进程中包括多个线程，每个线程都是作为利用CPU的基本单位，是花费最小开销的实体。线程具有以下属性。
1）轻型实体
线程中的实体基本上不拥有系统资源，只是有一点必不可少的、能保证独立运行的资源。
线程的实体包括程序、数据和TCB。线程是动态概念，它的动态特性由线程控制块TCB（Thread Control Block）描述。
TCB包括以下信息：（1）线程状态。（2）当线程不运行时，被保存的现场资源。（3）一组执行堆栈。（4）存放每个线程的局部变量主存区。（5）访问同一个进程中的主存和其它资源。用于指示被执行指令序列的程序计数器、保留局部变量、少数状态参数和返回地址等的一组寄存器和堆栈。
TCB包括以下信息
2）独立调度和分派的基本单位。
在多线程OS中，线程是能独立运行的基本单位，因而也是独立调度和分派的基本单位。由于线程很“轻”，故线程的切换非常迅速且开销小（在同一进程中的）。
3）共享进程资源。
线程在同一进程中的各个线程，都可以共享该进程所拥有的资源，这首先表现在：所有线程都具有相同的进程id，这意味着，线程可以访问该进程的每一个内存资源；此外，还可以访问进程所拥有的已打开文件、定时器、信号量机构等。由于同一个进程内的线程共享内存和文件，所以线程之间互相通信不必调用内核。
4 ）可并发执行。
在一个进程中的多个线程之间，可以并发执行，甚至允许在一个进程中所有线程都能并发执行；同样，不同进程中的线程也能并发执行，充分利用和发挥了处理机与外围设备并行工作的能力。
内存中的线程
多个线程共享同一个进程的地址空间中的资源，是对一台计算机上多个进程的模拟，有时也称线程为轻量级的进程。
而对一台计算机上多个进程，则共享物理内存、磁盘、打印机等其他物理资源。多线程的运行也多进程的运行类似，是cpu在多个线程之间的快速切换。
不同的进程之间是充满敌意的，彼此是抢占、竞争cpu的关系，如果迅雷会和QQ抢资源。而同一个进程是由一个程序员的程序创建，所以同一进程内的线程是合作关系，一个线程可以访问另外一个线程的内存地址，大家都是共享的，一个线程干死了另外一个线程的内存，那纯属程序员脑子有问题。
类似于进程，每个线程也有自己的堆栈，不同于进程，线程库无法利用时钟中断强制线程让出CPU，可以调用thread_yield运行线程自动放弃cpu，让另外一个线程运行。
线程通常是有益的，但是带来了不小程序设计难度，线程的问题是：
1. 父进程有多个线程，那么开启的子线程是否需要同样多的线程
2. 在同一个进程中，如果一个线程关闭了文件，而另外一个线程正准备往该文件内写内容呢？
因此，在多线程的代码中，需要更多的心思来设计程序的逻辑、保护程序的数据。
在python中使用线程
threading模块
multiprocess模块的完全模仿了threading模块的接口，二者在使用层面，有很大的相似性，因而不再详细介绍（ 官方链接 ）
线程的创建
from threading import Threadimport timedef sayhi(name): time.sleep(2) print('%s say hello' %name)if __name__ == '__main__': t=Thread(target=sayhi,args=('egon',)) t.start() print('主线程')
创建线程的方式1
from threading import Threadimport timeclass Sayhi(Thread): def __init__(self,name): super().__init__() self.name=name def run(self): time.sleep(2) print('%s say hello' % self.name)if __name__ == '__main__': t = Sayhi('egon') t.start() print('主线程')
创建线程的方式2
多线程与多进程
from threading import Threadfrom multiprocessing import Processimport osdef work(): print('hello',os.getpid())if __name__ == '__main__': #part1:在主进程下开启多个线程,每个线程都跟主进程的pid一样 t1=Thread(target=work) t2=Thread(target=work) t1.start() t2.start() print('主线程/主进程pid',os.getpid()) #part2:开多个进程,每个进程都有不同的pid p1=Process(target=work) p2=Process(target=work) p1.start() p2.start() print('主线程/主进程pid',os.getpid())
pid的比较
from threading import Threadfrom multiprocessing import Processimport osdef work(): print('hello')if __name__ == '__main__': #在主进程下开启线程 t=Thread(target=work) t.start() print('主线程/主进程') ''' 打印结果: hello 主线程/主进程 ''' #在主进程下开启子进程 t=Process(target=work) t.start() print('主线程/主进程') ''' 打印结果: 主线程/主进程 hello '''
开启效率的较量
from threading import Threadfrom multiprocessing import Processimport osdef work(): global n n=0if __name__ == '__main__': # n=100 # p=Process(target=work) # p.start() # p.join() # print('主',n) #毫无疑问子进程p已经将自己的全局的n改成了0,但改的仅仅是它自己的,查看父进程的n仍然为100 n=1 t=Thread(target=work) t.start() t.join() print('主',n) #查看结果为0,因为同一进程内的线程之间共享进程内的数据同一进程内的线程共享该进程的数据？
内存数据的共享问题
练习 ：多线程实现socket
#_*_coding:utf-8_*_#!/usr/bin/env pythonimport multiprocessingimport threadingimport sockets=socket.socket(socket.AF_INET,socket.SOCK_STREAM)s.bind(('127.0.0.1',8080))s.listen(5)def action(conn): while True: data=conn.recv(1024) print(data) conn.send(data.upper())if __name__ == '__main__': while True: conn,addr=s.accept() p=threading.Thread(target=action,args=(conn,)) p.start()
server
#_*_coding:utf-8_*_#!/usr/bin/env pythonimport sockets=socket.socket(socket.AF_INET,socket.SOCK_STREAM)s.connect(('127.0.0.1',8080))while True: msg=input('>>: ').strip() if not msg:continue s.send(msg.encode('utf-8')) data=s.recv(1024) print(data)
client
Thread类的其他方法
Thread实例对象的方法 # isAlive(): 返回线程是否活动的。 # getName(): 返回线程名。 # setName(): 设置线程名。threading模块提供的一些方法： # threading.currentThread(): 返回当前的线程变量。 # threading.enumerate(): 返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前，不包括启动前和终止后的线程。 # threading.activeCount(): 返回正在运行的线程数量，与len(threading.enumerate())有相同的结果。
from threading import Threadimport threadingfrom multiprocessing import Processimport osdef work(): import time time.sleep(3) print(threading.current_thread().getName())if __name__ == '__main__': #在主进程下开启线程 t=Thread(target=work) t.start() print(threading.current_thread().getName()) print(threading.current_thread()) #主线程 print(threading.enumerate()) #连同主线程在内有两个运行的线程 print(threading.active_count()) print('主线程/主进程') ''' 打印结果: MainThread <_MainThread(MainThread, started 140735268892672)> [<_MainThread(MainThread, started 140735268892672)>, <Thread(Thread-1, started 123145307557888)>] 主线程/主进程 Thread-1 '''
代码示例
from threading import Threadimport timedef sayhi(name): time.sleep(2) print('%s say hello' %name)if __name__ == '__main__': t=Thread(target=sayhi,args=('egon',)) t.start() t.join() print('主线程') print(t.is_alive()) ''' egon say hello 主线程 False '''
join方法
守护线程
无论是进程还是线程，都遵循：守护xx会等待主xx运行完毕后被销毁。 需要强调的是：运行完毕并非终止运行
#1.对主进程来说，运行完毕指的是主进程代码运行完毕#2.对主线程来说，运行完毕指的是主线程所在的进程内所有非守护线程统统运行完毕，主线程才算运行完毕
#1 主进程在其代码结束后就已经算运行完毕了（守护进程在此时就被回收）,然后主进程会一直等非守护的子进程都运行完毕后回收子进程的资源(否则会产生僵尸进程)，才会结束，#2 主线程在其他非守护线程运行完毕后才算运行完毕（守护线程在此时就被回收）。因为主线程的结束意味着进程的结束，进程整体的资源都将被回收，而进程必须保证非守护线程都运行完毕后才能结束。
详细解释
from threading import Threadimport timedef sayhi(name): time.sleep(2) print('%s say hello' %name)if __name__ == '__main__': t=Thread(target=sayhi,args=('egon',)) t.setDaemon(True) #必须在t.start()之前设置 t.start() print('主线程') print(t.is_alive()) ''' 主线程 True '''
守护线程例1
from threading import Threadimport timedef foo(): print(123) time.sleep(1) print("end123")def bar(): print(456) time.sleep(3) print("end456")t1=Thread(target=foo)t2=Thread(target=bar)t1.daemon=Truet1.start()t2.start()print("main-------")
守护线程例2
同步锁
from threading import Threadimport os,timedef work(): global n temp=n time.sleep(0.1) n=temp-1if __name__ == '__main__': n=100 l=[] for i in range(100): p=Thread(target=work) l.append(p) p.start() for p in l: p.join() print(n) #结果可能为99
多个线程抢占资源的情况
import threadingR=threading.Lock()R.acquire()'''对公共数据的操作'''R.release()
from threading import Thread,Lockimport os,timedef work(): global n lock.acquire() temp=n time.sleep(0.1) n=temp-1 lock.release()if __name__ == '__main__': lock=Lock() n=100 l=[] for i in range(100): p=Thread(target=work) l.append(p) p.start() for p in l: p.join() print(n) #结果肯定为0，由原来的并发执行变成串行，牺牲了执行效率保证了数据安全
同步锁的引用
#不加锁:并发执行,速度快,数据不安全from threading import current_thread,Thread,Lockimport os,timedef task(): global n print('%s is running' %current_thread().getName()) temp=n time.sleep(0.5) n=temp-1if __name__ == '__main__': n=100 lock=Lock() threads=[] start_time=time.time() for i in range(100): t=Thread(target=task) threads.append(t) t.start() for t in threads: t.join() stop_time=time.time() print('主:%s n:%s' %(stop_time-start_time,n))'''Thread-1 is runningThread-2 is running......Thread-100 is running主:0.5216062068939209 n:99'''#不加锁:未加锁部分并发执行,加锁部分串行执行,速度慢,数据安全from threading import current_thread,Thread,Lockimport os,timedef task(): #未加锁的代码并发运行 time.sleep(3) print('%s start to run' %current_thread().getName()) global n #加锁的代码串行运行 lock.acquire() temp=n time.sleep(0.5) n=temp-1 lock.release()if __name__ == '__main__': n=100 lock=Lock() threads=[] start_time=time.time() for i in range(100): t=Thread(target=task) threads.append(t) t.start() for t in threads: t.join() stop_time=time.time() print('主:%s n:%s' %(stop_time-start_time,n))'''Thread-1 is runningThread-2 is running......Thread-100 is running主:53.294203758239746 n:0'''#有的同学可能有疑问:既然加锁会让运行变成串行,那么我在start之后立即使用join,就不用加锁了啊,也是串行的效果啊#没错:在start之后立刻使用jion,肯定会将100个任务的执行变成串行,毫无疑问,最终n的结果也肯定是0,是安全的,但问题是#start后立即join:任务内的所有代码都是串行执行的,而加锁,只是加锁的部分即修改共享数据的部分是串行的#单从保证数据安全方面,二者都可以实现,但很明显是加锁的效率更高.from threading import current_thread,Thread,Lockimport os,timedef task(): time.sleep(3) print('%s start to run' %current_thread().getName()) global n temp=n time.sleep(0.5) n=temp-1if __name__ == '__main__': n=100 lock=Lock() start_time=time.time() for i in range(100): t=Thread(target=task) t.start() t.join() stop_time=time.time() print('主:%s n:%s' %(stop_time-start_time,n))'''Thread-1 start to runThread-2 start to run......Thread-100 start to run主:350.6937336921692 n:0 #耗时是多么的恐怖'''）
互斥锁与join的区别
死锁与可重入锁
进程也有死锁与可重入锁，使用方法都是一样的，所以放到这里一起说：
所谓死锁： 是指两个或两个以上的进程或线程在执行过程中，因争夺资源而造成的一种互相等待的现象，若无外力作用，它们都将无法推进下去。此时称系统处于死锁状态或系统产生了死锁，这些永远在互相等待的进程称为死锁进程，如下就是死锁。
from threading import Lock as Lockimport timemutexA=Lock()mutexA.acquire()mutexA.acquire()print(123)mutexA.release()mutexA.release()
死锁
解决方法：使用可重入锁，在Python中为了支持在同一线程中多次请求同一资源，提供了可重入锁RLock。
这个RLock内部维护着一个Lock和一个counter变量，counter记录了acquire的次数，从而使得资源可以被多次require。直到一个线程所有的acquire都被release，其他的线程才能获得资源。上面的例子如果使用RLock代替Lock，则不会发生死锁：
from threading import RLockimport timemutexA=RLock()mutexA.acquire()mutexA.acquire()print(123)mutexA.release()mutexA.release()
可重入锁示例
典型问题：科学家吃面
import timefrom threading import Thread,Locknoodle_lock = Lock()fork_lock = Lock()def eat1(name): noodle_lock.acquire() print('%s 抢到了面条'%name) fork_lock.acquire() print('%s 抢到了叉子'%name) print('%s 吃面'%name) fork_lock.release() noodle_lock.release()def eat2(name): fork_lock.acquire() print('%s 抢到了叉子' % name) time.sleep(1) noodle_lock.acquire() print('%s 抢到了面条' % name) print('%s 吃面' % name) noodle_lock.release() fork_lock.release()for name in ['哪吒','egon','yuan']: t1 = Thread(target=eat1,args=(name,)) t2 = Thread(target=eat2,args=(name,)) t1.start() t2.start()
死锁问题
import timefrom threading import Thread,RLockfork_lock = noodle_lock = RLock()def eat1(name): noodle_lock.acquire() print('%s 抢到了面条'%name) fork_lock.acquire() print('%s 抢到了叉子'%name) print('%s 吃面'%name) fork_lock.release() noodle_lock.release()def eat2(name): fork_lock.acquire() print('%s 抢到了叉子' % name) time.sleep(1) noodle_lock.acquire() print('%s 抢到了面条' % name) print('%s 吃面' % name) noodle_lock.release() fork_lock.release()for name in ['哪吒','egon','yuan']: t1 = Thread(target=eat1,args=(name,)) t2 = Thread(target=eat2,args=(name,)) t1.start() t2.start()
可重入锁解决死锁问题
全局解释器锁GIL
GIL介绍
首先需要明确的一点是GIL并不是Python的特性，它是在实现Python解析器(CPython)时所引入的一个概念。
Python也一样，同样一段代码可以通过CPython，PyPy，Psyco等不同的Python执行环境来执行。像其中的JPython就没有GIL。然而因为CPython是大部分环境下默认的Python执行环境。所以在很多人的概念里CPython就是Python，也就想当然的把GIL归结为Python语言的缺陷。所以这里要先明确一点：GIL并不是Python的特性，Python完全可以不依赖于GIL。
简单来说，在Cpython解释器中，因为有GIL锁的存在同一个进程下开启的多线程，同一时刻只能有一个线程执行，无法利用多核优势。
常见问题1
我们有了GIL锁为什么还要自己在代码中加锁呢？
GIL保护的是Python解释器级别的数据资源，自己代码中的数据资源就需要自己加锁防止竞争。如下图：
常见问题2
有了GIL的存在，同一时刻同一进程中只有一个线程被执行，进程可以利用多核，但是开销大，而Python的多线程开销小，但却无法利用多核优势，也就是说Python这语言难堪大用。
其实编程所解决的现实问题大致分为IO密集型和计算密集型。
对于IO密集型的场景，Python的多线程编程完全OK，而对于计算密集型的场景，Python中有很多成熟的模块或框架如Pandas等能够提高计算效率。
推荐阅读：理解GIL
池 —— concurrent.futures
Python标准模块--concurrent.futures
https://docs.python.org/dev/library/concurrent.futures.html
concurrent.futures模块提供了高度封装的异步调用接口，其中：
ThreadPoolExecutor：线程池
ProcessPoolExecutor: 进程池
借助上面两个类，我们可以很方便地创建进程池对象和线程池对象。
p_pool = ProcessPoolExecutor(max_workers=5) # 创建一个最多5个woker的进程池t_pool = ThreadPoolExecutor(max_workers=5) # 创建一个最多5个woker的线程池
可用方法介绍：
# 基本方法#submit(fn, *args, **kwargs)提交任务# map(func, *iterables, timeout=None, chunksize=1) 取代for循环submit的操作# shutdown(wait=True) 相当于进程池的pool.close()+pool.join()操作wait=True，等待池内所有任务执行完毕回收完资源后才继续wait=False，立即返回，并不会等待池内的任务执行完毕但不管wait参数为何值，整个程序都会等到所有任务执行完毕submit和map必须在shutdown之前# result(timeout=None)取得结果# add_done_callback(fn)回调函数
#介绍The ProcessPoolExecutor class is an Executor subclass that uses a pool of processes to execute calls asynchronously. ProcessPoolExecutor uses the multiprocessing module, which allows it to side-step the Global Interpreter Lock but also means that only picklable objects can be executed and returned.class concurrent.futures.ProcessPoolExecutor(max_workers=None, mp_context=None)An Executor subclass that executes calls asynchronously using a pool of at most max_workers processes. If max_workers is None or not given, it will default to the number of processors on the machine. If max_workers is lower or equal to 0, then a ValueError will be raised.#用法from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutorimport os,time,randomdef task(n): print('%s is runing' %os.getpid()) time.sleep(random.randint(1,3)) return n**2if __name__ == '__main__': executor=ProcessPoolExecutor(max_workers=3) futures=[] for i in range(11): future=executor.submit(task,i) futures.append(future) executor.shutdown(True) print('+++>') for future in futures: print(future.result())
ProcessPoolExecutor
#介绍ThreadPoolExecutor is an Executor subclass that uses a pool of threads to execute calls asynchronously.class concurrent.futures.ThreadPoolExecutor(max_workers=None, thread_name_prefix='')An Executor subclass that uses a pool of at most max_workers threads to execute calls asynchronously.Changed in version 3.5: If max_workers is None or not given, it will default to the number of processors on the machine, multiplied by 5, assuming that ThreadPoolExecutor is often used to overlap I/O instead of CPU work and the number of workers should be higher than the number of workers for ProcessPoolExecutor.New in version 3.6: The thread_name_prefix argument was added to allow users to control the threading.Thread names for worker threads created by the pool for easier debugging.#用法与ProcessPoolExecutor相同
ThreadPoolExecutor
使用线程池改进网络编程的例子：
import socketfrom concurrent.futures import ThreadPoolExecutort_pool = ThreadPoolExecutor(max_workers=3)server = socket.socket()server.bind(('127.0.0.1', 8080))server.listen(5)def comm(conn): while 1: try: data = conn.recv(1024) if not data: break conn.send(data.upper()) except ConnectionResetError: break conn.close()def serve(): while 1: conn, addr = server.accept() t_pool.submit(comm, conn) server.close()if __name__ == '__main__': serve()
server端
import socketclient = socket.socket()client.connect(('127.0.0.1', 8080))while 1: msg = input('>>>').strip() if not msg: continue client.send(msg.encode('utf8')) data = client.recv(1024) print(data.decode('utf8'))client.close()
client端
其他方法：
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutorimport os,time,randomdef task(n): print('%s is runing' %os.getpid()) time.sleep(random.randint(1,3)) return n**2if __name__ == '__main__': executor=ThreadPoolExecutor(max_workers=3) # for i in range(11): # future=executor.submit(task,i) executor.map(task,range(1,12)) #map取代了for+submit
map的用法
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutorfrom multiprocessing import Poolimport requestsimport jsonimport osdef get_page(url): print('<进程%s> get %s' %(os.getpid(),url)) respone=requests.get(url) if respone.status_code == 200: return {'url':url,'text':respone.text}def parse_page(res): res=res.result() print('<进程%s> parse %s' %(os.getpid(),res['url'])) parse_res='url:<%s> size:[%s]\n' %(res['url'],len(res['text'])) with open('db.txt','a') as f: f.write(parse_res)if __name__ == '__main__': urls=[ 'https://www.baidu.com', 'https://www.python.org', 'https://www.openstack.org', 'https://help.github.com/', 'http://www.sina.com.cn/' ] # p=Pool(3) # for url in urls: # p.apply_async(get_page,args=(url,),callback=pasrse_page) # p.close() # p.join() p=ProcessPoolExecutor(3) for url in urls: p.submit(get_page,url).add_done_callback(parse_page) #parse_page拿到的是一个future对象obj，需要用obj.result()拿到结果
回调函数
还没有人抢沙发呢~