Cpython解释器下实现并发编程——多进程、多线程、协程、IO模型

 1 from multiprocessing import Process
 2 import time,random
 3 
 4 class MyProcess(Process):
 5     def __init__(self,name):
 6         super(MyProcess,self).__init__()
 7         self.name=name
 8 
 9     def run(self):
10         print(‘%s is piaoing‘ %self.name)
11         time.sleep(random.randint(1,3))
12         print(‘%s is over‘ % self.name)
13 
14 
15 if __name__ == ‘__main__‘:
16     p=MyProcess(‘P1‘)
17     p.start() #仅仅只是向操作系统发送了一个创建子进程p的信号
18     print(‘主‘)

Since Windows has no fork, the multiprocessing module starts a new Python process and imports the calling module. 
If Process() gets called upon import, then this sets off an infinite succession of new processes (or until your machine runs out of resources). 
This is the reason for hiding calls to Process() inside
if __name__ == "__main__"
since statements inside this if-statement will not get called upon import.
由于Windows没有fork，多处理模块启动一个新的Python进程并导入调用模块。 
如果在导入时调用Process（），那么这将启动无限继承的新进程（或直到机器耗尽资源）。 
这是隐藏对Process（）内部调用的原，使用if __name__ == “__main __”，这个if语句中的语句将不会在导入时被调用。

 1 from multiprocessing import Process
 2 import time,random
 3 
 4 def piao(name):
 5     print(‘%s is piaoing‘ %name)
 6     time.sleep(random.randint(1,3))
 7     print(‘%s is over‘ % name)
 8 
 9 
10 if __name__ == ‘__main__‘:
11     # p=Process(target=piao,kwargs={‘name‘:‘alex‘})
12     p=Process(target=piao,args=(‘alex‘,))
13     p.start() #仅仅只是向操作系统发送了一个创建子进程p的信号
14     print(‘主‘)

 1 from multiprocessing import Process
 2 import time,random
 3 
 4 class MyProcess(Process):
 5     def __init__(self,name):
 6         super(MyProcess,self).__init__()
 7         self.name=name
 8 
 9     def run(self):
10         print(‘%s is piaoing‘ %self.name)
11         time.sleep(random.randint(1,3))
12         print(‘%s is over‘ % self.name)
13 
14 
15 if __name__ == ‘__main__‘:
16     p=MyProcess(‘P1‘)
17     p.start() #仅仅只是向操作系统发送了一个创建子进程p的信号
18     print(‘主‘)

 1 from socket import *
 2 from multiprocessing import Process
 3 
 4 server=socket(AF_INET,SOCK_STREAM)
 5 server.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
 6 server.bind((‘127.0.0.1‘,8080))
 7 server.listen(5)
 8 
 9 def talk(conn,client_addr):
10     while True:
11         try:
12             msg=conn.recv(1024)
13             if not msg:break
14             conn.send(msg.upper())
15         except Exception:
16             break
17 
18 if __name__ == ‘__main__‘: #windows下start进程一定要写到这下面
19     while True:
20         conn,client_addr=server.accept()
21         p=Process(target=talk,args=(conn,client_addr))
22         p.start()

 1 from socket import *
 2 
 3 client=socket(AF_INET,SOCK_STREAM)
 4 client.connect((‘127.0.0.1‘,8080))
 5 
 6 
 7 while True:
 8     msg=input(‘>>: ‘).strip()
 9     if not msg:continue
10 
11     client.send(msg.encode(‘utf-8‘))
12     msg=client.recv(1024)
13     print(msg.decode(‘utf-8‘))

 1 from multiprocessing import Process
 2 import time
 3 import random
 4 
 5 class Piao(Process):
 6     def __init__(self,name):
 7         self.name=name
 8         super().__init__()
 9     def run(self):
10         print(‘%s is piaoing‘ %self.name)
11         time.sleep(random.randrange(1,3))
12         print(‘%s is piao end‘ %self.name)
13 
14 
15 p=Piao(‘egon‘)
16 p.start()
17 p.join(0.0001) #等待p停止,等0.0001秒就不再等了
18 print(‘开始‘)

 1 from multiprocessing import Process
 2 import time
 3 import random
 4 def piao(name):
 5     print(‘%s is piaoing‘ %name)
 6     time.sleep(random.randint(1,3))
 7     print(‘%s is piao end‘ %name)
 8 
 9 p1=Process(target=piao,args=(‘egon‘,))
