python 复习—并发编程——IO多路复用协程

Posted 胖虎是只mao

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了python 复习—并发编程——IO多路复用协程相关的知识,希望对你有一定的参考价值。

一、铺垫:基于socket发送http请求

1、需求一:向百度发送请求搜索关键字“alex”,有如下两种方式:


    import requests
    ret = requests.get('https://www.baidu.com/s?wd=alex')

方式二(socket方式,也是requests的原理):

	import socket
    sk = socket.socket()
    # 与百度创建连接: 阻塞
    sk.connect(('www.baidu.com',80))
    # 跟说百度我要什么?
    sk.sendall(b'GET /s?wd=alex     HTTP/1.0\\r\\nhost:www.baidu.com\\r\\n\\r\\n')
    # 等着接收百度给我的回复
    chunk_list = []
    while 1:
        chunk = sk.recv(8096)
        if not chunk:
            break
        chunk_list.append(chunk)
    body = b''.join(chunk_list)
    print(body.decode('utf8'))


2、需求二:向百度发送请求搜索三个关键字

    import requests
    key_list = ['alex','db','sb']
    for item in key_list:
        ret = requests.get('https://www.baidu.com/s?wd=%s' %item)
	import socket
    def get_data(key):
        client = socket.socket()
        # 跟百度创建连接: 阻塞
        client.connect(('www.baidu.com',80))
        # 跟百度说我要什么?
        client.sendall(b'GET /s?wd='+key.encode('utf-8')+b' HTTP/1.0\\r\\nhost:www.baidu.com\\r\\n\\r\\n')
        # 我等着接收百度给我的回复
        chunk_list = []
        while True:
            chunk = client.recv(8096)
            if not chunk:
                break
            chunk_list.append(chunk)

        body = b''.join(chunk_list)
        print(body.decode('utf-8'))

    key_list = ['alex','db','sb']
    for item in key_list:
        get_data(item)

方式二:

分析上述需求二的代码,我们发现,这两种方式去向浏览器发送请求的时候都是串行的,也就是等第一个请求得到响应之后再发送下一个请求,并没有实现并发。现在你可能会想:可以创建多线程来分别去发送请求,代码如下:

# #################### 解决并发:多线程 ####################
    import threading

    key_list = ['alex','db','sb']
    for item in key_list:
        t = threading.Thread(target=get_data,args=(item,))
        t.start()

多线程虽然提高了效率,实现了并发,但是同时也浪费了资源,那我们想一下能不能用单线程实现并发,也就是这个线程去发送完第一个请求(IO请求)后不等待相应结果,而是直接去发送第二个请求,再继续发送第三个请求,等请求响应之后才去处理响应结果,这样就实现了单线程并发,即节省了资源又实现了并发,那具体怎么实现呢?

首先需要解决两个问题:
第一:如何判断是IO请求?
第二:如何知道响应数据回来了?

二、基于IO多路复用+socket实现单线程并发

	# ################ 解决并发:单线程+IO不等待 ################
  import socket
  import select

  client1 = socket.socket()
  client1.setblocking(False) # 将原来阻塞的位置变成非阻塞(报错)
  try:
      client1.connect(('www.baidu.com',80))
  except BlockingIOError as e:
      pass

  client2 = socket.socket()
  client2.setblocking(False)  # 将原来阻塞的位置变成非阻塞(报错)
  try:
      client2.connect(('www.sogou.com',80))
  except BlockingIOError as e:
      pass

  client3 = socket.socket()
  client3.setblocking(False)  # 将原来阻塞的位置变成非阻塞(报错)
  try:
      client3.connect(('www.sina.com.cn',80))
  except BlockingIOError as e:
      pass

  socket_list = [client1,client2,client3]
  conn_list = [client1,client2,client3]

  while True:
      rlist,wlist,elist = select.select(socket_list,conn_list,[],0.005)
      # rlist中表示已经接收到数据的socket对象   
      # wlist中表示已经连接成功的socket对象
      for sk in wlist:
          if sk == client1:
              sk.sendall(b'GET /s?wd=alex HTTP/1.0\\r\\nhost:www.baidu.com\\r\\n\\r\\n')
          elif sk == client2:
              sk.sendall(b'GET /web?query=fdf HTTP/1.0\\r\\nhost:www.sogou.com\\r\\n\\r\\n')
          else:
              sk.sendall(b'GET /mid/search.shtml?q=alex HTTP/1.0\\r\\nhost:www.sina.com.cn\\r\\n\\r\\n')
          conn_list.remove(sk)
      for sk in rlist:
          chunk_list = []
          while True:
              try:
                  chunk = sk.recv(8096)
                  if not chunk:
                      break
                  chunk_list.append(chunk)
              except BlockingIOError as e:
                  break
          body = b''.join(chunk_list)
          print('------------>',body)
          sk.close()
          socket_list.remove(sk)
      if not socket_list:
          break

