1 from multiprocessing import Process
 2 import os
 3 import time
 4
 5 def func(arg1):
 6     print("开启子进程的函数")
 7     print("主进程的子进程：",os.getpid())  # 查看当前进程的进程号
 8     print("子进程的父进程：", os.getppid())
 9     time.sleep(2)
10     print("%s是个pig"%arg1)
11
12 if __name__ == '__main__':
13     p = Process(target=func,args=("yl",))
14     #创建一个进程,传入参数的时候必须传元组的形式
15     # p是一个进程对象,还没有启动进程
16     p.start()   #开启了一个子进程
17     print('主进程 :',os.getpid())
18     # 此处的进程号等于func函数中的子进程的父进程的进程号
19     print('主进程的父进程 :',os.getppid()) # 查看当前进程的父进程
20     print("我是主进程")
21
22 # 输出结果：
23 主进程 : 80268
24 主进程的父进程 : 50360
25 我是主进程
26 开启子进程的函数
27 主进程的子进程： 77768
28 子进程的父进程： 80268
29 yl是个pig
30
31 # 开启了子进程的主进程 :
32     # 主进程自己的代码如果长,等待自己的代码执行结束,
33     # 子进程的执行时间长,主进程会在主进程代码执行完毕之后等待子进程执行完毕之后 主进程才结束

 1 import time
 2 from multiprocessing import Process
 3
 4 def func(arg1,arg2):
 5     print('*'*arg1)
 6     time.sleep(2)
 7     print('*'*arg2)
 8
 9 if __name__ == '__main__':
10     p = Process(target=func,args=(10,20))
11     p.start()
12     print('hahahaha')
13     p.join()     # 是感知一个子进程的结束,将异步的程序改为同步
14     # 相当于主进程在此处被阻塞，等待子进程运行结束之后再运行此处后面的代码
15     print('====== : 运行完了')
16
17 # 输出结果：
18 hahahaha
19 **********
20 ********************
21 ====== : 运行完了

 1 import os
 2 from multiprocessing import Process
 3
 4 def func():
 5     global n   # 声明了一个全局变量
 6     n = 0       # 重新定义了一个n
 7     print('pid : %s'%os.getpid(),n)
 8
 9 if __name__ == '__main__':
10     n = 100
11     p = Process(target=func)
12     p.start()
13     p.join()
14     print(os.getpid(),n)
15     #此处的n仍然为，主进程的n，未受到子进程创建的全局变量的影响
16
17 # 执行结果：
18 # pid : 884 0
19 # 11932 100
20
21 #多进程之间的数据是隔离的，互不影响

 1 # 可以传入参数的
 2 from multiprocessing import Process
 3
 4 class MyProcess(Process):   #继承Process类
 5     def __init__(self,arg1,arg2):
 6         super().__init__()  # 调用父类的__init__方法（必须写，否则报错）
 7         self.arg1 = arg1
 8         self.arg2 = arg2
 9
10     # 必须有run方法，子进程执行的就是run方法中的代码
11     def run(self):
12         print(self.pid)     #打印进程的进程号
13         print(self.name)    #打印进程的进程名，等同于os.getpid()
14         print(self.arg1)
15         print(self.arg2)
16
17 if __name__ == '__main__':
18     p1 = MyProcess(1,2)
19     p1.start()
20     p2 = MyProcess(3,4)
21     p2.start()
22 使用面向对象中类的知识来创建进程

 1 from multiprocessing import Process
 2 import time
 3 import os
 4
 5 def func(arg1,arg2):
 6     print('*'*arg1)
 7     time.sleep(2)
 8     print('*'*arg2)
 9
10 if __name__ == '__main__':
11     p = Process(target=func,args=(5,10))
12     p.start()
13     p1 = Process(target=func,args=(5,10))
14     p1.start()
15     p2 = Process(target=func,args=(5,10))
16     p2.start()
17     p3 = Process(target=func,args=(5,10))
18     p3.start()
19
20 # 开启了4个子进程，4个子进程同时运行,运行顺序不能确定先后顺序
21 #
22 # if __name__ == '__main__':
23 #     for i in range(10):
24 #         p = Process(target=func, args=(5*i, 10*i))
25 #         p.start()
26 # # 开启的多个子进程的运行顺序没有先后（但是是按照顺序给操作系统发的指令开启进程）
27 #         # p.join()    # 多个进程变成同步，执行效率低
28 #
29 #     # p.join()
30 # 此处的p相当于p9，总是在i等于9时开的之进程执行完成之后执行 print('执行完了'),
31 # 因为所开的子进程不按顺序执行，所以不能保证 print('执行完了') 最后执行
32 #     print('执行完了')
33
34
35 # if __name__ == '__main__':
36 #     p_lst = []
37 #     for i in range(10):
38 #         p = Process(target=func,args=(5*i,10*i))
39 #         p_lst.append(p)
40 #         p.start()
41 #
42 #     [p.join() for p in p_lst]
43 # （列表推导式）之前的所有进程必须在这里都执行完才能执行下面的代码
44 #     print('执行完了')
45 # 这种方法可以保证 print('执行完了')，是最后执行。

 1 # 子进程 -- > 守护进程
 2 # 主进程代码执行结束守护进程立即结束
 3 import time
 4 from multiprocessing import Process
 5
 6 def func():
 7     while True:
 8         time.sleep(0.2)
 9         print('我还活着')