10 p2=Process(target=piao,args=(‘alex‘,))
11 p3=Process(target=piao,args=(‘yuanhao‘,))
12 p4=Process(target=piao,args=(‘wupeiqi‘,))
13 
14 p1.start()
15 p2.start()
16 p3.start()
17 p4.start()
18 
19 #有的同学会有疑问:既然join是等待进程结束,那么我像下面这样写,进程不就又变成串行的了吗?
20 #当然不是了,必须明确：p.join()是让谁等？
21 #很明显p.join()是让主线程等待p的结束，卡住的是主线程而绝非进程p，
22 
23 #详细解析如下：
24 #进程只要start就会在开始运行了,所以p1-p4.start()时,系统中已经有四个并发的进程了
25 #而我们p1.join()是在等p1结束,没错p1只要不结束主线程就会一直卡在原地,这也是问题的关键
26 #join是让主线程等,而p1-p4仍然是并发执行的,p1.join的时候,其余p2,p3,p4仍然在运行,等#p1.join结束,可能p2,p3,p4早已经结束了,这样p2.join,p3.join.p4.join直接通过检测，无需等待
27 # 所以4个join花费的总时间仍然是耗费时间最长的那个进程运行的时间
28 p1.join()
29 p2.join()
30 p3.join()
31 p4.join()
32 
33 print(‘主线程‘)
34 
35 
36 #上述启动进程与join进程可以简写为
37 # p_l=[p1,p2,p3,p4]
38 # 
39 # for p in p_l:
40 #     p.start()
41 # 
42 # for p in p_l:
43 #     p.join()

 1 from multiprocessing import Process
 2 import time
 3 import random
 4 
 5 class Piao(Process):
 6     def __init__(self,name):
 7         self.name=name
 8         super().__init__()
 9     def run(self):
10         print(‘%s is piaoing‘ %self.name)
11         time.sleep(random.randrange(1,3))
12         print(‘%s is piao end‘ %self.name)
13 
14 
15 p=Piao(‘egon‘)
16 p.daemon=True #一定要在p.start()前设置,设置p为守护进程,禁止p创建子进程,并且父进程代码执行结束,p即终止运行
17 p.start()
18 print(‘主‘)

 1 from multiprocessing import Process
 2 import time
 3 
 4 def foo():
 5     print(123)
 6     time.sleep(1)
 7     print(‘end123‘)
 8 
 9 def bar():
10     print(456)
11     time.sleep(3)
12     print(‘end456‘)
13 
14 if __name__ == ‘__main__‘:
15     p1 = Process(target=foo)
16     p2 = Process(target=bar)
17 
18     # p1.daemon = False
19     p1.daemon = True #一定要在p1.start()前设置,设置p1为守护进程,禁止p1创建子进程,并且父进程代码执行结束,p1即终止运行
20     p1.start()
21     p2.start()
22 
23     print("主")
24 # 打印该行则主进程代码结束，则守护进程p1应该被终止，可能会有p1任务执行的打印信息123，因为主进程打印"主"，p1也执行了，但是随即被终止

 1 from multiprocessing import Process
 2 import os,time
 3 def work():
 4     print(‘%s is running‘ %os.getpid())
 5     time.sleep(2)
 6     print(‘%s is done‘ %os.getpid())
 7 
 8 if __name__ == ‘__main__‘:
 9     for i in range(3):
10         p=Process(target=work)
11         p.start()

 1 from multiprocessing import Process,Lock
 2 import time
 3 
 4 def work(name,mutex):
 5     mutex.acquire()  #加锁,可以设定锁的时间
 6     print(‘task %s is running‘ %name)
 7     time.sleep(2)
 8     print(‘task %s is done‘ %name)
 9     mutex.release() #解锁
10 
11 if __name__ == ‘__main__‘:
12     mutex = Lock()
13     p1 = Process(target=work,args=(‘wzs‘,mutex))  #args传参数以元组的实行传
14     p2 = Process(target=work,args=(‘egon‘,mutex))  #args传参数以元组的实行传
15     p1.start()
16     p2.start()
17 
18     print(‘主‘)