上面示例可以进行封装,但是封装前先来看这样两段代码:

	# 代码一:
    v = [
        [11,22], # 每个都有一个append方法
        [22,33], # 每个都有一个append方法
        [33,44], # 每个都有一个append方法
    ]
    for item in v:
        print(item.append)
	# 代码二(为了不改变for循环代码,可以进行如下封装)
    class Foo(object):
        def __init__(self,data):
            self.row = data

        def append(self,item):
            self.row.append(item)
    v = [
        Foo([11,22]), # 每个都有一个append方法
        Foo([22,33]), # 每个都有一个append方法
        Foo([33,44]), # 每个都有一个append方法
    ]

    for item in v:
        print(item.append)
# ############## 单线程并发高级版:封装上面示例 ##############
    import socket
    import select

    class Req(object):
        def __init__(self,sk,func):
            self.sock = sk
            self.func = func

        def fileno(self):
            return self.sock.fileno()

    class Nb(object):
        def __init__(self):
            self.conn_list = []
            self.socket_list = []

        def add(self,url,func):
            client = socket.socket()
            client.setblocking(False)  # 非阻塞
            try:
                client.connect((url, 80))
            except BlockingIOError as e:
                pass
            obj = Req(client,func)
            self.conn_list.append(obj)
            self.socket_list.append(obj)

        def run(self):
            while True:
                rlist,wlist,elist = select.select(self.socket_list,self.conn_list,[],0.005)
                for sk in wlist:
                    # 发生变换的req对象
                    sk.sock.sendall(b'GET /s?wd=alex HTTP/1.0\\r\\nhost:www.baidu.com\\r\\n\\r\\n')
                    self.conn_list.remove(sk)
                for sk in rlist:
                    chunk_list = []
                    while True:
                        try:
                            chunk = sk.sock.recv(8096)
                            if not chunk:
                                break
                            chunk_list.append(chunk)
                        except BlockingIOError as e:
                            break
                    body = b''.join(chunk_list)
                    sk.func(body)
                    sk.sock.close()
                    self.socket_list.remove(sk)
                if not self.socket_list:
                    break

    def baidu_repsonse(body):
        print('百度下载结果:',body)

    def sogou_repsonse(body):
        print('搜狗下载结果:', body)

    def sina_repsonse(body):
        print('新浪下载结果:', body)

    t1 = Nb()
    t1.add('www.baidu.com',baidu_repsonse)
    t1.add('www.sogou.com',sogou_repsonse)
    t1.add('www.sina.com.cn',sina_repsonse)
    t1.run()

封装版

总结:

1、socket默认是否是阻塞的?阻塞体现在哪里?

      是,体现在等待连接和等待接收数据。

2、如何让socket编程非阻塞?

      通过设置client.setblocking(False)

3、IO多路复用作用?

      检测多个socket是否已经发生变化(是否已经连接成功/是否已经获取数据)(可写/可读)

      操作系统检测socket是否发生变化,有三种模式:

             select:最多1024个socket,循环去检测;

             poll:不限制监听socket个数,循环去检测(水平触发);

             epoll:不限制监听socket个数,回调方式(边缘触发);

      Python模块:

             select.select

             select.epoll(windows不支持,linux中可以用)

4、提高并发方案:

      - 多进程

      - 多线程

      - 异步非阻塞模块(Twisted), 爬虫中学的scrapy框架(内部是用单线程完成并发)

5、什么是异步非阻塞?

      - 非阻塞,不等待。

             比如创建socket对某个地址进行connect、获取接收数据recv时默认都会等待(连接成功或接收到数据),才执行后续操作。

             如果设置setblocking(False),以上两个过程就不再等待,但是会报BlockingIOError的错误,只要捕获即可。

      - 异步,通知,执行完成之后自动执行回调函数或自动执行某些操作(通知)。

             比如做爬虫中向某个地址baidu.com发送请求,当请求执行完成之后自动执行回调函数。

6、什么是同步阻塞?

      - 阻塞:等

      - 同步:按照顺序逐步执行,例如:
key_list = ['alex','db','sb']
    for item in key_list:
        ret = requests.get('https://www.baidu.com/s?    wd=%s' %item)
        print(ret.text)

三、协程

