第三十八篇 vue

Posted 2023-04-05 caix-1987 - 个人博客

概述

Vue 在大多数常见场景下性能都是很优秀的，通常不需要手动优化。然而，总会有一些具有挑战性的场景需要进行针对性的微调。在本节中，我们将讨论用 Vue 开发的应用在性能方面该注意些什么

首先，让我们区分一下 web 应用性能的两个主要方面

  1、页面加载性能
  
    首次访问时，应用展示出内容与达到可交互状态的速度。这通常会用 Google 所定义的一系列 Web 指标 (Web Vitals) 来进行衡量，如最大内容绘制 (Largest Contentful Paint，缩写为 LCP) 和首次输入延迟 (First Input Delay，缩写为 FID)。

  2、更新性能
  
    应用响应用户输入更新的速度。比如当用户在搜索框中输入时结果列表的更新速度，或者用户在一个单页面应用 (SPA) 中点击链接跳转页面时的切换速度。

虽然最理想的情况是将两者都最大化，但是不同的前端架构往往会影响到在这些方面是否能达到更理想的性能。此外，你所构建的应用的类型极大地影响了你在性能方面应该优先考虑的问题。因此，优化性能的第一步是为你的应用类型确定合适的架构

分析选项

为了提高性能，我们首先需要知道如何衡量它。在这方面，有一些很棒的工具可以提供帮助：

1、用于生产部署的负载性能分析：

  1、PageSpeed Insights
  
  2、WebPageTest
  
2、用于本地开发期间的性能分析：

  1、Chrome 开发者工具“性能”面板

   app.config.performance 将会开启 Vue 特有的性能标记，标记在 Chrome 开发者工具的  性能时间线上
   
  2、Vue 开发者扩展也提供了性能分析的功能。

页面加载优化

页面加载优化有许多跟框架无关的方面 - 这份 web.dev 指南提供了一个全面的总结。这里，我们将主要关注和 Vue 相关的技巧

选用正确的架构

选用正确的架构

如果你的用例对页面加载性能很敏感，请避免将其部署为纯客户端的 SPA，而是让服务器直接发送包含用户想要查看的内容的 HTML 代码。纯客户端渲染存在首屏加载缓慢的问题，这可以通过服务器端渲染 (SSR) 或静态站点生成 (SSG) 来缓解。查看 SSR 指南以了解如何使用 Vue 实现 SSR。如果应用对交互性要求不高，你还可以使用传统的后端服务器来渲染 HTML，并在客户端使用 Vue 对其进行增强。

如果你的主应用必须是 SPA，但还有其他的营销相关页面 (落地页、关于页、博客等)，请单独部署这些页面！理想情况下，营销页面应该是包含尽可能少 JS 的静态 HTML，并用 SSG 方式部署

包体积与 Tree-shaking 优化

一个最有效的提升页面加载速度的方法就是压缩 JavaScript 打包产物的体积。当使用 Vue 时有下面一些办法来减小打包产物体积：

1、尽可能地采用构建步骤

    1、如果使用的是相对现代的打包工具，许多 Vue 的 API 都是可以被 tree-shake 的。举例来说，如果你根本没有使用到内置的 <Transition> 组件，它将不会被打包进入最终的产物里。Tree-shaking 也可以移除你源代码中其他未使用到的模块。

    2、当使用了构建步骤时，模板会被预编译，因此我们无须在浏览器中载入 Vue 编译器。这在同样最小化加上 gzip 优化下会相对缩小 14kb 并避免运行时的编译开销。

2、在引入新的依赖项时要小心包体积膨胀！在现实的应用中，包体积膨胀通常因为无意识地引入了过重的依赖导致的。

    1、如果使用了构建步骤，应当尽量选择提供 ES 模块格式的依赖，它们对 tree-shaking 更友好。举例来说，选择 lodash-es 比 lodash 更好。

    2、查看依赖的体积，并评估与其所提供的功能之间的性价比。如果依赖对 tree-shaking 友好，实际增加的体积大小将取决于你从它之中导入的 API。像 bundlejs.com 这样的工具可以用来做快速的检查，但是根据实际的构建设置来评估总是最准确的。

