并发之协程

一, 回顾与思考

• 回顾:

  • 串行: 多个任务执行时,一个任务从开始执行,遇到了IO阻塞,原地等待,等待IO阻塞结束之后,继续执行,一个任务结束后再执行下一个
  • 并行: 多核,多个线程或进程同时执行,4个cpu,同时执行4个任务
  • 并发: 多个任务,cpu在多个任务之间来回切换(遇到IO阻塞或者执行时间过长),看起来像是同时执行
  • 并发的本质:
    1. 遇到IO阻塞或者执行时间过长切换cpu
    2. 保持原来的状态

• 当任务量不大时,是否有一种方法,可以实现一个线程的并发

  协程本质上就是一个线程,线程任务的切换是由操作系统控制的,遇到I/O自动切换,现在用协程的目的就是较少操作系统切换的开销(开关线程,创建寄存器,堆栈等,在他们之间进行切换等),在自己的程序里面来控制任务的切换

• yield: 类似协程

  #1 yiled可以保存状态,yield的状态保存与操作系统的保存线程状态很像,但是yield是代码级别控制的,更轻量级
  #2 send可以把一个函数的结果传给另外一个函数,以此实现单线程内程序之间的切换
  import time
  def func():
      for i in range(11):
          yield
          print('这是我第%s次打印了' % i)
          time.sleep(1)
  
  def func2():
      g = func()
      for k in range(10):
          print('哈哈,我第%s次打印了' % k)
          next(g)
  func2()
  # yield虽然可以实现两个任务来回切换,并且能够保存原来的状态,而且还是一个线程
  # 但是它只能遇到yield才能切换,遇到IO还是阻塞
  ----------------------------------------------------------
  # 计算密集型串行与协程的效率对比
  import time
  def task():
      res = 1
      for i in range(1, 100000):
          res += i
  def task2():
      res = 1
      for i in range(1, 100000):
          res -= i
  
  start_time = time.time()
  task()
  task2()
  print('串行消耗的时间:', time.time() - start_time)
  # 串行消耗的时间: 0.019963502883911133
  
  import time
  def task():
      res = 1
      for i in range(1, 100000):
          res += i
          yield res
  
  def task2():
      g = task()
      res = 1
      for i in range(1, 100000):
          res -= i
          next(g)
  start_time = time.time()
  task2()
  print('协程消耗的时间:', time.time() - start_time)
  # 协程消耗的时间: 0.03234148025512695
  

• 在自己的程序中(即用户程序级别,而非操作系统级别)控制单线程下的多个任务能在一个任务遇到IO阻塞时就切换到另外一个任务去计算,这样就保证了该线程能够最大限度地处于就绪态,即随时都可以被cpu执行的状态,相当于我们在用户程序级别将自己的IO操作最大限度地隐藏起来,从而可以迷惑操作系统,让其看到: 该线程好像是一直在计算,IO比较少,从而更多的将cpu的执行权限分配给我们的线程

• 协程的本质就是在单线程下,由用户自己控制一个任务遇到IO阻塞了就切换另外一个任务去执行,以此来提升效率.为了实现它,我们需要找寻一种可以同时满足以下条件的解决方案:

  1. 可以控制多个任务之间的切换,切换之前将任务的状态保存下来,以便重新运行时,可以基于暂停的位置继续执行.
  2. 可以检测IO操作,在遇到IO操作的情况下才发生切换.

二, 协程介绍

• 协程: 是单线程下的并发,又称微线程,纤程.英文名Coroutine.
• 一句话说明什么是协程: 协程是一种用户态的轻量级线程,即协程是由用户程序自己控制调度的
• python的线程属于内核级别的,即由操作系统控制调度(如单线程遇到IO或执行时间过长就会被迫交出cpu执行权限,切换其他线程运行)
• 对比操作系统控制线程的切换,用户在单线程内控制协程的切换,优点:
  • 协程的切换开销更小,属于程序级别的切换,操作系统完全感知不到,因而更加轻量级
  • 单线程内就可以实现并发的效果,最大限度地利用cpu
  • 修改共享数据不需要加锁?