并发编程有三种方式:

  • 多线程 Thread

  • 多进程 Process

  • 多协程 Coroutine

    进程和线程都是操作系统中存在的,而协程是由程序员创造出来的一个不是真实存在的东西。

    协程:是微线程,对一个线程进行分片,使得线程在代码块之间进行来回切换执行,而不是原来的逐行执行。如下示例:

	import greenlet   
    # 引入greenlet模块帮助我们实现协程,安装方式:pip3 install greenlet

    def f1():
        print(11)
        gr2.switch()
        print(22)
        gr2.switch()

    def f2():
        print(33)
        gr1.switch()
        print(44)

    gr1 = greenlet.greenlet(f1)  # 创建协程 gr1
    gr2 = greenlet.greenlet(f2)  # 创建协程 gr2

    gr1.switch()  # 执行协程gr1

创建协程

分析:单纯的协程没有意义,反而可能会让性能降低,那么协程的存在意义在哪里呢?结合上面单线程实现并发的示例,思考一下假如当我们执行了一段代码后遇到IO操作,此时我们不再等待,而是切换到另一段代码去执行,然后遇到IO操作的时候再去切换,这样是不是也能提高性能,实现并发,但是greenlet只能做协程,不能实现遇到IO就切换,所以协程如果再加上遇到IO就切换,那么便能实现单线程并发了。那么谁能做到遇到IO就切换呢?那就是另外一个模块geven,安装方法:pip3 install gevent。

gevent内部要依赖greenlet,也就是greenlet + IO切换,所以gevent就牛逼了!写法如下:

	from gevent import monkey
  monkey.patch_all() # 以后代码中遇到IO都会自动执行greenlet的switch进行切换
  import requests
  import gevent

  def get_page1(url):
      ret = requests.get(url)
      print(url,ret.content)

  def get_page2(url):
      ret = requests.get(url)
      print(url,ret.content)

  def get_page3(url):
      ret = requests.get(url)
      print(url,ret.content)

  gevent.joinall([
      gevent.spawn(get_page1, 'https://www.python.org/'), # 创建协程1
      gevent.spawn(get_page2, 'https://www.yahoo.com/'),  # 创建协程2
      gevent.spawn(get_page3, 'https://github.com/')     # 创建协程3
  ])

面通过gevent实现了单线程并发,提高了效率,通过对比,我们发现,上面IO多路复用的示例中是一个线程在不停的执行,而是gevent是在代码间进行切换,虽然原理不行,但是都提高了效率,实现单线程并发。

总结:

   1、协程可以提高并发吗?

          协程自己本身无法实现并发,甚至性能会降低,而协程+IO切换性能就可以提高了。

   2、单线程提高并发的方法有哪些?

          a、协程+遇到就IO切换:gevent;   注意:不是异步,无回调函数,但本质也是基于事件循环

          b、基于时间循环的异步非阻塞框架:Twisted;

   3、线程、进程、协程的区别?

进程cpu资源分配的最小单元,主要用来做数据隔离,那么线程是cpu工作的最小单元,一个应用程序可以有多个进程(默认有一个),一个进程可以有多个线程(默认有一个),这是它们的一个简单区别;基本上在其他语言中没有进程这个概念,大都用线程,而在python中由于有GIL锁,它保证了同一时刻一个进程中只能有一个线程被cpu调度,为了利用多核优势就要创建多个进程,多线程没有用,所以计算密集型的用多进程,IO密集型的用多线程就行,因为IO操作不占用CPU。而协程是程序员人为创造出来的不真实存在的,它可以让程序员控制代码执行顺序,在函数之间来回切换,本身协程存在没有意义,但是能跟IO切换放在一起就厉害了,相当于将线程切片,程序遇到IO就切换到其他代码,IO完成后再切回来,达到让线程不停去工作的效果,实现协程的模块是greenlet,实现协程+IO切换的模块是gevent,这就是三者的区别。

4、手动实现协程:yield关键字生成器(没有意义,了解即可)

	def f1():
        print(11)
        yield
        print(22)
        yield
        print(33)

    def f2():
        print(55)
        yield
        print(66)
        yield
        print(77)

    v1 = f1()
    v2 = f2()

    next(v1) # v1.send(None)
    next(v2) # v1.send(None)
    next(v1) # v1.send(None)
    next(v2) # v1.send(None)
    next(v1) # v1.send(None)
    next(v2) # v1.send(None)

手动实现协程

进程、线程、协程区别:

python 多线程只能同时使用一个CPU但是多线程也能用于IO 的密集型计算,因为CPU和IO是可以并行进行的。而进程只能同时使用单个CPU,多进程就可以同时利用多核CPU

在这里插入图片描述
在这里插入图片描述

以上是关于python 复习—并发编程——IO多路复用协程的主要内容,如果未能解决你的问题,请参考以下文章

IO多路复用/基于IO多路复用+socket实现并发请求/协程

python并发编程-多路复用IO

进程,线程,协程,io多路复用 总结

协程与IO多路复用

python全栈开发day36-IO多路复用

Python并发编程-IO模型-IO多路复用实现SocketServer