3、如果你只在渐进式增强的场景下使用 Vue，并想要避免使用构建步骤，请考虑使用 petite-vue (只有 6kb) 来代替

代码分割

代码分割是指构建工具将构建后的 JavaScript 包拆分为多个较小的，可以按需或并行加载的文件。通过适当的代码分割，页面加载时需要的功能可以立即下载，而额外的块只在需要时才加载，从而提高性能。

像 Rollup (Vite 就是基于它之上开发的) 或者 webpack 这样的打包工具可以通过分析 ESM 动态导入的语法来自动进行代码分割：

// lazy.js 及其依赖会被拆分到一个单独的文件中
// 并只在 `loadLazy()` 调用时才加载

function loadLazy() 
  return import(\'./lazy.js\')


懒加载对于页面初次加载时的优化帮助极大，它帮助应用暂时略过了那些不是立即需要的功能。在 Vue 应用中，这可以与 Vue 的异步组件搭配使用，为组件树创建分离的代码块：

import  defineAsyncComponent  from \'vue\'

// 会为 Foo.vue 及其依赖创建单独的一个块
// 它只会按需加载
//（即该异步组件在页面中被渲染时）

const Foo = defineAsyncComponent(() => import(\'./Foo.vue\'))

对于使用了 Vue Router 的应用，强烈建议使用异步组件作为路由组件。Vue Router 已经显性地支持了独立于 defineAsyncComponent 的懒加载。查看懒加载路由了解更多细节

更新优化

Props 稳定性

在 Vue 之中，一个子组件只会在其至少一个 props 改变时才会更新。思考以下示例：

<ListItem
  v-for="item in list"
  :id="item.id"
  :active-id="activeId" />
  
在 <ListItem> 组件中，它使用了 id 和 activeId 两个 props 来确定它是否是当前活跃的那一项。虽然这是可行的，但问题是每当 activeId 更新时，列表中的每一个 <ListItem> 都会跟着更新！

理想情况下，只有活跃状态发生改变的项才应该更新。我们可以将活跃状态比对的逻辑移入父组件来实现这一点，然后让 <ListItem> 改为接收一个 active prop

<ListItem
  v-for="item in list"
  :id="item.id"
  :active="item.id === activeId" />
  
现在，对于大多数的组件来说，activeId 改变时，它们的 active prop 都会保持不变，因此它们无需再更新。总结一下，这个技巧的核心思想就是让传给子组件的 props 尽量保持稳定

v-once

v-once 是一个内置的指令，可以用来渲染依赖运行时数据但无需再更新的内容。它的整个子树都会在未来的更新中被跳过。查看它的 API 参考手册可以了解更多细节

v-memo

v-memo 是一个内置指令，可以用来有条件地跳过某些大型子树或者 v-for 列表的更新。查看它的 API 参考手册可以了解更多细节

通用优化

大型虚拟列表

所有的前端应用中最常见的性能问题就是渲染大型列表。无论一个框架性能有多好，渲染成千上万个列表项都会变得很慢，因为浏览器需要处理大量的 DOM 节点。

但是，我们并不需要立刻渲染出全部的列表。在大多数场景中，用户的屏幕尺寸只会展示这个巨大列表中的一小部分。我们可以通过列表虚拟化来提升性能，这项技术使我们只需要渲染用户视口中能看到的部分。

要实现列表虚拟化并不简单，幸运的是，你可以直接使用现有的社区库：

1、vue-virtual-scroller

2、vue-virtual-scroll-grid

3、vueuc/VVirtualList

减少大型不可变数据的响应性开销

Vue 的响应性系统默认是深度的。虽然这让状态管理变得更直观，但在数据量巨大时，深度响应性也会导致不小的性能负担，因为每个属性访问都将触发代理的依赖追踪。好在这种性能负担通常这只有在处理超大型数组或层级很深的对象时，例如一次渲染需要访问 100,000+ 个属性时，才会变得比较明显。因此，它只会影响少数特定的场景。

Vue 确实也为此提供了一种解决方案，通过使用 shallowRef() 和 shallowReactive() 来绕开深度响应。浅层式 API 创建的状态只在其顶层是响应式的，对所有深层的对象不会做任何处理。这使得对深层级属性的访问变得更快，但代价是，我们现在必须将所有深层级对象视为不可变的，并且只能通过替换整个根状态来触发更新

const shallowArray = shallowRef([
  /* 巨大的列表，里面包含深层的对象 */
])

// 这不会触发更新...
shallowArray.value.push(newObject)
// 这才会触发更新
shallowArray.value = [...shallowArray.value, newObject]

// 这不会触发更新...
shallowArray.value[0].foo = 1
// 这才会触发更新
shallowArray.value = [
  
    ...shallowArray.value[0],
    foo: 1
  ,
  ...shallowArray.value.slice(1)
]

避免不必要的组件抽象

有些时候我们会去创建无渲染组件或高阶组件 (用来渲染具有额外 props 的其他组件) 来实现更好的抽象或代码组织。虽然这并没有什么问题，但请记住，组件实例比普通 DOM 节点要昂贵得多，而且为了逻辑抽象创建太多组件实例将会导致性能损失。

需要提醒的是，只减少几个组件实例对于性能不会有明显的改善，所以如果一个用于抽象的组件在应用中只会渲染几次，就不用操心去优化它了。考虑这种优化的最佳场景还是在大型列表中。想象一下一个有 100 项的列表，每项的组件都包含许多子组件。在这里去掉一个不必要的组件抽象，可能会减少数百个组件实例的无谓性能消耗

补充

无渲染组件

一些组件可能只包括了逻辑而不需要自己渲染内容，视图输出通过作用域插槽全权交给了消费者组件。我们将这种类型的组件称为无渲染组件

<MouseTracker v-slot=" x, y ">
  Mouse is at:  x ,  y 
</MouseTracker>

Python之路(第三十八篇) 并发编程：进程同步锁/互斥锁信号量事件队列生产者消费者模型

一、进程锁（同步锁/互斥锁）

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

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

例子

  #并发运行,效率高,但竞争同一打印终端,带来了打印错乱
  from multiprocessing import Process
  import os,time
  def work():
      print(‘%s is running‘ %os.getpid())
      time.sleep(2)
      print(‘%s is done‘ %os.getpid())
  ?
  if __name__ == ‘__main__‘:
      for i in range(3):
          p=Process(target=work)
          p.start()

　　

 

加锁后

  
  #加锁后由并发变成了串行,牺牲了运行效率,但避免了竞争
  ?
  from multiprocessing import Process,Lock
  import os,time
  def work(mutex):
      mutex.acquire() #开始加锁
      print(‘%s is running‘ %os.getpid())
      time.sleep(2)
      print(‘%s is done‘ %os.getpid())
      mutex.release() #释放锁，在加锁期间别的进程都要等
  ?
  if __name__ == ‘__main__‘:
      mutex = Lock()
      for i in range(3):
          p=Process(target=work,args=(mutex,))
          p.start()

　　

 

例子2

多个进程共享同一文件

文件当数据库，模拟抢票

 

未加锁版

  #文件db.txt的内容为：{"count":1}
  #注意一定要用双引号，不然json无法识别
  ?
  # 并发运行，效率高，但竞争写同一文件，数据写入错乱
  from multiprocessing import Process,Lock
  import time,json,random,os
  def search():
      dic=json.load(open(‘db.txt‘))
      print(‘33[43m剩余票数%s33[0m‘ %dic[‘count‘])
  ?
  def get():
      dic=json.load(open(‘db.txt‘))
      time.sleep(0.1) #模拟读数据的网络延迟
      if dic[‘count‘] >0:
          dic[‘count‘]-=1
          time.sleep(0.2) #模拟写数据的网络延迟
          json.dump(dic,open(‘db.txt‘,‘w‘))
          print(‘%s33[43m购票成功33[0m‘%(os.getpid()))
  ?
  def task(lock):
      search()
      get()
  if __name__ == ‘__main__‘:
      lock=Lock()
      for i in range(10): #模拟并发10个客户端抢票
          p=Process(target=task,args=(lock,))
          p.start()

　　

输出结果

  剩余票数1
  剩余票数1
  剩余票数1
  剩余票数1
  剩余票数1
  剩余票数1
  剩余票数1
  剩余票数1
  剩余票数1
  剩余票数1
  4120购票成功
  2692购票成功
  7328购票成功
  13444购票成功
  13632购票成功
  13560购票成功
  13752购票成功
  12564购票成功
  13720购票成功
  13488购票成功

　　

加锁版

 

  
  import multiprocessing,time,json,random
  ?
  def search(name):
      with open("db.txt","r",encoding="utf-8") as f:
          data_dic = json.load(f)
          time.sleep(random.uniform(0,2))
          if data_dic["count"] >= 1 :
              print("已查询到票还有%s张,当前系统时间 %s"%(data_dic["count"],time.asctime()))
          else:
              print("系统票源不足！当前系统时间 %s"%time.asctime())
  ?
  def buy(name):
      with open("db.txt","r+",encoding="utf-8") as f:
          data_dic = json.load(f)
      if data_dic["count"] > 0 :
          with open("db.txt", "w", encoding="utf-8") as g:
              new_ticket_count = data_dic["count"] - 1
              data_dic.update({"count":new_ticket_count})
              json.dump(data_dic,g)
          print("%s购票成功！"%name)
      else:
          print("%s购票失败！"%name)
  ?
  def task(name,mutex):
      search(name)  # 查询无需加锁
      mutex.acquire()
      buy(name)  #针对修改文件的关键操作加锁
      mutex.release()
  ?
  ?
  if __name__ == "__main__":
      mutex = multiprocessing.Lock()
      for i in range(10):
          p = multiprocessing.Process(target=task,args=("乘客%s"%i,mutex))
          p.start()
 

　　

分析

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

　　

 

二、信号量（multiprocess.Semaphore）

互斥锁同时只允许一个线程更改数据，而信号量Semaphore是同时允许一定数量的线程更改数据 。实现：信号量同步基于内部计数器，每调用一次acquire()，计数器减1；每调用一次release()，计数器加1.当计数器为0时，acquire()调用被阻塞。这是迪科斯彻（Dijkstra）信号量概念P()和V()的Python实现。信号量同步机制适用于访问像服务器这样的有限资源。信号量与进程池的概念很像，但是要区分开，信号量涉及到加锁的概念。

例子

  # 多进程中的组件
  # ktv
  # 4个
  # 一套资源  同一时间 只能被n个人访问
  # 某一段代码 同一时间 只能被n个进程执行
  import time
  import random
  from multiprocessing import Process
  from multiprocessing import Semaphore
  ?
  # sem = Semaphore(4)
  # sem.acquire()
  # print(‘拿到第一把钥匙‘)
  # sem.acquire()
  # print(‘拿到第二把钥匙‘)
  # sem.acquire()
  # print(‘拿到第三把钥匙‘)
  # sem.acquire()
  # print(‘拿到第四把钥匙‘)
  # sem.acquire()
  # print(‘拿到第五把钥匙‘)
  def ktv(i,sem):
      sem.acquire()    #获取钥匙
      print(‘%s走进ktv‘%i)
      time.sleep(random.randint(1,5))
      print(‘%s走出ktv‘%i)
      sem.release()
  ?
  ?
  if __name__ == ‘__main__‘ :
      sem = Semaphore(4)
      for i in range(20):
          p = Process(target=ktv,args=(i,sem))
          p.start()

　　

 

三、事件(multiprocess.Event)

python线程的事件用于主线程控制其他线程的执行，事件主要提供了三个方法 set、wait、clear。

事件处理的机制：全局定义了一个“Flag”，如果“Flag”值为 False，那么当程序执行 event.wait 方法时就会阻塞，如果“Flag”值为True，那么event.wait 方法时便不再阻塞。

clear：将“Flag”设置为False,set：将“Flag”设置为True.

 

例子

  ?

  from multiprocessing import Event
  ?
  e = Event()
  print(e.is_set()) #初始设置为False
  print("数据111")
  e.set()  #设置之后为True
  print("数据222")
  print(e.is_set()) #打印设置之后的状态
  e.wait()  #当值为False会阻塞，当值为Ture是，不会阻塞
  print("数据333")
  e.clear() #清除事件状态，设置为False
  print(e.is_set())  #打印清除之后的状态
  print("数据444")
  e.wait()  #此时值为False，程序会一直阻塞
  print("数据555")

　　

输出结果

  False
  数据111
  数据222
  True
  数据333
  False
  数据444

　　

 

例子

简单的红绿灯事件

  from multiprocessing import Event,Process
  import time
  import random
  ?
  ?
  def cars(e,num):
      if not e.is_set(): # 进程刚开启，is_set()的值是False，模拟信号灯为红色
          print("%s车正在等待通行"%num)
          e.wait() # 阻塞，等待信号灯切换
      print("%s车已经通过" % num) #打印已经通过的进程
  ?
  ?
  def light(e):
  ?
      #模拟定时切换红绿灯
      while True:
          if e.is_set():
              e.clear() #>将is_set()的值设置为False
              print("33[31m红灯亮了33[0m")
          else:
              e.set() #>将is_set()的值设置为True
              print("33[32m绿灯亮了33[0m")
          time.sleep(2)
  ?
  if __name__ == "__main__":
      e = Event()
      traffic = Process(target=light,args=(e,))
      traffic.start() #启动红绿灯进程
      for i in range(20):
          car = Process(target=cars,args=(e,"布加迪%s"%i))
          car.start()
          time.sleep(random.random())

　　

 

四、进程间通信——队列和管道

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

队列

　队列就相当于一个容器，里面可以放数据，特点是先放进去先拿出来，即先进先出。

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

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

　　

参数

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

　　

　 方法介绍：

  
  q.put方法用以插入数据到队列中，put方法还有两个可选参数：blocked和timeout。如果blocked为True（默认值），并且timeout为正值，该方法会阻塞timeout指定的时间，直到该队列有剩余的空间。如果超时，会抛出Queue.Full异常。如果blocked为False，但该Queue已满，会立即抛出Queue.Full异常。
  q.get方法可以从队列读取并且删除一个元素。同样，get方法有两个可选参数：blocked和timeout。如果blocked为True（默认值），并且timeout为正值，那么在等待时间内没有取到任何元素，会抛出Queue.Empty异常。如果blocked为False，有两种情况存在，如果Queue有一个值可用，则立即返回该值，否则，如果队列为空，则立即抛出Queue.Empty异常.
   
  q.get_nowait():同q.get(False)
  q.put_nowait():同q.put(False)
  ?
  q.empty():调用此方法时q为空则返回True，该结果不可靠，比如在返回True的过程中，如果队列中又加入了项目。
  q.full()：调用此方法时q已满则返回True，该结果不可靠，比如在返回True的过程中，如果队列中的项目被取走。
  q.qsize():返回队列中目前项目的正确数量，结果也不可靠，理由同q.empty()和q.full()一样

　　

其他方法(了解)：

  
  q.close() 
  关闭队列，防止队列中加入更多数据。调用此方法时，后台线程将继续写入那些已入队列但尚未写入的数据，但将在此方法完成时马上关闭。如果q被垃圾收集，将自动调用此方法。关闭队列不会在队列使用者中生成任何类型的数据结束信号或异常。例如，如果某个使用者正被阻塞在get（）操作上，关闭生产者中的队列不会导致get（）方法返回错误。
  ?
  q.cancel_join_thread() 
  不会再进程退出时自动连接后台线程。这可以防止join_thread()方法阻塞。
  ?
  q.join_thread() 
  连接队列的后台线程。此方法用于在调用q.close()方法后，等待所有队列项被消耗。默认情况下，此方法由不是q的原始创建者的所有进程调用。调用q.cancel_join_thread()方法可以禁止这种行为。

　　

例子

  from multiprocessing import Queue
  ?
  q = Queue(3)  # 创建一个队列对象，并给他设置容器大小，即能放几个数据
  q.put(1)  # put（）方法是往容器里放数据
  q.put([2,3])
  q.put({"k1":4})
  # q.put("mi") # 如果队列已经满了，程序就会停在这里，等待数据被别人取走，再将数据放入队列。
  try:
      q.put_nowait(3) # 可以使用put_nowait，如果队列满了不会阻塞，但是会因为队列满了而报错。
  except: # 因此我们可以用一个try语句来处理这个错误。这样程序不会一直阻塞下去，但是会丢掉这个消息。
      print(‘队列已经满了‘)
  ?
  # 因此，我们再放入数据之前，可以先看一下队列的状态，如果已经满了，就不继续put了。
  print(q.full()) #返回True ,满了
  print(q.get())  #get（）方法是从容器里拿数据
  print(q.get())
  print(q.get())
  # 同put方法一样，如果队列已经空了，那么继续取就会出现阻塞。
  try:
      q.get_nowait() # 可以使用get_nowait，如果队列满了不会阻塞，但是会因为没取到值而报错。
  except: # 因此我们可以用一个try语句来处理这个错误。这样程序不会一直阻塞下去。
      print(‘队列已经空了‘)
  ?
  print(q.empty()) #空了

　　

例子

  import time
  from multiprocessing import Queue, Process
  ?
  ?
  def task(q):
      q.put(" hello! 时间%s"%time.asctime())  # 调用主函数中p进程传递过来的进程参数 put函数为向队列中添加一条数据。
  ?
  ?
  if __name__ == ‘__main__‘:
      q = Queue(3)#创建一个Queue对象
      p = Process(target=task, args=(q,)) #创建一个子进程
      p.start()
      print(q.get()) #在主进程打印从子进程获取的数据

　　

 

生产者消费者模型

生产者消费者模型

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

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

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

什么是生产者和消费者模式

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

 

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

  from multiprocessing import Process, Queue
  import time, random, os
  ?
  ?
  def consumer(q):
      while True:
          res = q.get()
          time.sleep(random.randint(1, 3))
          print(‘33[45m%s 吃 %s33[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(‘33[44m%s 生产了 %s33[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出死循环

  import time, random, os
  from multiprocessing import Process, Queue
  ?
  ?
  def consumer(q):
      while True:
          res = q.get()
          if res is None: break  # 收到结束信号则结束
          time.sleep(random.randint(1, 3))
          print(‘33[45m%s 吃 %s33[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(‘33[44m%s 生产了 %s33[0m‘ % (os.getpid(), res))
      q.put(None)  # 发送结束信号，生产者在生产完毕后发送结束信号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信号。

  import multiprocessing
  import time
  import random
  ?
  ?
  def producer(name, q):
      for i in range(2):
          res = "包子%s" % i
          time.sleep(random.randint(0, 1))
          print("%s生产了%s" % (name, res))
          q.put(res)
  ?
  ?
  def consumer(name, q):
      while True:
          res = q.get()
          if q.get() is None:  # 收到结束信号则结束
              print("没包子吃了")
              break
          print("%s吃了%s" % (name, res))
  ?
  ?
  if __name__ == "__main__":
      q = multiprocessing.Queue()
      p1 = multiprocessing.Process(target=producer, args=("jack", q))
      p2 = multiprocessing.Process(target=producer, args=("charles", q))
      p3 = multiprocessing.Process(target=producer, args=("pony", q))
      c1 = multiprocessing.Process(target=consumer, args=("nick", q))
      c2 = multiprocessing.Process(target=consumer, args=("nicholas", q))
      p_list = []
      p_list.append(p1)
      p_list.append(p2)
      p_list.append(p3)
      for p in p_list:
          p.start()
      c1.start()
      c2.start()
      p1.join() #必须保证生产者全部生产完毕,才应该发送结束信号
      p2.join()
      p3.join()
      q.put(None)  # 发送结束信号，有几个消费者就应该发送几次结束信号None
      q.put(None)  # 发送结束信号
      print("end........")
 

　　

　　

这里有另外一种队列提供了这种机制，JoinableQueue。

JoinableQueue([maxsize])

其实就是一种队列，但又比队列要多两种方法，task_done()和join()方法，正是有这两种方法就可以解决上面的问题。

创建可连接的共享进程队列。这就像是一个Queue对象，但队列允许项目的使用者通知生产者项目已经被成功处理。通知进程是使用共享的信号和条件变量来实现的。

 

方法介绍

  JoinableQueue的实例p除了与Queue对象相同的方法之外，还具有以下方法：
  ?
  q.task_done() 
  使用者使用此方法发出信号，表示q.get()返回的项目已经被处理。如果调用此方法的次数大于从队列中删除的项目数量，将引发ValueError异常。
  ?
  q.join() 
  生产者将使用此方法进行阻塞，直到队列中所有项目均被处理。阻塞将持续到为队列中的每个项目均调用q.task_done()方法为止。 
  下面的例子说明如何建立永远运行的进程，使用和处理队列上的项目。生产者将项目放入队列，并等待它们被处理。

　　

例子

import multiprocessing
import time
import random


def producer(name, q):
    for i in range(2):
        res = "包子%s" % i
        time.sleep(random.randint(0, 1))
        print("%s生产了%s" % (name, res))
        q.put(res)
    q.join()  # 只有顾客把队列的包子全部拿走后，三个生产者进程才能全部结束


def consumer(name, q):
    while True:
        res = q.get()
        print("%s吃了%s" % (name, res))
        q.task_done()  # 发信号告诉队列，又吃完了一个，从队列中取走一个数据并处理完成


if __name__ == "__main__":
    # q = multiprocessing.Queue()
    q = multiprocessing.JoinableQueue()
    p1 = multiprocessing.Process(target=producer, args=("jack", q))
    p2 = multiprocessing.Process(target=producer, args=("charles", q))
    p3 = multiprocessing.Process(target=producer, args=("pony", q))
    c1 = multiprocessing.Process(target=consumer, args=("nick", q))
    c2 = multiprocessing.Process(target=consumer, args=("nicholas", q))
    p_list = []
    p_list.append(p1)
    p_list.append(p2)
    p_list.append(p3)
    for p in p_list:
        p.start()
    c1.daemon = True  # 将c1c2设置成守护进程，只要主进程结束了，那么顾客就收到了所有的数据
    c2.daemon = True
    c1.start()
    c2.start()
    p1.join()
    p2.join()
    p3.join()

    print("end........")
# 主进程等--->p1,p2,p3等---->c1,c2
# p1,p2,p3结束了,证明c1,c2肯定全都收完了p1,p2,p3发到队列的数据
# 因而c1,c2也没有存在的价值了,不需要继续阻塞在进程中影响主进程了。
# 应该随着主进程的结束而结束,所以设置成守护进程就可以了。