• 缺点:
  • 协程的本质是单线程下,无法利用多核,可以是一个程序开启多个进程,每个进程内开启多个线程,每个线程内开启协程
  • 协程指的是单个线程,因而一旦协程出现阻塞,将会阻塞整个线程
• 协程的特点:
  1. 必须在只有一个单线程里实现并发
  2. 修改共享数据不需加锁
  3. 用户程序里自己保存多个控制流的上下文栈

三, Greenlet模块

• 如果我们在单个线程内有20个任务,要想实现在多个任务之间切换,使用yield生成器的方式过于麻烦(需要先得到初始化一次的生成器,然后再调用send,非常麻烦),而使用greenlet模块可以非常简单地实现这20个任务直接的切换

  # 真正的协程模块就是使用greenlet完成的切换
  from greenlet import greenlet
  
  def eat(name):
      print('%s eat 1' % name)  #2
      g2.switch('猪八戒')        #3
      print('%s eat 2' % name)  #6
      g2.switch()               #7
  def play(name):
      print('%s play 1' % name) #4
      g1.switch()               #5
      print('%s play 2' % name) #8
  
  g1 = greenlet(eat)
  g2 = greenlet(play)
  
  g1.switch('孙悟空') # 可以在第一次switch时传入参数，以后都不需要  1
  

• 仍然是没有解决遇到IO自动切换来提升效率的问题

• 我们想要实现的是:

  

  单线程里的20个任务的代码通常会既有计算操作又有阻塞操作,完全可以在执行任务1时遇到阻塞,就利用阻塞的时间去执行任务2.如此,才能提高效率,这就用到了Gevent模块

四, Gevent模块

• Gevent是一个第三方库,可以轻松通过Gevent实现并发同步或异步编程,在Gevent中用到的主要模式是Greenlet,它是以C扩展模块形式接入Python的轻量级协程.Greenlet全部运行在主程序操作系统进程的内部,但它们被协作式地调度

• 用法:

  1. g = gevent.spawn(func,*args,**kwargs)创建一个协程对象g,spawn括号内第一个参数是函数名,后面可以有多个参数,可以是位置实参或关键字实参,都是传给函数的,spawn是异步提交任务
  2. g.join() 等待g1结束
  3. gevent.joinall([g1])   可以合并多个协程对象的join
  4. g.value  拿到函数的返回值
  

• 模拟IO阻塞

  import gevent
  
  def eat(name):
      print(name, '吃1')
      gevent.sleep(2)
      print(name, '吃2')
  
  def play(name):
      print(name, '玩1')
      gevent.sleep(1)
      print(name, '玩2')
  
  g1 = gevent.spawn(eat, '猪八戒')
  g2 = gevent.spawn(play, '孙悟空')
  g1.join()
  g2.join()
  # 或gevent.joinall([g1, g2])
  print('主线程')
  
  # 此时的gevent只能识别gevent.sleep()
  

• 完整版Gevent

  # 想要让gevent识别所有阻塞,必须在最开头加入:
  from gevent import monkey
  monkey.patch_all()
  -------------------------------------------------------
  from gevent import monkey
  monkey.patch_all()
  # 必须写在最上面,这句话后面的所有阻塞全部能够识别了
  import gevent
  import time
  def eat(name):
      print(name, '吃1')
      time.sleep(2)
      print(name, '吃2')
  
  def play(name):
      print(name, '玩1')
      time.sleep(1)
      print(name, '玩2')
  
  g1 = gevent.spawn(eat, '猪八戒')
  g2 = gevent.spawn(play, '孙悟空')
  
  gevent.joinall([g1, g2])
  print('主线程')
  
  # 可以用threading.current_thread().getName()来查看每个g1和g2,查看的结果为DummyThread-n,即假线程,虚拟线程,其实都在一个线程里面;进程线程的任务切换是由操作系统自行切换的,自己不能控制;协程是通过自己的程序(代码)来进行切换的,自己能够控制,只有遇到协程模块能够识别的IO操作的时候,程序才会进行任务切换,实现并发效果,如果所有程序都没有IO操作,那么就基本属于串行执行了
  