10
11 def func2():
12     print('in func2 start')
13     time.sleep(3)
14     print('in func2 finished')
15
16 if __name__ == '__main__':
17     p = Process(target=func)
18     p.daemon = True   # 设置子进程为守护进程，必须在start前设置
19     p.start()
20     p2 = Process(target=func2)
21     p2.start()
22     p2.terminate()     # 结束一个子进程，不会立即关闭,所以is_alive立刻查看的结果可能还是存活
23     time.sleep(1)。
24     print(p2.is_alive())  # 检验一个进程是否还活着
25     # time.sleep(1)
26     print(p2.name)
27     # i = 0
28     # while i<5:
29     #     print('我是socket server')
30     #     time.sleep(1)
31     #     i+=1
32
33 # 输出的结果：
34 我还活着
35 我还活着
36 我还活着
37 False
38 Process-2
39
40 # 守护进程会随着主进程的代码执行完毕 而 结束
41 # 在主进程内结束一个子进程 p.terminate()
42     #  结束一个进程不是在执行方法之后立即生效,需要一个操作系统响应的过程
43 # 检验一个进程是否活着的状态 p.is_alive()
44 # p.name p.pid 这个进程的名字和进程号

 1 import os
 2 import time
 3 from multiprocessing import Process
 4
 5 class Myprocess(Process):
 6     def __init__(self,person):
 7         super().__init__()
 8         self.person = person
 9     def run(self):
10         print(os.getpid(),self.name)
11         print('%s正在和女主播聊天' %self.person)
12
13 if __name__ == '__main__':
14     p=Myprocess('哪吒')
15     p.daemon=True
16     #一定要在p.start()前设置,设置p为守护进程,禁止p创建子进程,并且父进程代码执行结束,p即终止运行
17     p.start()
18     time.sleep(1) # 在sleep时查看进程id对应的进程
19     print('主')
20
21 # 输出的结果：
22 92480 Myprocess-1
23 哪吒正在和女主播聊天
24 主

View Code

import json
import time
from multiprocessing import Process,Lock
# from multiprocessing import Lockdef show(i):with open('ticket') as f:dic = json.load(f)print('余票: %s'%dic['ticket'])def buy_ticket(i,lock):lock.acquire() #拿钥匙进门with open('ticket') as f:dic = json.load(f)time.sleep(0.1)if dic['ticket'] > 0 :dic['ticket'] -= 1print('\033[32m%s买到票了\033[0m'%i)else:print('\033[31m%s没买到票\033[0m'%i)time.sleep(0.1)with open('ticket','w') as f:json.dump(dic,f)lock.release()      # 还钥匙if __name__ == '__main__':for i in range(10):p = Process(target=show,args=(i,))p.start()lock = Lock()   # 加锁for i in range(10):p = Process(target=buy_ticket, args=(i,lock))   # 将lock作为一个参数传入
        p.start()# 输出结果：
余票: 2
余票: 2
余票: 2
余票: 2
余票: 2
0买到票了
2买到票了
1没买到票
3没买到票
4没买到票

 1 # ktv小实例
 2 # 一个KTV只能同时容纳4个人，当不满4人时可以进入。当有人唱完歌出来后可以有人再进去
 3
 4 from multiprocessing import Process
 5 import random,time
 6 from multiprocessing import Semaphore
 7
 8 def ktv(i,sem):
 9     sem.acquire()   #每个人进入KTV时获取钥匙