1 {"count": 6}

 1 import json
 2 import os
 3 import time
 4 from multiprocessing import Process,Lock
 5 def search():
 6     dic = json.load(open(‘db.txt‘))
 7     print(‘\033[32m[%s] 看到剩余票数<%s> \033[0m‘ %(os.getpid(),dic[‘count‘]))
 8 
 9 def get_tickert():
10     ‘‘‘
11     模拟抢票过程中，数据库操作
12     :return:
13     ‘‘‘
14     dic = json.load(open(‘db.txt‘))
15     time.sleep(0.5)  ##模拟读取数据库网络延迟
16     if dic[‘count‘] > 0:
17         dic[‘count‘] -= 1
18         time.sleep(0.5) ##模拟写数据库的网络延迟
19         json.dump(dic,open(‘db.txt‘,‘w‘))
20         print(‘\033[31m[%s] 购票成功 \033[0m‘ %os.getpid())
21 
22 def task(mutex):
23     ‘‘‘
24     抢票过程，加锁（多人操作（读写）同一份数据）
25     :param mutex:
26     :return:
27     ‘‘‘
28     search()
29     mutex.acquire()
30     get_tickert()
31     mutex.release()
32 
33 
34 if __name__ == ‘__main__‘:
35     mutex = Lock()
36     for i in range(10):
37         p = Process(target=task,args=(mutex,))
38         p.start()

 1 from multiprocessing import Process,Queue
 2 import time,os
 3 def producer(q,name):
 4     for i in range(3):
 5         time.sleep(1)
 6         res=‘%s%s‘ %(name,i)
 7         q.put(res)
 8         print(‘\033[45m<%s> 生产了 [%s]\033[0m‘ %(os.getpid(),res))
 9 
10 
11 def consumer(q):
12     while True:
13         res=q.get()
14         if res is None:break
15         time.sleep(1.5)
16         print(‘\033[34m<%s> 吃了 [%s]\033[0m‘ % (os.getpid(), res))
17 
18 if __name__ == ‘__main__‘:
19     q=Queue()
20     #生产者们：即厨师们
21     p1=Process(target=producer,args=(q,‘包子‘))
22     p2=Process(target=producer,args=(q,‘饺子‘))
23     p3=Process(target=producer,args=(q,‘馄饨‘))
24 
25     #消费者们：即吃货们
26     c1=Process(target=consumer,args=(q,))
27     c2=Process(target=consumer,args=(q,))
28 
29     p1.start()
30     p2.start()
31     p3.start()
32     c1.start()
33     c2.start()
34 
35     p1.join()
36     p2.join()
37     p3.join()
38     q.put(None)
39     q.put(None)
40 
41     print(‘主‘)

 1 from multiprocessing import Process, JoinableQueue
 2 import time, os
 3 
 4 
 5 def producer(q, name):
 6     for i in range(3):
 7         time.sleep(1)
 8         res = ‘%s%s‘ % (name, i)
 9         q.put(res)
10         print(‘\033[45m<%s> 生产了 [%s]\033[0m‘ % (os.getpid(), res))
11     q.join()
12 
13 def consumer(q):
14     while True:
15         res = q.get()
16         time.sleep(1.5)
17         print(‘\033[34m<%s> 吃了 [%s]\033[0m‘ % (os.getpid(), res))
18         q.task_done()
19 
20 if __name__ == ‘__main__‘:
21     q = JoinableQueue()
22 
23     # 生产者们：即厨师们
24     p1 = Process(target=producer, args=(q, ‘包子‘))
25     p2 = Process(target=producer, args=(q, ‘饺子‘))
26     p3 = Process(target=producer, args=(q, ‘馄饨‘))
27 
28     # 消费者们：即吃货们
29     c1 = Process(target=consumer, args=(q,))
30     c2 = Process(target=consumer, args=(q,))
31 
32     c1.daemon=True
33     c2.daemon=True
34     p1.start()
35     p2.start()
36     p3.start()
37     c1.start()
38     c2.start()
39 
40 
41     p1.join()
42 
43     print(‘主‘)

 1 from multiprocessing import Process,Manager,Lock
 2 # 不加锁，数据就乱了,是并行的
 3 def task(dic):
 4     dic[‘count‘] -=1
 5 
 6 if __name__ == ‘__main__‘:
 7     m = Manager()
 8     dic = m.dict({‘count‘:100})
 9     p_l = []
10     for i in range(100):
11         p = Process(target=task,args=(dic,))
12         p_l.append(p)
13         p.start()
14     for p in p_l:
15         p.join()
16     print(dic)

 1 from multiprocessing import Process,Manager,Lock
 2 # 加锁，保证数据安全（处理效率会低）
 3 def task(dic,mutex):
 4     with mutex:
 5         dic[‘count‘] -=1
 6 
 7 if __name__ == ‘__main__‘:
 8     mutex = Lock()
 9     m = Manager()
10     dic = m.dict({‘count‘:100})
11     p_l = []
12     for i in range(100):
13         p = Process(target=task,args=(dic,mutex))
14         p_l.append(p)
15         p.start()
16     for p in p_l:
17         p.join()
18     print(dic)

1 p.apply(func [, args [, kwargs]]):在一个池工作进程中执行func(*args,**kwargs),然后返回结果。需要强调的是：此操作并不会在所有池工作进程中并执行func函数。如果要通过不同参数并发地执行func函数，必须从不同线程调用p.apply()函数或者使用p.apply_async()
2 p.apply_async(func [, args [, kwargs]]):在一个池工作进程中执行func(*args,**kwargs),然后返回结果。此方法的结果是AsyncResult类的实例，callback是可调用对象，接收输入参数。当func的结果变为可用时，将理解传递给callback。callback禁止执行任何阻塞操作，否则将接收其他异步操作中的结果。
3    
4 p.close():关闭进程池，防止进一步操作。如果所有操作持续挂起，它们将在工作进程终止前完成
5 P.join():等待所有工作进程退出。此方法只能在close（）或teminate()之后调用

1 方法apply_async()和map_async（）的返回值是AsyncResul的实例obj。实例具有以下方法
2 obj.get():返回结果，如果有必要则等待结果到达。timeout是可选的。如果在指定时间内还没有到达，将引发一场。如果远程操作中引发了异常，它将在调用此方法时再次被引发。
3 obj.ready():如果调用完成，返回True
4 obj.successful():如果调用完成且没有引发异常，返回True，如果在结果就绪之前调用此方法，引发异常
5 obj.wait([timeout]):等待结果变为可用。
6 obj.terminate()：立即终止所有工作进程，同时不执行任何清理或结束任何挂起工作。如果p被垃圾回收，将自动调用此函数