zoukankan      html  css  js  c++  java

  • 并发编程——多进程——multiprocessing开启进程的方式及其属性（3）

    • 开启进程的两种方式——Process
      • 方式一：函数方法
      •  1 from multiprocessing import Process
 2 import time
 3 def task(name):
 4     print('%s is running'%name)
 5     time.sleep(3)
 6     print('%s is done' % name)
 7 if __name__ == '__main__':
 8     p = Process(target=task,args=('子进程1',))
 9     p.start() #仅仅只是给操作系统发送了一个信号
10 
11     print('主进程')
        方式一
      • 方式二：类方法
      •  1 from multiprocessing import Process
 2 import time,os
 3 class MyProcess(Process):
 4     def __init__(self,name):
 5         super().__init__()
 6         self.name = name
 7     def run(self):
 8         print('%s is running'%self.name,os.getppid())
 9         time.sleep(3)
10         print('%s is done' % self.name,os.getppid())
11 if __name__ == '__main__':
12     p = MyProcess('子进程1')
13     p.start()
14 
15     print('主进程',os.getpid())
        方式二
    • 进程id
      • 子进程id ： os.getpid()
      • 父进程id ： os.getppid()
    • 僵尸进程与孤儿进程（仅存在于linux和unix系统中）
      • 僵尸进程：在Linux系统中，任何子进程都会经历僵尸进程模式，以便父进程查看子进程状态,pid不会丢失，占用pid资源，在父进程一直不死的情况下有害
      • 孤儿进程：父进程已死，子进程依然活着；孤儿进程会被一个INIT进程收留，是无害的。
    • Process的方法介绍
      • p.start()：启动进程，并调用该子进程中的p.run()
      • p.run()：进程启动时运行的方法，正是它去调用target指定的函数，我们自定义类的类中一定要实现该方法
      • p.terminate():强制终止进程p，不会进行任何清理操作，如果p创建了子进程，该子进程就成了僵尸进程，使用该方法需要特别小心这种情况。如果p还保存了一个锁那么也将不会被释放，进而导致死锁
      • p.is_alive()：如果进程p仍然运行，返回True
      • p.join([timeout])：主进程等待p终止（强调：是主线程处于等的状态，而p是处于运行的状态）。timeout 是可选的超时时间。
      • p.name：查看子进程的进程名，即使子进程终止了，也可以查到
    • join方法 
      •  1 def talk(name):
 2     print("%s 开始说话"%name)
 3     time.sleep(random.randint(1,5))
 4     print("%s 说话结束" %name)
 5 if __name__ == '__main__':
 6     # freeze_support()
 7     p1 = Process(target=talk,args=(1,))
 8     p2 = Process(target=talk,args=(2,))
 9     p3 = Process(target=talk,args=(3,))
10     p = [p1,p2,p3]
11     for i in p:
12         i.start()
13     # p1.terminate()
14     for i in p:
15         i.join()
16 
17     print("主进程结束")
        join方法
      • join是主进程在等子进程结束，当有多个子进程时，子进程之间依然是并发的
    • 守护进程
      • 守护进程会在主进程代码执行结束后就终止，字随父死
      • 守护进程内无法再开启子进程，否则抛出异常
      •  1 #-*- coding:utf-8 -*-
 2 import time
 3 from multiprocessing import Process
 4 def task(name):
 5     print("子进程%s开始执行"%name)
 6     time.sleep(2)
 7     # p = Process(target=time.time,args=(3,))
 8     # p.start()
 9     print("子进程%s执行完毕"%name)
10 if __name__ == '__main__':
11     p = Process(target=task,args = (1,))
12     p.daemon = True
13     p.start()
14     # p.join()
15     print("主")
        守护进程
    • 互斥锁
      • 把并发变成串行，牺牲效率，保证数据的正确性
      • 前提：共享同一套文件系统（存储于硬盘中）
      • 存在的问题：效率低，需要自己加锁处理
      •  1 #-*- coding:utf-8 -*-
 2 from multiprocessing import Process,Lock
 3 import os,time
 4 def work(lock):
 5     lock.acquire()#加锁
 6     print('%s is running' %os.getpid())
 7     time.sleep(2)
 8     print('%s is done' %os.getpid())
 9     lock.release()#释放锁
10 
11 if __name__ == '__main__':
12     lock = Lock()
13     for i in range(3):
14         p=Process(target=work,args=(lock,))
15         p.start()
        互斥锁
      • 与join的区别
        • join是将一个任务整体串行
        • 互斥锁：可以将一个任务中的某一段代码串行
      •  1 # -*- coding:utf-8 -*-
 2 import json
 3 import time
 4 from multiprocessing import Process,Lock
 5 
 6 
 7 def search(name):
 8     time.sleep(1)
 9     dic = json.load(open('db.txt','r',encoding='utf-8'))
10     print('<%s>查看到剩余票数[%s]' % (name, dic['count']))
11 
12 
13 def get(name,lock):
14     lock.acquire()
15     time.sleep(1)
16     dic = json.load(open('db.txt', 'r', encoding='utf-8'))
17     if dic['count'] > 0:
18         dic['count'] -= 1
19         time.sleep(3)
20         json.dump(dic, open('db.txt', 'w', encoding='utf-8'))
21         print('<%s>购票成功' % name)
22     lock.release()
23 
24 
25 def task(name,lock):
26     search(name)
27     get(name,lock)
28 
29 if __name__ == '__main__':
30     lock = Lock()
31 
32 
33     for i in range(1,10):
34         p = Process(target=task, args=(i,lock))
35         p.start()
36         # p.join()
        模拟抢票
    • 队列
      • 队列和管道都是将数据存放于内存中，而队列又是基于（管道+锁）实现的，可以让我们从复杂的锁问题中解脱出来，因而队列才是进程间通信（IPC）的最佳选择
      • 创建队列的类：Queue([maxsize])
        • [maxsize]是队列中允许最大项数，省略则无大小限制
        • 队列内存放的是消息而非大数据
        • 队列占用的是内存空间，因而maxsize即便是无大小限制也受限于内存大小
      •  1 #-*- coding:utf-8 -*-
 2 from multiprocessing import Process,Queue
 3 q = Queue(3)
 4 q.put(1)
 5 q.put(2)
 6 q.put(3)
 7 print(q.full())
 8 # q.put(4)#再放就阻塞了
 9 
