标签：__ get res 编程 print 并发 进程 import

process——进程

一.multiprocessing模块介绍

python中的多线程无法利用多核优势，如果想要充分地使用多核CPU的资源（os.cpu_count()查看），在python中大部分情况需要使用多进程。Python提供了
multiprocessing。 multiprocessing模块用来开启子进程，并在子进程中执行我们定制的任务（比如函数），该模块与多线程模块threading的编程接口类似。

multiprocessing模块的功能众多：支持子进程、通信和共享数据、执行不同形式的同步，提供了Process、Queue、Pipe、Lock等组件。

需要再次强调的一点是：与线程不同，进程没有任何共享状态，进程修改的数据，改动仅限于该进程内。

二.开启进程的两种方式

注意：在windows中Process()必须放到# if __name__ == '__main__':下

由于Windows没有fork，多处理模块启动一个新的Python进程并导入调用模块。

如果在导入时调用Process（），那么这将启动无限继承的新进程（或直到机器耗尽资源）。

这是隐藏对Process（）内部调用的原理，使用if __name__ == '__main __'，这个if语句中的语句将不会在导入时被调用。

方式一：multiprocessing模块

• from multiprocessing import Process
  import time
  
  def task(name):
      print('%s is running' % name)
      time.sleep(2)
  
  if __name__ == '__main__':
      # 1 创建一个进程对象
      p = Process(target=task, args=('dzg',))
      # 2 开启进程
      p.start() # 告诉操作系统帮你创建一个进程  异步
      print('main...')
  
  # 运行结果
  # main...
  # dzg is running
  # 2秒后 程序结束
  

方式二：类的继承

• from multiprocessing import Process
  import time
  
  class MyProcess(Process):
  
      def run(self):
          print('running...')
          time.sleep(2)
          print('over')
  
  if __name__ == '__main__':
      p = MyProcess()
      p.start()
      print('main')
  
  # 运行结果
  # main
  # running...
  # 2秒后 over  
  # 程序结束
  

  创建进程就是在内存中申请一块内存空间将需要运行的代码丢进去

  一个进程对应在内存中就是一块独立的内存空间

  多个进程对应在内存中就是多块独立的内存空间

  进程与进程之间数据默认情况下是无法直接交互,如果想交互可以借助于第三方工具、模块

三.join方法

join是让主进程等待子进程代码运行结束之后，再继续运行。不影响其他子进程的执行

• from multiprocessing import Process
  import time
  
  def task(i):
      print('%s is running...' % i)
      time.sleep(1)
      print('%s is over' % i)
  
  if __name__ == '__main__':
      p1 = Process(target=task, args=(1,))
      p1.start()
      p1.join()
  
      print('main')
  
  # 运行结果
  # 1 is running...
  # 1 is over
  # main  
  # 程序结束
  

四.进程对象及其他方法

一台计算机上面运行着很多进程，那么计算机是如何区分并管理这些进程服务端的呢？

计算机会给每一个运行的进程分配一个PID号

如何查看

windows电脑

进入cmd输入tasklist即可查看

tasklist |findstr PID查看具体的进程

mac电脑

进入终端之后输入ps aux

ps aux|grep PID查看具体的进程

4.1查看进程pid: current_process

• from multiprocessing import Process, current_process
  import time
  
  def task():
      print('子pid is %s' % current_process().pid)
  
  if __name__ == '__main__':
      p1 = Process(target=task)
      p1.start()
      p1.join()
      print('主pid is', current_process().pid)
  
  # 运行结果
  # 子pid is 6204
  # 主pid is 17592 
  # 程序结束
  

4.2查看进程pid: os模块

• from multiprocessing import Process, current_process
  import os
  
  def task():
      print('子pid is', os.getpid())
      print('父pid is',os.getppid())
  
  if __name__ == '__main__':
      p1 = Process(target=task)
      p1.start()
      p1.join()
      print('主pid is', os.getpid())
  
  # 运行结果
  # 子pid is 6528
  # 父pid is 4128
  # 主pid is 4128 
  # 程序结束
  