10     print("%s 走进来了"%i)
11     # 此处采用random和time模块模拟唱歌所花费的时间(3-8秒)
12     time.sleep(random.randint(3,8))
13     print('%s唱完歌，走出来了'%i)
14     sem.release()   #归还钥匙
15
16 # 开启20个进程模拟20个人进入KTV
17 if __name__ == "__main__":
18     # 设置信号量，相当于设置了KTV门有4把钥匙
19     sem_number = Semaphore(4)
20     # 利用for循环设置20个进程
21     for i in range(20):
22         p = Process(target = ktv,args=(i,sem_number))
23         # 将实例化的信号量当做一个参数传入
24         p.start()

 1 # 一个信号可以使所有的进程都进入阻塞状态
 2 # 也可以控制所有的进程解除阻塞
 3 # 一个事件被创建之后,默认是阻塞状态
 4 from multiprocessing import Event
 5 e = Event()  # 创建了一个事件
 6 print(e.is_set())   # 查看一个事件的状态,默认被设置成阻塞
 7 e.set()      # 将这个事件的状态改为True
 8 print(e.is_set())
 9 e.wait()     # 是依据e.is_set()的值来决定是否阻塞的
10 print(123456)
11 e.clear()    # 将这个事件的状态改为False
12 print(e.is_set())
13 e.wait()     # 等待 事件的信号被变成True
14 print('*'*10)

 1 # 红绿灯事件练习
 2
 3 from multiprocessing import Process
 4 import random,time
 5 from multiprocessing import Event
 6
 7 def cars(e,car_number):
 8     if not e.is_set():
 9         print('car%i在等待' % car_number)
10         e.wait()  # 阻塞 直到得到一个 事件状态变成 True 的信号
11
12     print('\033[0;32;40mcar%i通过\033[0m' % car_number)
13
14 def light(e):
15     while True:
16         if e.is_set():
17             e.clear()
18             print("\033[31m红灯亮了\033[0m")
19         else:
20             e.set()
21             print("\033[32m绿灯亮了\033[0m")
22
23         time.sleep(2)
24
25 if __name__ == "__main__":
26     e = Event()
27     traffic_light = Process(target=light,args=(e,))
28     traffic_light.start()
29     # 用20个进程模拟将要路过红绿灯的20辆车
30     for i in range(20):
31         car = Process(target=cars,args=(e,i))
32         car.start()
33         time.sleep(random.randint(0,1))

View Code

 1 # 队列：先进先出
 2
 3 from multiprocessing import Queue
 4
 5 q = Queue(5) # 创建一个容量为5的队列
 6 q.put('a')        # 在队列中放值，如果队列满了就发生阻塞（即不往下执行）处于等待状态，直到队列中放入了值才继续执行
 7 print(q.get())  # 从队列中取值，如果队列空了就发生阻塞（即不往下执行）处于等待状态，直到队列中有值被取走才继续执行
 8 print(q.empty())    # 判断队列是否为空
 9 print(q.full())     # 判断队列是否是满的
10
11 print(q.get_nowait())   #从队列中取值，当队列中没有值的时候会报错
12 q.put_nowait()        # 把值放入队列中，当队列满的时候会报错

 1 from multiprocessing import Queue,Process
 2 def produce(q):
 3     q.put('hello')
 4
 5 def consume(q):
 6     print(q.get())
 7
 8 if __name__ == '__main__':
 9     q = Queue()
10     p = Process(target=produce,args=(q,))
11     p.start()
12     c = Process(target=consume, args=(q,))
13     c.start()

 1 # 生产者消费者模型
 2 from multiprocessing import Process
 3 from multiprocessing import Queue
 4 import time,random
 5
 6 def produce(name,what,queue):
 7     for i  in range(5):
 8         time.sleep(random.randint(1,3))
 9         w = "%s生产了%s,%s"%(name,what,i)
10         print(w)
11         queue.put(w)
12
13 def consuer(q,name):
14     while True:
15         food = q.get()
16         if food is None:
17             print("%s消费完了，得到了一个空"%name)
18             break
19         print('\033[31m%s消费了%s\033[0m'%(name,food))
20         time.sleep(random.randint(1,3))
21
22 if __name__ == "__main__":
23     q = Queue(20)
24     p1 = Process(target=produce,args=("alex","包子",q))
25     p1.start()
26     p2 = Process(target=produce, args=("武sir", "烧饼", q))
27     p2.start()
28     c1 = Process(target=consuer,args=(q,"dog"))
29     c1.start()
30     c2 = Process(target=consuer,args=(q,"cat"))
31     c2.start()
32     p1.join()
33     p2.join()
34     q.put(None)
35     q.put(None)
36 #  在消费者这一端:
37     # 每次获取一个数据
38     # 处理一个数据
39     # 发送一个记号 : 标志一个数据被处理成功
40
41 # 在生产者这一端:
42     # 每一次生产一个数据,
43     # 且每一次生产的数据都放在队列中
44     # 在队列中刻上一个记号
45     # 当生产者全部生产完毕之后,
46     # join信号 : 已经停止生产数据了
47                 # 且要等待之前被刻上的记号都被消费完
48                 # 当数据都被处理完时,join阻塞结束
49
50 # consumer 中把所有的任务消耗完
51 # producer 端 的 join感知到,停止阻塞
52 # 所有的producer进程结束
53 # 主进程中的p.join结束
54 # 主进程中代码结束
55 # 守护进程(消费者的进程)结束