10 print(q.get())
11 print(q.get())
12 print(q.get())
13 print(q.empty())#空了
14 # print(q.get())#再取就阻塞了
        队列
    • 生产者消费者模型
      • 什么是生产者消费者模型
        • 生产者消费者模型是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列中取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。
      • 程序中有两类角色
        • 一类负责生产数据（生产者）
        • 一类负责处理数据（消费者）
      • 引入生产者消费者模型是为了解决的问题是
        • 平衡生产者与消费者之间的速度差
        • 程序解开耦合
      • 如何实现生产者消费者模型
        • 生产者<--->队列<--->消费者
      •  1 #-*- coding:utf-8 -*-
 2 from multiprocessing import Process,Queue
 3 import time
 4 import random
 5 import os
 6 def consumer(q,name):
 7     while True:
 8         res = q.get()
 9         if res is None:break
10         time.sleep(random.randint(1,3))
11         print('33[43m%s 吃 %s33[0m' %(name,res))
12 def producer(q,name,food):
13     for i in range(3):
14         time.sleep(random.randint(1,3))
15         res = '%s%s'%(food,i)
16         q.put(res)
17         print('33[45m%s 生产了 %s33[0m' % (name, res))
18 if __name__ == '__main__':
19     q = Queue()
20     #生产者们，即厨师
21     p1 = Process(target=producer,args=(q,'egon1','包子'))
22     p2 = Process(target=producer,args=(q,'egon2','骨头'))
23     p3 = Process(target=producer,args=(q,'egon3','饺子'))
24 
25     #消费者们：即吃货们
26     c1 = Process(target=consumer,args=(q,'alex1'))
27     c2 = Process(target=consumer,args=(q,'alex2'))
28 
29     #开始
30     p1.start()
31     p2.start()
32     p3.start()
33     c1.start()
34     c2.start()
35     p1.join()
36     p2.join()
37     p3.join()
38     q.put(None)
39     q.put(None)
40     q.put(None)
41     print('主程序结束')
        生产者消费者模型
    • JoinableQueue([maxsize])
      • JoinableQueue的实例p除了与Queue对象相同的方法之外还具有：
      • q.task_done()：使用者使用此方法发出信号，表示q.get()的返回项目已经被处理。如果调用此方法的次数大于从队列中删除项目的数量，将引发ValueError异常
      • q.join()：生产者调用此方法进行阻塞，知道队列中所有的项目均被处理。阻塞将持续到队列中的每个项目均调用q.task_done()方法为止。
      •  1 #-*- coding:utf-8 -*-
 2 from multiprocessing import Process,JoinableQueue
 3 import time,random,os
 4 def consumer(q,name):
 5     while True:
 6         res = q.get()
 7         time.sleep(random.randint(1,3))
 8         print('33[43m%s 吃 %s33[0m' %(name,res))
 9         q.task_done()#发送信号给q.join()，说明已经从队列中取走一个数据并处理完毕了
10 def producer(q,name,food):
11     for i in range(3):
12         time.sleep(random.randint(1,3))
13         res = '%s%s'%(food,i)
14         q.put(res)
15         print('33[45m%s 生产了 %s33[0m' % (name, res))
16     q.join()#等到消费者把自己放入队列中的所有的数据都取走之后，生产者才结束
17 
18 if __name__ == "__main__":
19     q = JoinableQueue()#使用JoinableQueue()
20     #生产者们：即厨师们
21     p1 = Process(target=producer, args=(q, 'egon1', '包子'))
22     p2 = Process(target=producer, args=(q, 'egon2', '骨头'))
23     p3 = Process(target=producer, args=(q, 'egon3', '泔水'))
24     # 消费者们:即吃货们
25     c1 = Process(target=consumer, args=(q, 'alex1'))
26     c2 = Process(target=consumer, args=(q, 'alex2'))
27     c1.daemon = True
28     c2.daemon = True
29     # 开始
30     p1.start()
31     p2.start()
32     p3.start()
33     c1.start()
34     c2.start()
35     p1.join()
36     p2.join()
37     p3.join()
38     #1、主进程等生产者p1、p2、p3结束
39     #2、而p1、p2、p3是在消费者把所有数据都取干净之后才会结束
40     #3、所以一旦p1、p2、p3结束了，证明消费者也没必要存在了，应该随着主进程一块死掉，因而需要将生产者们设置成守护进程
41     print('主')
        使用JoinableQueue实现消费者生产者模型
  • 相关阅读:
    php 制表符(\t) 与单引号的疑惑。
    preg_replace 正则替换的疑惑
    命令行下使用curl,采集数据遇到的问题。
    vmware player 里的window xp 安装wamp遇到的问题
    2014年11月05日
    2014年11月05日
    让开发人员搞业务是不对的
    2014年11月05日
    web应用.表格很重要
    业务复杂
  • 原文地址：https://www.cnblogs.com/GraceZ/p/8452069.html
Copyright © 2011-2022 走看看