4.3杀死当前进程与判断当前进程是否存活

p.terminate() # 杀死当前进程

p.is_alive() # 判断当前进程是否存活

• from multiprocessing import Process, current_process
  import time
  
  def task():
      print('run')
  
  if __name__ == '__main__':
      p1 = Process(target=task)
      p1.start()
      p1.terminate()
      time.sleep(0.1)
      print(p1.is_alive())
  
  # 运行结果
  # False
  # 程序结束
  

五.僵尸进程与孤儿进程

僵尸进程：

Process.join（）

死了但是没有死透

当你开设了子进程之后 该进程死后不会立刻释放占用的进程号

因为我要让父进程能够查看到它开设的子进程的一些基本信息 占用的pid号 运行时 间。。。

所有的进程都会步入僵尸进程

父进程不死并且在无限制的创建子进程并且子进程也不结束

回收子进程占用的pid号

父进程调用join方法

父进程等待子进程运行结束在继续往下执行

孤儿进程：

Process.daemon = True （默认为False）

子进程存活，父进程意外死亡

操作系统有一个专门管理孤儿进程回收相关资源

六.守护进程

主进程创建守护进程

其一：守护进程会在主进程代码执行结束后就终止

其二：守护进程内无法再开启子进程,否则抛出异常：AssertionError: daemonic processes are not allowed to have children

注意：进程之间是互相独立的，主进程代码运行结束，守护进程随即终止

将进程p设置成守护进程 这一句一定要放在start方法上面才有效否则会直接报错

• from multiprocessing import Process, current_process
  import time
  
  def task():
      print('run...')
      time.sleep(1)
      print('over...')
  
  if __name__ == '__main__':
      p1 = Process(target=task)
      p1.daemon = True #将进程p设置成守护进程  这一句一定要放在start方法上面才有效否则会直接报错
      p1.start()
      time.sleep(0.5)
      print('主结束') 
  
  # 运行结果
  # run...
  # 主结束
  # 程序结束
  

七.互斥锁

进程之间数据不共享,但是共享同一套文件系统,所以访问同一个文件,或同一个打印终端,是没有问题的,

而共享带来的是竞争，竞争带来的结果就是错乱，如何控制，就是加锁处理

from multiprocessing import Lock

• from multiprocessing import Process, Lock
  import time
  import json
  import random
  
  def check_ticket(i):
      with open('ticket.json', 'r', encoding='utf-8') as f:
          ticket_dic = json.load(f)
      print('%s查询余票为%s' % (i, ticket_dic.get('num')))
  
  def buy(i):
      with open('ticket.json', 'r', encoding='utf-8') as f:
          ticket_dic = json.load(f)
  
      time.sleep(random.randint(1, 3))
      if ticket_dic.get('num') > 0:
          ticket_dic['num'] -= 1
          with open('ticket.json', 'w', encoding='utf-8') as f:
              json.dump(ticket_dic, f)
          print('%s 购买成功' % i)
      else:
          print('%s 购买失败' % i)
  
  def f1(i, mutex):
      check_ticket(i)
      mutex.acquire()
      buy(i)
      mutex.release()
  
  if __name__ == '__main__':
      mutex = Lock()
      for i in range(1, 4):
          p = Process(target=f1, args=(i, mutex))
          p.start()
  
  # 运行结果
  # 1查询余票为1
  # 2查询余票为1
  # 3查询余票为1
  # 1 购买成功
  # 2 购买失败
  # 3 购买失败
  # 程序结束
  

  

  加锁：购票行为由并发变成了串行，牺牲了运行效率，但保证了数据安全

  总结：

  #加锁可以保证多个进程修改同一块数据时，同一时间只能有一个任务可以进行修改，即串行的修改，没错，速度是慢了，但牺牲了速度却保证了数据安全。
  
  虽然可以用文件共享数据实现进程间通信，但问题是：
  
  1.效率低（共享数据基于文件，而文件是硬盘上的数据）
  
  2.需要自己加锁处理
  
  # 因此我们最好找寻一种解决方案能够兼顾：
  
  1、效率高（多个进程共享一块内存的数据）
  
  2、帮我们处理好锁问题。这就是mutiprocessing模块为我们提供的基于消息的IPC通信机制：队列和管道。
  
  1.队列和管道都是将数据存放于内存中
  
  2.队列又是基于（管道+锁）实现的，可以让我们从复杂的锁问题中解脱出来，
  
  我们应该尽量避免使用共享数据，尽可能使用消息传递和队列，避免处理复杂的同步和锁问题，而且在进程数目增多时，往往可以获得更好的可获展性。
  