View Code

 1 # 管道基础
 2 from multiprocessing import Pipe
 3
 4 conn1,conn2 = Pipe()
 5 conn1.send("吃了么？")  # 不需要转换成bytes类型
 6 print(conn2.recv())
 7
 8 # 输出结果：吃了么？
 9
10
11
12 # 管道结合进程使用
13 from multiprocessing import Pipe,Process
14
15 def func(conn):
16     conn.send('hello, world!')
17
18 if __name__ == '__main__':
19     conn1,conn2 = Pipe()
20     p = Process(target=func,args=(conn1,))
21     p.start()
22     print(conn2.recv())

 1 # 处理多进程发送消息结束之后的阻塞方法一
 2
 3 from multiprocessing import Process,Pipe
 4
 5 def func(conn):
 6     while True:
 7         mes = conn.recv()
 8         if mes is None:
 9             break
10         else:
11             print(mes)
12
13 if __name__ == "__main__":
14     conn1,conn2 = Pipe()
15     p = Process(target=func,args = (conn2,))
16     p.start()
17     for i in range(20):
18         conn1.send('Hello World！')
19     conn1.send(None)

View Code

 1 # 处理多进程发送消息结束之后的阻塞方法二
 2 from multiprocessing import Pipe,Process
 3
 4 def func(conn1,conn2):
 5     conn2.close()
 6     while True:
 7         try :
 8             msg = conn1.recv()
 9             print(msg)
10         except EOFError:
11             conn1.close()
12             break
13
14 if __name__ == '__main__':
15     conn1, conn2 = Pipe()
16     Process(target=func,args = (conn1,conn2)).start()
17     conn1.close()
18     for i in range(20):
19         conn2.send('吃了么')
20     conn2.close()
21
22 # 只有当管道都关闭之后才会引发EOFError异常

View Code

 1 # 管道实现生产者消费者模型,会出现'数据安全问题'
 2 from multiprocessing import Lock,Pipe,Process
 3 def producer(con,pro,name,food):
 4     con.close()
 5     for i in range(100):
 6         f = '%s生产%s%s'%(name,food,i)
 7         print(f)
 8         pro.send(f)
 9     pro.send(None)
10     pro.send(None)
11     pro.send(None)
12     pro.close()
13
14 def consumer(con,pro,name,lock):
15     pro.close()
16     while True:
17             lock.acquire()
18             food = con.recv()
19             lock.release()
20             if food is None:
21                 con.close()
22                 break
23             print('%s吃了%s' % (name, food))
24 if __name__ == '__main__':
25     con,pro = Pipe()
26     lock= Lock()
27     p = Process(target=producer,args=(con,pro,'egon','泔水'))
28     c1 = Process(target=consumer, args=(con, pro, 'alex',lock))
29     c2 = Process(target=consumer, args=(con, pro, 'bossjin',lock))
30     c3 = Process(target=consumer, args=(con, pro, 'wusir',lock))
31     c1.start()
32     c2.start()
33     c3.start()
34     p.start()
35     con.close()
36     pro.close()
37
38 # pipe 数据不安全性
39 # 通过加锁来控制操作管道的行为 来避免进程之间争抢数据造成的数据不安全现象
40 # 队列 进程之间数据安全的
41 # 队列的底层就是 管道 + 锁 实现的

View Code

 1 # 加锁之后不会出现数据安全问题
 2 from multiprocessing import Process,Pipe,Lock
 3 def consumer(produce, consume,name,lock):
 4     produce.close()
 5     while True:
 6         lock.acquire()
 7         baozi=consume.recv()
 8         lock.release()
 9         if baozi:
10             print('%s 收到包子:%s' %(name,baozi))
11         else:
12             consume.close()
13             break
14
15 def producer(produce, consume,n):
16     consume.close()
17     for i in range(n):
18         produce.send(i)
19     produce.send(None)
20     produce.send(None)
21     produce.close()
22
23 if __name__ == '__main__':
24     produce,consume=Pipe()
25     lock = Lock()
26     c1=Process(target=consumer,args=(produce,consume,'c1',lock))
27     c2=Process(target=consumer,args=(produce,consume,'c2',lock))
28     p1=Process(target=producer,args=(produce,consume,30))
29     c1.start()
30     c2.start()
31     p1.start()
32     produce.close()
33     consume.close()

