pythonNetday06

进程

Process（target,name,args,kwargs）

p.pid : 创建的新的进程的PID号
p.is_alive() 判断进程是否处于alive状态

p.daemon = True 默认为False 如果设置为True 则一般不需要加join，并且主进程退出时子进程也会结束

* daemon 属性设置要在start() 前

注意：
如果多个子进程拷贝同一个父进程中的对象，则多个子进程使用的是同一个对象
（如文件对象，套接字，队列，管道。。。）
如果是在创建子进程后单独创建的对象，则多个子进程各不相同

 from multiprocessing import Process
 from time import sleep
 import os 

 def th1():
     sleep(3)
     print("吃饭")
     print(os.getpid(),'---',os.getppid())   # 当前进程的PID号，父进程的PID号
 def th2():
     sleep(4)
     print("睡觉")
     print(os.getpid(),'---',os.getppid())
 def th3():
     sleep(2)
     print("打豆豆")
     print(os.getpid(),'---',os.getppid())

 things = [th1,th2,th3]
 process = []

 for th in things:
     p = Process(target = th)
     process.append(p)  #保存进程对象
     p.start()

 #回收进程
 for i in process:
     i.join()

 # 打豆豆
 # 5625 --- 5622
 # 吃饭
 # 5623 --- 5622
 # 睡觉
 # 5624 --- 5622

 from multiprocessing import Process
 from time import sleep

 def worker(sec,name):
     for i in range(3):
         sleep(sec)
         print("I'm %s"%name)
         print("I'm working...")

 p = Process(target = worker,name = "Worker",args = (2,),kwargs = {'name':'Alex'})
 p.start()

 print(p.name)   # Worker
 print("Child PID:",p.pid) # 创建新进程的PID号
 print("is alive?  ",p.is_alive()) # 判断进程状态

 p.join()
 print("is alive?  ",p.is_alive())   # False

 # Worker
 # Child PID: 5994
 # is alive?   True
 # I'm Alex
 # I'm working...
 # I'm Alex
 # I'm working...
 # I'm Alex
 # I'm working...
 # is alive?   False

 from multiprocessing import Process
 from time import sleep,ctime 

 def tm1():
     while True:
         sleep(2)
         print(ctime())

 p = Process(target = tm1)
 p.daemon = True
 p.start()
 sleep(3)
 print("main process over")

 # Wed Aug 15 11:08:28 2018
 # main process over

p.daemon = True

创建自定义进程类

1. 编写类继承Process
2. 在自定义类中加载父类__init__以获取父类属性，同时可以自定义新的属性
3. 重写run方法，在调用start时自动执行该方法

进程的缺点：进程在创建和销毁的过程中消耗的资源相对较多

 from multiprocessing import Process
 import time 

 class ClockProcess(Process):
     def __init__(self,value):
         #调用父类init
         super().__init__()
         self.value = value
     #重写run方法
     def run(self):
         for i in range(5):
             time.sleep(self.value)
             print("The time is {}".format(time.ctime()))

 p = ClockProcess(2)
 #自动执行run
 p.start()

 p.join()

 # The time is Wed Aug 15 12:32:54 2018
 # The time is Wed Aug 15 12:32:56 2018
 # The time is Wed Aug 15 12:32:58 2018
 # The time is Wed Aug 15 12:33:00 2018
 # The time is Wed Aug 15 12:33:02 2018

自定义进程类

进程池技术

产生原因：如果有大量的任务需要多进程完成，而任务周期又比较短且需要频繁创建。此时可能产生大量进程频繁创建销毁的情况，消耗计算机资源较大

使用方法 ：
1. 创建进程池，在池内放入适当数量的进程
2. 将事件封装函数，放入到进程池
3. 事件不断运行，知道所有放入进程池事件运行完成
4. 关闭进程池，回收进程

from multiprocessing import Pool

Pool(processes)　　创建进程池对象
参数：进程数量
返回：进程池对象