注意：

  1.锁不要轻易的使用，容易造成死锁现象(我们写代码一般不会用到，都是内部封装好的)

  2.锁只在处理数据的部分加来保证数据安全(只在争抢数据的环节加锁处理即可)

八.进程间通信-队列

进程彼此之间互相隔离，要实现进程间通信（IPC），multiprocessing模块支持两种形式：队列和管道，这两种方式都是使用消息传递的

创建队列的类（底层就是以管道和锁定的方式实现）：

1 Queue([maxsize]):创建共享的进程队列，Queue是多进程安全的队列，可以使用Queue实现多进程之间的数据传递。

参数介绍：

1 maxsize是队列中允许最大项数，省略则无大小限制。

方法介绍：

主要方法：

1 q.put方法用以插入数据到队列中，put方法还有两个可选参数：blocked和timeout。如果blocked为True（默认值），并且timeout为正值，该方法会阻塞timeout指定的时间，直到该队列有剩余的空间。如果超时，会抛出Queue.Full异常。如果blocked为False，但该Queue已满，会立即抛出Queue.Full异常。

2 q.get方法可以从队列读取并且删除一个元素。同样，get方法有两个可选参数：blocked和timeout。如果blocked为True（默认值），并且timeout为正值，那么在等待时间内没有取到任何元素，会抛出Queue.Empty异常。如果blocked为False，有两种情况存在，如果Queue有一个值可用，则立即返回该值，否则，如果队列为空，则立即抛出Queue.Empty异常.  

3 q.get_nowait():同q.get(False)

4 q.put_nowait():同q.put(False) 

5 q.empty():调用此方法时q为空则返回True，该结果不可靠，比如在返回True的过程中，如果队列中又加入了项目。

6 q.full()：调用此方法时q已满则返回True，该结果不可靠，比如在返回True的过程中，如果队列中的项目被取走。

7 q.qsize():返回队列中目前项目的正确数量，结果也不可靠，理由同q.empty()和q.full()一样

其他方法(了解)：

1 q.cancel_join_thread():不会在进程退出时自动连接后台线程。可以防止join_thread()方法阻塞

2 q.close():关闭队列，防止队列中加入更多数据。调用此方法，后台线程将继续写入那些已经入队列但尚未写入的数据，但将在此方法完成时马上关闭。如果q被垃圾收集，将调用此方法。关闭队列不会在队列使用者中产生任何类型的数据结束信号或异常。例如，如果某个使用者正在被阻塞在get()操作上，关闭生产者中的队列不会导致get()方法返回错误。

3 q.join_thread()：连接队列的后台线程。此方法用于在调用q.close()方法之后，等待所有队列项被消耗。默认情况下，此方法由不是q的原始创建者的所有进程调用。调用q.cancel_join_thread方法可以禁止这种行为

应用：

• '''
  multiprocessing模块支持进程间通信的两种主要形式:管道和队列
  都是基于消息传递实现的,但是队列接口
  '''
  from multiprocessing import Process,Queue
  import time
  q=Queue(3)
  
  #put ,get ,put_nowait,get_nowait,full,empty
  q.put(3)
  q.put(3)
  q.put(3)
  print(q.full()) # 满了
  
  print(q.get())
  print(q.get())
  print(q.get())
  print(q.empty()) # 空了
  

生产者消费者模型

> 在并发编程中使用生产者和消费者模式能够解决绝大多数并发问题。该模式通过平衡生产线程和消费线程的工作能力来提高程序的整体处理数据的速度。