View Code

 1 from multiprocessing import Manager,Process
 2
 3 def main(dic):
 4     dic['count'] -= 1
 5     print(dic)
 6
 7 if __name__ == '__main__':
 8     m = Manager()
 9     dic=m.dict({'count':100})
10     p_lst = []
11     p = Process(target=main, args=(dic,))
12     p.start()
13     p.join()
14
15 # 输出结果：
16 # {'count': 99}

 1 from multiprocessing import Manager,Process,Lock
 2 def main(dic,lock):
 3     dic['count'] -= 1
 4
 5 if __name__ == '__main__':
 6     m = Manager()
 7     l = Lock()
 8     dic=m.dict({'count':100})
 9     p_lst = []
10     for i in range(50):
11         p = Process(target=main,args=(dic,l))
12         p.start()
13         p_lst.append(p)
14     for i in p_lst: i.join()
15     print('主进程',dic)
16
17 # 输出结果：
18 # 主进程 {'count': 50}

 1 import os
 2 import time
 3 from multiprocessing import Pool
 4 def func(n):
 5     print('start func%s'%n,os.getpid())
 6     time.sleep(1)
 7     print('end func%s'%n,os.getpid())
 8
 9 if __name__ == '__main__':
10     p = Pool(5)     # 一般参数为CPU的核数加一
11     for i in range(10):
12         p.apply(func,args=(i,))   # 同步创建进程，效率很低
13         p.apply_async(func,args=(i,))   # 异步创建进程，显著提高效率
14     p.close()  # 结束进程池接收任务
15     p.join()   # 感知进程池中的任务执行结束
16
17 # p.map(funcname,iterable)     默认异步的执行任务,且自带close和join
18 # p.apply   同步调用的
19 # p.apply_async 异步调用 和主进程完全异步 需要手动close 和 join

 1 # 进程池的返回值
 2 from multiprocessing import Pool
 3 def func(i):
 4     return i*i
 5
 6 if __name__ == '__main__':
 7     p = Pool(5)
 8     for i in range(10):
 9         res = p.apply(func,args=(i,))   # apply的结果就是func的返回值
10         print(res)
11
12
13 import time
14 from multiprocessing import Pool
15 def func(i):
16     time.sleep(0.5)
17     return i*i
18
19 if __name__ == '__main__':
20     p = Pool(5)
21     res_l = []
22     for i in range(10):
23         res = p.apply_async(func,args=(i,))
24         # apply的结果就是func的返回值
25         res_l.append(res)
26     for res in res_l:print(res.get())
27     # 等着 func的return结果
28
29 # 返回值 : apply_async返回的对象obj
30         #          为了用户能从中获取func的返回值obj.get()
31         # get会阻塞直到对应的func执行完毕拿到结果

进程池的返回值

 1 # 进程池的回调函数
 2 import os
 3 from multiprocessing import Pool
 4 def func1(n):
 5     print('in func1',os.getpid())
 6     return n*n
 7
 8 def func2(nn):
 9     print('in func2',os.getpid())
10     print(nn)
11
12 if __name__ == '__main__':
13     print('主进程 :',os.getpid())
14     p = Pool(5)
15     for i in range(10):
16         p.apply_async(func1,args=(10,),callback=func2)
17     # callback回调函数，先执行进程1调用func1，然后将func1的返回值传给func2
18     p.close()
19     p.join()
20     

进程池的回调函数

 1 # 进程池回调函数的爬虫实例
 2 #进程池的回调函数适合于爬虫的处理，用回调函数处理获取的网页，能够显著提高爬虫的效率
 3
 4 import requests
 5 from multiprocessing import Pool
 6
 7 def get(url):
 8     response = requests.get(url[0])
 9     if response.status_code == 200:
10         return url,response.content.decode('utf-8')
11
12 def deal_get(args):
13     # url,content = args
14     url = args[0]
15     content = args[1]
16     print(url,len(content))
17
18 if __name__ == "__main__":
19     p = Pool(5)
20     url_list = [
21         'http://www.baidu.com',
22         "http://www.sohu.com/",
23         "https://www.cnblogs.com/",
24         "https://www.sogou.com/",
25     ]
26     for i in url_list:
27         p.apply_async(get,args=(i,),callback=deal_get)
28     p.close()
29     p.join()

进程池回调函数的爬虫实例