pool.apply_async(fun,args,kwds)　　将事件放入进程池执行
参数： fun 要执行的事件函数
args 以元组为fun传参
kwds 以字典为fun传参
返回值：返回一个事件对象通过get()属性函数可以获取fun的返回值

pool.apply(fun,args,kwds)　　将事件放入进程池执行

参数： fun 要执行的事件函数
args 以元组为fun传参
kwds 以字典为fun传参

pool.close()　　关闭进程池，无法再加入事件

pool.join()　　回收进程池

 from multiprocessing import Pool
 from time import sleep,ctime 

 def worker(msg):
     sleep(2)
     print(msg)
     return ctime()

 pool = Pool(processes = 4)  #创建进程池对象

 result = []
 for i in range(10):
     msg = "hello %d"%i
     r = pool.apply_async(func = worker,args = (msg,))   #将事件放入进程池
     result.append(r)
     # pool.apply(func = worker,args = (msg,))   #同步执行

 pool.close()    #关闭进程池
 pool.join() #回收
 for i in result:    #获取事件函数返回值
     print(i.get())

 # hello 3
 # hello 1
 # hello 2
 # hello 0
 # hello 4
 # hello 7
 # hello 5
 # hello 6
 # hello 9
 # hello 8
 # Wed Aug 15 11:52:01 2018
 # Wed Aug 15 11:52:01 2018
 # Wed Aug 15 11:52:01 2018
 # Wed Aug 15 11:52:01 2018
 # Wed Aug 15 11:52:03 2018
 # Wed Aug 15 11:52:03 2018
 # Wed Aug 15 11:52:03 2018
 # Wed Aug 15 11:52:03 2018
 # Wed Aug 15 11:52:05 2018
 # Wed Aug 15 11:52:05 2018

pool(process)

pool.map(func,iter)　　将要执行的事件放入到进程池
参数： func 要执行的函数
iter 迭代对象，给func传参
返回值：返回 func的返回值列

 from multiprocessing import Pool
 import time 

 def fun(n):
     time.sleep(1)
     print("执行 pool map事件",n)
     return n ** 2 

 pool = Pool(4)
 r = pool.map(fun,range(6))  # 在进程池放入6个事件
 print("返回值列表:",r)

 pool.close()
 pool.join()

 # 执行 pool map事件 0
 # 执行 pool map事件 3
 # 执行 pool map事件 1
 # 执行 pool map事件 2
 # 执行 pool map事件 4
 # 执行 pool map事件 5
 # 返回值列表: [0, 1, 4, 9, 16, 25]

pool.map(func,iter)

进程间通信（IPC）

由于进程间空间独立，资源无法共享，此时在进程间通信就需要专门的通信方法。

进程间通信方法：管道消息队列共享内存信号
信号量套接字

管道通信

　　管道：在内存中开辟一段空间，形成管道结构，多进程使用同一个管道，进程可以对管道进行读写操作

multiprocess ---> Pipe

fd1,fd2 = Pipe(duplex = True)

功能：创建一个管道

参数：默认为双向管道，如果设置为False,则为单向管道

返回值：如果双向管道，fd1,fd2 都可以进行读写操作;如果是单向管道，则fd1只可读，fd2只可写。

fd.recv()

功能;从管道读取内容2

返回值：读到的内容

　　# 当管道为空则阻塞

fd.send(data)

功能：向管道写入内容

参数：要发送的内容

　　# 管道满是会阻塞，几乎可以发送所有python支持的数据

 from multiprocessing import Process,Pipe
 import os,time 

 fd1,fd2 = Pipe(False)   #创建管道

 def fun(name):
     time.sleep(3)

     fd2.send({'a':1,'b':2}) #向管道写入内容

 jobs = []
 for i in range(5):
     p = Process(target = fun,args = (i,))
     jobs.append(p)
     p.start()
 for i in range(5):

     data = fd1.recv()   #读取管道
     print(data)

 for i in jobs:
     i.join()

 # {'b': 2, 'a': 1}
 # {'b': 2, 'a': 1}
 # {'b': 2, 'a': 1}
 # {'b': 2, 'a': 1}
 # {'b': 2, 'a': 1}