为什么要使用生产者和消费者模式

> 在线程世界里，生产者就是生产数据的线程，消费者就是消费数据的线程。在多线程开发当中，如果生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。为了解决这个问题于是引入了生产者和消费者模式。

什么是生产者消费者模式

> 生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。

基于队列实现生产者消费者模型

• from multiprocessing import Process,Queue
  import time,random,os
  def 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出死循环

生产者在生产完毕后发送结束信号None

• from multiprocessing import Process,Queue
  import time,random,os
  def 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，不一定要由生产者发，主进程里同样可以发，但主进程需要等生产者结束后才应该发送该信号
  

主进程在生产者生产完毕后发送结束信号None

• from multiprocessing import Process,Queue
  import time,random,os
  def 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('主')
  

但上述解决方式，在有多个生产者和多个消费者时，我们则需要用一个很low的方式去解决

有几个消费者就需要发送几次结束信号：相当low

• from multiprocessing import Process,Queue
  import time,random,os
  def 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('主')
  

九.共享数据

展望未来，基于消息传递的并发编程是大势所趋

即便是使用线程，推荐做法也是将程序设计为大量独立的线程集合

通过消息队列交换数据。这样极大地减少了对使用锁定和其他同步手段的需求，

还可以扩展到分布式系统中

进程间通信应该尽量避免使用本节所讲的共享数据的方式

> 进程间数据是独立的，可以借助于队列或管道实现通信，二者都是基于消息传递的

> 虽然进程间数据独立，但可以通过Manager实现数据共享，事实上Manager的功能远不止于此

> A manager object returned by Manager() controls a server process which holds Python objects and allows other processes to manipulate them using proxies.

> A manager returned by Manager() will support types list, dict, Namespace, Lock, RLock, Semaphore, BoundedSemaphore, Condition, Event, Barrier, Queue, Value and Array. For example,

进程之间操作共享的数据

• from multiprocessing import Manager,Process,Lock
  import os
  def work(d,lock):
      # with lock: # 不加锁而操作共享的数据,肯定会出现数据错乱
          d['count']-=1
  
  if __name__ == '__main__':
      lock=Lock()
      with Manager() as m:
          dic=m.dict({'count':100})
          p_l=[]
          for i in range(100):
              p=Process(target=work,args=(dic,lock))
              p_l.append(p)
              p.start()
          for p in p_l:
              p.join()
          print(dic)
          #{'count': 94}
  

十.进程池

在利用Python进行系统管理的时候，特别是同时操作多个文件目录，或者远程控制多台主机，并行操作可以节约大量的时间。多进程是实现并发的手段之一，需要注意的问题是：

> 1.很明显需要并发执行的任务通常要远大于核数
> 2.一个操作系统不可能无限开启进程，通常有几个核就开几个进程
> 3.进程开启过多，效率反而会下降（开启进程是需要占用系统资源的，而且开启多余核数目的进程也无法做到并行）

例如当被操作对象数目不大时，可以直接利用multiprocessing中的Process动态成生多个进程，十几个还好，但如果是上百个，上千个。。。手动的去限制进程数量却又太过繁琐，此时可以发挥进程池的功效。

我们就可以通过维护一个进程池来控制进程数目，比如httpd的进程模式，规定最小进程数和最大进程数... ps：对于远程过程调用的高级应用程序而言，应该使用进程池，Pool可以提供指定数量的进程，供用户调用，当有新的请求提交到pool中时，如果池还没有满，那么就会创建一个新的进程用来执行该请求；但如果池中的进程数已经达到规定最大值，那么该请求就会等待，直到池中有进程结束，就重用进程池中的进程。

创建进程池的类：如果指定numprocess为3，则进程池会从无到有创建三个进程，然后自始至终使用这三个进程去执行所有任务，不会开启其他进程

1 Pool([numprocess  [,initializer [, initargs]]]):创建进程池

参数介绍：