• Gevent异步与同步对比

  from gevent import monkey
  monkey.patch_all()
  import gevent
  import time
  def task(id):
      # 此处放一些任务
      time.sleep(0.5)
      print(f'Task {id} done')
  
  def synchronous():   # 同步
      for i in range(10):
          task(i)
  
  def asynchronous():  # 异步
      g_l = [gevent.spawn(task, i) for i in range(10)]
      gevent.joinall(g_l)
  
  if __name__ == '__main__':
      print('同步:')
      synchronous()
      print('异步:')
      asynchronous()
  # 存在阻塞时,异步效率远高于同步
  

• Gevent应用举例一

  from gevent import monkey
  monkey.patch_all()
  import gevent
  import requests
  import time
  
  def get_page(url):
      print('分析:', url)
      response = requests.get(url)
      if response.status_code == 200:
          print(url, '分析结果:', len(response.text))
  
  start_time = time.time()
  gevent.joinall([
      gevent.spawn(get_page, 'https://www.python.org/'),
      gevent.spawn(get_page, 'https://www.yahoo.com/'),
      gevent.spawn(get_page, 'https://github.com/')
  ])
  print('用时:', time.time() - start_time)
  # 用时: 1.9783523082733154
  -------------------------------------------------------
  start_time = time.time()
  get_page('https://www.python.org/')
  get_page('https://www.yahoo.com/')
  get_page('https://github.com/')
  print('用时:', time.time() - start_time)
  # 用时: 5.849104881286621
  

• Gevent应用举例二

  # 服务端
  from gevent import monkey
  monkey.patch_all()
  import socket
  import gevent
  # 如果不想用money.patch_all()打补丁,可以用gevent自带的socket
  # from gevent import socket
  # s = socket.socket()
  
  def server(server_ip, port):
      s = socket.socket()
      s.bind((server_ip, port))
      s.listen(5)
      while 1:
          conn, addr = s.accept()
          gevent.spawn(talk, conn, addr)
  
  def talk(conn, addr):
      try:
          while 1:
              f_c = conn.recv(1024).decode('utf-8')
              print(f'来自{addr}的消息:{f_c}')
              conn.send(f_c.upper().encode('utf-8'))
      except Exception:
          print(f'断开与{addr}的连接')
      finally:
          conn.close()
  if __name__ == '__main__':
      server('127.0.0.1', 2019)
  --------------------------------------------------------
  # 客户端
  import socket
  c = socket.socket()
  c.connect(('127.0.0.1', 2019))
  
  while 1:
      t_s = input('>>>:').strip()
      if not t_s:
          continue
      c.send(t_s.encode('utf-8'))
      f_s = c.recv(1024).decode('utf-8')
      print('来自服务器的回复:', f_s)
  --------------------------------------------------------
  # 多线程并发多个客户端
  from threading import Thread
  from threading import current_thread
  import socket
  
  def client(server_ip, port):
      c = socket.socket()
      # 套接字对象一定要加到函数内,即局部名称空间内,放在函数外则被所有线程共享,
      # 则大家公用一个套接字对象,那么客户端端口永远一样了
      c.connect((server_ip, port))
  
      count = 0
      while 1:
          c.send(f'{current_thread().name}说:{count}'.encode('utf-8'))
          f_s = c.recv(1024).decode('utf-8')
          print(f_s)
          count += 1
  if __name__ == '__main__':
      for i in range(250):
          t = Thread(target=client, args=('127.0.0.1', 2019))
          t.start()