fd1,fd2 = Pipe(False)

消息队列

队列：先进先出

从内存中开辟队列结构空间，多个进程可以向队列投放消息，在取出来的时候按照存入顺序取出

创建队列

q = Queue(maxsize = 0)　　创建队列对象
参数： maxsize ：默认表示系统自动分配队列空间
如果传入正整数则表示最多存放多少条消息
返回值：队列对象

q.put(data,[block,timeout])　　向队列中存入消息

参数：data 存放消息（python数据类型）

　　block　　默认为True表示当前队列满的时候阻塞，设置为False则表示非阻塞

　　timeout　　当block为True表示超时时间

返回值：返回获取的消息

q.full()　　判断队列是否为满

q.empty()　　判断队列是否为空

q.qsize()　　判断当前队列有多少消息

q.close()　　关闭队列

 from multiprocessing import Queue
 from time import sleep 

 q = Queue(3)    # 创建队列

 q.put(1)
 sleep(0.1)
 print(q.empty())
 q.put("Process Queue")
 q.put([1,2,3])
 print(q.full())
 #如设置为非阻塞则产生Full异常
 # q.put(666,False) #非阻塞
 # q.put(666,True,3)  #超时
 print(q.get())
 print(q.qsize()) #查看消息数量
 q.close()

 # False
 # True

q = Queue(3)

 from multiprocessing import Process,Queue
 import time 

 #创建队列
 q = Queue()

 def fun1():
     time.sleep(1)
     q.put({'})

 def fun2():
     print("收到消息：",q.get())

 p1 = Process(target = fun1)
 p2 = Process(target = fun2)
 p1.start()
 p2.start()
 p1.join()
 p2.join()

 # 收到消息： {'name': 'Abby', 'passwd': '123'}

queue

共享内存

在内存中开辟一段空间，存储数据，对多个进程可见，每次写入共享内存中的数据会覆盖之前的内容

from multiprocessing import Value

obj = Value(ctype,obj)

功能：开辟共享内存空间

参数：ctype　　字符串　　要转变的c的数据类型，对比类型对照表

　　obj　　共享内存的初始化数据

返回：共享内存对象

obj = Array(ctype,obj)

功能：开辟共享内存

参数：ctype　　要转化的c的类型

　　obj　　要存入共享的数据

　　　　如果是列表　　将列表存入共享内存，要求数据类型一致

　　　　如果是正整数　　表示开辟几个数据空间

 from  multiprocessing import Process,Value
 import time
 import random  

 #创建共享内存
 money = Value('i',6000)

 #存钱
 def deposite():
     for i in range(100):
         time.sleep(0.05)
         #对value的修改就是对共享内存的修改
         money.value += random.randint(1,200)
 #花销
 def withdraw():
     for i in range(100):
         time.sleep(0.04)
         #对value的修改就是对共享内存的修改
         money.value -= random.randint(1,200)

 d = Process(target = deposite)
 w = Process(target = withdraw)

 d.start()
 w.start()

 d.join()
 w.join()
 print(money.value)

obj = Value(ctype,obj)

　　　　　　管道　　　　　　　　消息队列　　　　共享内存

开辟空间　　内存　　　　　　　　内存　　　　　　内存

读写方式　两端读写(双向/单向)　　先进先出　　　　操作覆盖

效率　　一般　　　　　　　　　一般　　　　　　较快

应用　　多用于父子进程　　　　使用广泛　　　复杂、需要同步互斥操作　　　　通信

cookie
获取文件大小
size = os.path.getsize("./timg.jpeg")

巴特西

pythonNetday06

进程

创建自定义进程类

管道通信

消息队列

共享内存

最新文章

热门文章