1 numprocess:要创建的进程数，如果省略，将默认使用cpu_count()的值
2 initializer：是每个工作进程启动时要执行的可调用对象，默认为None
3 initargs：是要传给initializer的参数组

方法介绍：

主要方法：

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 p.close():关闭进程池，防止进一步操作。如果所有操作持续挂起，它们将在工作进程终止前完成
4 P.jion():等待所有工作进程退出。此方法只能在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被垃圾回收，将自动调用此函数

应用：

同步调用apply

• from multiprocessing import Pool
  import os,time
  def work(n):
      print('%s run' %os.getpid())
      time.sleep(3)
      return n**2
  
  if __name__ == '__main__':
      p=Pool(3) # 进程池中从无到有创建三个进程,以后一直是这三个进程在执行任务
      res_l=[]
      for i in range(10):
          res=p.apply(work,args=(i,)) # 同步调用，直到本次任务执行完毕拿到res，等待任务work执行的过程中可能有阻塞也可能没有阻塞，但不管该任务是否存在阻塞，同步调用都会在原地等着，只是等的过程中若是任务发生了阻塞就会被夺走cpu的执行权限
          res_l.append(res)
  
      print(res_l)
  

异步调用apply_async

• from multiprocessing import Pool
  import os,time
  def work(n):
      print('%s run' %os.getpid())
      time.sleep(3)
      return n**2
  
  if __name__ == '__main__':
      p=Pool(3) # 进程池中从无到有创建三个进程,以后一直是这三个进程在执行任务
      res_l=[]
      for i in range(10):
          res=p.apply_async(work,args=(i,)) # 同步运行,阻塞、直到本次任务执行完毕拿到res
          res_l.append(res)
  
      # 异步apply_async用法：如果使用异步提交的任务，主进程需要使用jion，等待进程池内任务都处理完，然后可以用get收集结果，否则，主进程结束，进程池可能还没来得及执行，也就跟着一起结束了
      p.close()
      p.join()
      for res in res_l:
          print(res.get()) # 使用get来获取apply_aync的结果,如果是apply,则没有get方法,因为apply是同步执行,立刻获取结果,也根本无需get
  

详解：apply_async与apply

• # 一：使用进程池（异步调用,apply_async）
  # coding: utf-8
  from multiprocessing import Process,Pool
  import time
  
  def func(msg):
      print( "msg:", msg)
      time.sleep(1)
      return msg
  
  if __name__ == "__main__":
      pool = Pool(processes = 3)
      res_l=[]
      for i in range(10):
          msg = "hello %d" %(i)
          res=pool.apply_async(func, (msg, ))   # 维持执行的进程总数为processes，当一个进程执行完毕后会添加新的进程进去
          res_l.append(res)
      print("==============================>") # 没有后面的join，或get，则程序整体结束，进程池中的任务还没来得及全部执行完也都跟着主进程一起结束了
  
      pool.close() # 关闭进程池，防止进一步操作。如果所有操作持续挂起，它们将在工作进程终止前完成
      pool.join()   # 调用join之前，先调用close函数，否则会出错。执行完close后不会有新的进程加入到pool,join函数等待所有子进程结束
  
      print(res_l) # 看到的是<multiprocessing.pool.ApplyResult object at 0x10357c4e0>对象组成的列表,而非最终的结果,但这一步是在join后执行的,证明结果已经计算完毕,剩下的事情就是调用每个对象下的get方法去获取结果
      for i in res_l:
          print(i.get()) # 使用get来获取apply_aync的结果,如果是apply,则没有get方法,因为apply是同步执行,立刻获取结果,也根本无需get
  

• # 二：使用进程池（同步调用,apply）
  # coding: utf-8
  from multiprocessing import Process,Pool
  import time
  
  def func(msg):
      print( "msg:", msg)
      time.sleep(0.1)
      return msg
  
  if __name__ == "__main__":
      pool = Pool(processes = 3)
      res_l=[]
      for i in range(10):
          msg = "hello %d" %(i)
          res=pool.apply(func, (msg, ))   # 维持执行的进程总数为processes，当一个进程执行完毕后会添加新的进程进去
          res_l.append(res) # 同步执行，即执行完一个拿到结果，再去执行另外一个
      print("==============================>")
      pool.close()
      pool.join()   # 调用join之前，先调用close函数，否则会出错。执行完close后不会有新的进程加入到pool,join函数等待所有子进程结束
  
      print(res_l) # 看到的就是最终的结果组成的列表
      for i in res_l: # apply是同步的，所以直接得到结果，没有get()方法
          print(i)
  

回掉函数：

需要回调函数的场景：进程池中任何一个任务一旦处理完了，就立即告知主进程：我好了额，你可以处理我的结果了。主进程则调用一个函数去处理该结果，该函数即回调函数

我们可以把耗时间（阻塞）的任务放到进程池中，然后指定回调函数（主进程负责执行），这样主进程在执行回调函数时就省去了I/O的过程，直接拿到的是任务的结果。

• from multiprocessing import Pool
  import requests
  import json
  import os
  
  def 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 pasrse_page(res):
      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)
      res_l=[]
      for url in urls:
          res=p.apply_async(get_page,args=(url,),callback=pasrse_page)
          res_l.append(res)
  
      p.close()
      p.join()
      print([res.get() for res in res_l]) # 拿到的是get_page的结果,其实完全没必要拿该结果,该结果已经传给回调函数处理了
  
  '''
  打印结果:
  <进程3388> get https://www.baidu.com
  <进程3389> get https://www.python.org
  <进程3390> get https://www.openstack.org
  <进程3388> get https://help.github.com/
  <进程3387> parse https://www.baidu.com
  <进程3389> get http://www.sina.com.cn/
  <进程3387> parse https://www.python.org
  <进程3387> parse https://help.github.com/
  <进程3387> parse http://www.sina.com.cn/
  <进程3387> parse https://www.openstack.org
  [{'url': 'https://www.baidu.com', 'text': '<!DOCTYPE html>\r\n...',...}]
  '''
  

爬虫案例

• from multiprocessing import Pool
  import time,random
  import requests
  import re
  
  def get_page(url,pattern):
      response=requests.get(url)
      if response.status_code == 200:
          return (response.text,pattern)
  
  def parse_page(info):
      page_content,pattern=info
      res=re.findall(pattern,page_content)
      for item in res:
          dic={
              'index':item[0],
              'title':item[1],
              'actor':item[2].strip()[3:],
              'time':item[3][5:],
              'score':item[4]+item[5]
          }
          print(dic)
  if __name__ == '__main__':
      pattern1=re.compile(r'<dd>.*?board-index.*?>(\d+)<.*?title="(.*?)".*?star.*?>(.*?)<.*?releasetime.*?>(.*?)<.*?integer.*?>(.*?)<.*?fraction.*?>(.*?)<',re.S)
  
      url_dic={
          'http://maoyan.com/board/7':pattern1,
      }
  
      p=Pool()
      res_l=[]
      for url,pattern in url_dic.items():
          res=p.apply_async(get_page,args=(url,pattern),callback=parse_page)
          res_l.append(res)
  
      for i in res_l:
          i.get()
  
      # res=requests.get('http://maoyan.com/board/7')
      # print(re.findall(pattern,res.text))
  

如果在主进程中等待进程池中所有任务都执行完毕后，再统一处理结果，则无需回调函数

• from multiprocessing import Pool
  import time,random,os
  
  def work(n):
      time.sleep(1)
      return n**2
  if __name__ == '__main__':
      p=Pool()
  
      res_l=[]
      for i in range(10):
          res=p.apply_async(work,args=(i,))
          res_l.append(res)
  
      p.close()
      p.join() # 等待进程池中所有进程执行完毕
  
      nums=[]
      for res in res_l:
          nums.append(res.get()) # 拿到所有结果
      print(nums) # 主进程拿到所有的处理结果,可以在主进程中进行统一进行处理
  

标签：__,get,res,编程,print,并发,进程,import
From： https://www.cnblogs.com/ioubbu/p/17702313.html