Python并发编程之IO模型

五种IO模型

为了更好地了解IO模型,我们需要事先回顾下:同步、异步、阻塞、非阻塞

  • 同步(synchronous) IO
  • 异步(asynchronous) IO
  • 阻塞(blocking) IO
  • 非阻塞(non-blocking)IO

五种I/O模型包括:阻塞I/O、非阻塞I/O、信号驱动I/O(不常用)、I/O多路转接、异步I/O。其中,前四个被称为同步I/O。

上五个模型的阻塞程度由低到高为:阻塞I/O > 非阻塞I/O > 多路转接I/O > 信号驱动I/O > 异步I/O,因此他们的效率是由低到高的。

1、阻塞I/O模型

在linux中,默认情况下所有的socket都是blocking,除非特别指定,几乎所有的I/O接口 ( 包括socket接口 ) 都是阻塞型的。

如果所面临的可能同时出现的上千甚至上万次的客户端请求,“线程池”或“连接池”或许可以缓解部分压力,但是不能解决所有问题。总之,多线程模型可以方便高效的解决小规模的服务请求,但面对大规模的服务请求,多线程模型也会遇到瓶颈,可以用非阻塞接口来尝试解决这个问题。

2、非阻塞I/O模型

在非阻塞式I/O中,用户进程其实是需要不断的主动询问kernel数据准备好了没有。但是非阻塞I/O模型绝不被推荐。

非阻塞,不等待。比如创建socket对某个地址进行connect、获取接收数据recv时默认都会等待(连接成功或接收到数据),才执行后续操作。 
如果设置setblocking(False),以上两个过程就不再等待,但是会报BlockingIOError的错误,只要捕获即可。

异步,通知,执行完成之后自动执行回调函数或自动执行某些操作(通知)。比如做爬虫中向某个地址baidu。com发送请求,当请求执行完成之后自执行回调函数。

3、多路复用I/O模型(事件驱动)

基于事件循环的异步非阻塞框架:如Twisted框架,scrapy框架(单线程完成并发)。

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

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

  • select:最多1024个socket;循环去检测。
  • poll:不限制监听socket个数;循环去检测(水平触发)。
  • epoll:不限制监听socket个数;回调方式(边缘触发)。

Python模块:

  • select.select
  • select.epoll

基于IO多路复用+socket非阻塞,实现并发请求(一个线程100个请求)

import socket
# 创建socket
client = socket.socket()
# 将原来阻塞的位置变成非阻塞(报错)
client.setblocking(False)

# 百度创建连接: 阻塞
try:
    # 执行了但报错了
    client.connect(('www.baidu.com',80))
except BlockingIOError as e:
    pass

# 检测到已经连接成功

# 问百度我要什么?
client.sendall(b'GET /s?wd=alex 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'))

selectors模块

#服务端
from socket import *
import selectors

sel=selectors.DefaultSelector()
def accept(server_fileobj,mask):
    conn,addr=server_fileobj.accept()
    sel.register(conn,selectors.EVENT_READ,read)

def read(conn,mask):
    try:
        data=conn.recv(1024)
        if not data:
            print('closing',conn)
            sel.unregister(conn)
            conn.close()
            return
        conn.send(data.upper()+b'_SB')
    except Exception:
        print('closing', conn)
        sel.unregister(conn)
        conn.close()

server_fileobj=socket(AF_INET,SOCK_STREAM)
server_fileobj.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
server_fileobj.bind(('127.0.0.1',8088))
server_fileobj.listen(5)
server_fileobj.setblocking(False) #设置socket的接口为非阻塞
sel.register(server_fileobj,selectors.EVENT_READ,accept) #相当于网select的读列表里append了一个文件句柄
                                                         #server_fileobj,并且绑定了一个回调函数accept

while True:
    events=sel.select() #检测所有的fileobj,是否有完成wait data的
    for sel_obj,mask in events:
        callback=sel_obj.data #callback=accpet
        callback(sel_obj.fileobj,mask) #accpet(server_fileobj,1)

#客户端
from socket import *
c=socket(AF_INET,SOCK_STREAM)
c.connect(('127.0.0.1',8088))

while True:
    msg=input('>>: ')
    if not msg:continue
    c.send(msg.encode('utf-8'))
    data=c.recv(1024)
    print(data.decode('utf-8'))

4、异步I/O

asyncio是Python 3.4版本引入的标准库,直接内置了对异步IO的支持。

asyncio的编程模型就是一个消息循环。我们从asyncio模块中直接获取一个EventLoop的引用,然后把需要执行的协程扔到EventLoop中执行,就实现了异步IO。

asyncio实现Hello world代码如下:

import asyncio

@asyncio.coroutine
def hello():
    print("Hello world!")
    # 异步调用asyncio.sleep(1):
    r = yield from asyncio.sleep(1)
    print("Hello again!")

# 获取EventLoop:
loop = asyncio.get_event_loop()
# 执行coroutine
loop.run_until_complete(hello())
loop.close()

@asyncio.coroutine把一个generator标记为coroutine类型,然后,我们就把这个coroutine扔到EventLoop中执行。

hello()会首先打印出Hello world!,然后,yield from语法可以让我们方便地调用另一个generator。由于asyncio.sleep()也是一个coroutine,所以线程不会等待asyncio.sleep(),而是直接中断并执行下一个消息循环。当asyncio.sleep()返回时,线程就可以从yield from拿到返回值(此处是None),然后接着执行下一行语句。

asyncio.sleep(1)看成是一个耗时1秒的IO操作,在此期间,主线程并未等待,而是去执行EventLoop中其他可以执行的coroutine了,因此可以实现并发执行。

我们用Task封装两个coroutine试试:

import threading
import asyncio

@asyncio.coroutine
def hello():
    print('Hello world! (%s)' % threading.currentThread())
    yield from asyncio.sleep(1)
    print('Hello again! (%s)' % threading.currentThread())

loop = asyncio.get_event_loop()
tasks = [hello(), hello()]
loop.run_until_complete(asyncio.wait(tasks))
loop.close()

观察执行过程:

Hello world! (<_MainThread(MainThread, started 140735195337472)>)
Hello world! (<_MainThread(MainThread, started 140735195337472)>)
(暂停约1秒)
Hello again! (<_MainThread(MainThread, started 140735195337472)>)
Hello again! (<_MainThread(MainThread, started 140735195337472)>)

由打印的当前线程名称可以看出,两个coroutine是由同一个线程并发执行的。

如果把asyncio.sleep()换成真正的IO操作,则多个coroutine就可以由一个线程并发执行。

我们用asyncio的异步网络连接来获取sina、sohu和163的网站首页:

import asyncio

@asyncio.coroutine
def wget(host):
    print('wget %s...' % host)
    connect = asyncio.open_connection(host, 80)
    reader, writer = yield from connect
    header = 'GET / HTTP/1.0\r\nHost: %s\r\n\r\n' % host
    writer.write(header.encode('utf-8'))
    yield from writer.drain()
    while True:
        line = yield from reader.readline()
        if line == b'\r\n':
            break
        print('%s header > %s' % (host, line.decode('utf-8').rstrip()))
    # Ignore the body, close the socket
    writer.close()

loop = asyncio.get_event_loop()
tasks = [wget(host) for host in ['www.sina.com.cn', 'www.sohu.com', 'www.163.com']]
loop.run_until_complete(asyncio.wait(tasks))
loop.close()

执行结果如下:

wget www.sohu.com...
wget www.sina.com.cn...
wget www.163.com...
(等待一段时间)
(打印出sohu的header)
www.sohu.com header > HTTP/1.1 200 OK
www.sohu.com header > Content-Type: text/html
...
(打印出sina的header)
www.sina.com.cn header > HTTP/1.1 200 OK
www.sina.com.cn header > Date: Wed, 20 May 2015 04:56:33 GMT
...
(打印出163的header)
www.163.com header > HTTP/1.0 302 Moved Temporarily
www.163.com header > Server: Cdn Cache Server V2.0
...

可见3个连接由一个线程通过coroutine并发完成。

async/await

asyncio提供的@asyncio.coroutine可以把一个generator标记为coroutine类型,然后在coroutine内部用yield from调用另一个coroutine实现异步操作。

为了简化并更好地标识异步IO,从Python 3.5开始引入了新的语法asyncawait,可以让coroutine的代码更简洁易读。

请注意,asyncawait是针对coroutine的新语法,要使用新的语法,只需要做两步简单的替换:

  • @asyncio.coroutine替换为async
  • yield from替换为await

让我们对比一下上一节的代码:

@asyncio.coroutine
def hello():
    print("Hello world!")
    r = yield from asyncio.sleep(1)
    print("Hello again!")

用新语法重新编写如下:

async def hello():
    print("Hello world!")
    r = await asyncio.sleep(1)
    print("Hello again!")

剩下的代码保持不变。

小结

asyncio提供了完善的异步IO支持;

异步操作需要在coroutine中通过yield from完成;

多个coroutine可以封装成一组Task然后并发执行。

到此这篇关于Python并发编程之IO模型的文章就介绍到这了。希望对大家的学习有所帮助,也希望大家多多支持我们。

(0)

相关推荐

  • Python如何加载模型并查看网络

    目录 加载模型并查看网络 打开终端 神经网络_模型的保存,模型的加载 模型的保存(torch.save) 模型的加载(torch.load) 加载模型并查看网络 加载模型,以vgg19为例. 打开终端 > python Python 3.7.2 (tags/v3.7.2:9a3ffc0492, Dec 23 2018, 23:09:28) [MSC v.1916 64 bit (AMD64)] on win32 Type "help", "copyright"

  • Python自动化测试PO模型封装过程详解

    在自动化中, Selenium 自动化测试中有一个名字经常被提及 PageObject( 思想与面向对象的特征相 同 ) ,通常 PO 模型可以大大提高测试用例的维护效率 优点:可重用,业务和对象分离,代码结构清晰,方便代码维护 核心要素 1. 在 PO 模式中抽离封装集成一个BasePage 类,该基类应该拥有一个只实现 webdriver 实例的属性 2. 每一个 page 都继承 BasePage ,通过 driver 来管理本 page 中元素,将 page 中的操作封装成一个个方法 3

  • Python详细介绍模型封装部署流程

    目录 一.桌面应用软件 二.Pyside2&Qt designer 三.模型封装部署 四.Pyinstaller 五.总结 一.桌面应用软件 桌面应用软件是基于GUI(Graphical User Interface,图形用户界面)交互式程序,需要实现GUI库实现前端交互. Python常见的GUI库 TKinter:基于Tk的Python库,这是Python官方采用的标准库,优点是作为Python标准库.稳定.发布程序较小,缺点是控件相对较少: wxPython:基于wxWidgets的Pyt

  • python的numpy模块实现逻辑回归模型

    使用python的numpy模块实现逻辑回归模型的代码,供大家参考,具体内容如下 使用了numpy模块,pandas模块,matplotlib模块 1.初始化参数 def initial_para(nums_feature):     """initial the weights and bias which is zero"""     #nums_feature是输入数据的属性数目,因此权重w是[1, nums_feature]维     #

  • python封装成exe的超详细教程

    目录 第一种:.py文件直接封装成exe 第二种:整个项目封装成exe 补充说明: 总结 第一种:.py文件直接封装成exe 1.cmd进入py文件所在的目录 备注:在py文件所在的目录下,按住shift+鼠标右击,然后找到“在此处打开PowerShell窗口”,即可进入当前目录 2.输入以下代码: 备注:使用-D制作出来的exe比使用-F的快很多,因为-F把所有dll文件都打包到一个exe中了(-F这时候exe会很大,加载变慢,推荐-D) #-w:不显示后台 -i添加图标 pyinstalle

  • Python封装数据库连接池详解

    目录 一.数据库封装 1.1数据库基本配置 1.2 编写单例模式注解 1.3 构建连接池 1.4 封装Python操作MYSQL的代码 二.连接池测试 场景一:同一个实例,执行2次sql 场景二:依次创建2个实例,各自执行sql 场景三:启动2个线程,但是线程在创建连接池实例时,有时间间隔 场景四:启动2个线程,线程在创建连接池实例时,没有时间间隔 前言: 线程安全问题:当2个线程同时用到线程池时,会同时创建2个线程池.如果多个线程,错开用到线程池,就只会创建一个线程池,会共用一个线程池.我用的

  • Python实现自动驾驶训练模型

    目录 一.安装环境 二.配置环境 三.训练模型 1.数据处理 2.搭建模型 3.运行结果 四.总结 一.安装环境 gym是用于开发和比较强化学习算法的工具包,在python中安装gym库和其中子场景都较为简便. 安装gym: pip install gym 安装自动驾驶模块,这里使用Edouard Leurent发布在github上的包highway-env: pip install --user git+https://github.com/eleurent/highway-env 其中包含6

  • Python并发编程之IO模型

    五种IO模型 为了更好地了解IO模型,我们需要事先回顾下:同步.异步.阻塞.非阻塞 同步(synchronous) IO 异步(asynchronous) IO 阻塞(blocking) IO 非阻塞(non-blocking)IO 五种I/O模型包括:阻塞I/O.非阻塞I/O.信号驱动I/O(不常用).I/O多路转接.异步I/O.其中,前四个被称为同步I/O. 上五个模型的阻塞程度由低到高为:阻塞I/O > 非阻塞I/O > 多路转接I/O > 信号驱动I/O > 异步I/O,因

  • Java网络编程之IO模型阻塞与非阻塞简要分析

    目录 1.阻塞I/O模型 2.非阻塞I/O模型 1.阻塞I/O模型 阻塞IO模型是常见的IO模型,在读写数据时客户端会发生阻塞.阻塞IO模型的工作流程为: 1.1在用户线程发出IO请求之后,内核会检查数据是否就绪,此时用户线程一直阻塞等待内存数据就绪: 1.2在内存数据就绪后,内核将数据复制到用户线程中,并返回I/O执行结果到用户线程,此时用户线程将解除阻塞状态并开始处理数据. 典型的阻塞I/O模型的例子为data= socket.read(),如果内核数据没有就绪, Socket线程就会一直阻

  • python 并发编程 多路复用IO模型详解

    多路复用IO(IO multiplexing) 这种IO方式为事件驱动IO(event driven IO). 我们都知道,select/epoll的好处就在于单个进程process就可以同时处理多个网络连接的IO.它的基本原理就是select/epoll这个function会不断的轮询所负责的所有socket,当某个socket有数据到达了,就通知用户进程.它的流程如图: select是多路复用的一种 当用户进程调用了select,那么整个进程会被block,而同时,kernel会"监视&qu

  • python 并发编程 阻塞IO模型原理解析

    阻塞IO(blocking IO) 在linux中,默认情况下所有的socket都是blocking,一个典型的读操作流程大概是这样: 当用户进程调用了recvfrom这个系统调用,kernel内核就开始了IO的第一个阶段:准备数据.对于network io( 网络io )来说,很多时候数据在一开始还没有到达(比如,还没有收到一个完整的UDP包),这个时候kernel( 内核 )就要等待足够的数据到来. 等着对方把数据放到自己操作系统内存 而在用户进程这边,整个进程会被阻塞.当kernel一直等

  • 并发编程之Java内存模型volatile的内存语义

    1.volatile的特性 理解volatile特性的一个好办法是把对volatile变量的单个读/写,看成是使用同一个锁对单个读/写操作做了同步. 代码示例: package com.lizba.p1; /** * <p> * volatile示例 * </p> * * @Author: Liziba * @Date: 2021/6/9 21:34 */ public class VolatileFeatureExample { /** 使用volatile声明64位的long型

  • 并发编程之Java内存模型顺序一致性

    目录 1.数据竞争和顺序一致性 1.1 Java内存模型规范对数据竞争的定义 1.2 JMM对多线程程序的内存一致性做的保证 2.顺序一致性内存模型 2.1 特性 2.2 举例说明顺序一致性模型 2.3 同步程序的顺序一致性效果 2.4 未同步程序的执行特性 3. 64位long型和double型变量写原子性 3.1 CPU.内存和总线简述 3.2 long和double类型的操作 简介: 顺序一致性内存模型是一个理论参考模型,处理器的内存模型和编程语言的内存模型都会以顺序一致性内存模型作为参照

  • 并发编程之Java内存模型锁的内存语义

    目录 1.锁的释放-获取建立的happens-before关系 2.锁释放和获取的内存语义 3.锁内存的语义实现 4.concurrent包的实现 简介: 锁的作用是让临界区互斥执行.本文阐述所得另一个重要知识点--锁的内存语义. 1.锁的释放-获取建立的happens-before关系 锁是Java并发编程中最重要的同步机制.锁除了让临界区互斥执行外,还可以让释放锁的线程向获取同一个锁的线程发送消息. 锁释放-获取的示例代码: package com.lizba.p1; /** * <p>

  • Python网络编程之ZeroMQ知识总结

    一.ZeroMQ概述 ZeroMQ(又名ØMQ,MQ,或zmq)像一个可嵌入的网络库,但其作用就像一个并发框架. ZeroMQ类似于标准Berkeley套接字,其提供了各种传输工具,如进程内.进程间.TCP和组播中进行原子消息传送的套接字 可以使用各种模式实现N对N的套接字连接,这些模式包括:发布-订阅.任务分配.请求-应答. ZeroMQ的速度足够快,因此可充当集群产品的结构. ZeroMQ的异步I/O模型提供了可扩展的多核应用程序,用异步消息来处理任务 ZeroMQ核心由C语言编写,支持C.

  • Python 网络编程之TCP客户端/服务端功能示例【基于socket套接字】

    本文实例讲述了Python 网络编程之TCP客户端/服务端功能.分享给大家供大家参考,具体如下: demo.py(TCP客户端): import socket def main(): # 1. 创建tcp的套接字 tcp_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # 2. 链接服务器 # tcp_socket.connect(("192.168.33.11", 7890)) server_ip = input(

  • Python Web编程之WSGI协议简介

    本文实例讲述了Python Web编程之WSGI协议.分享给大家供大家参考,具体如下: WSGI简介 Web框架和Wen服务器之间需要进行通信,如果在设计时它们之间无法相互匹配,那么对框架的选择就会限制对Web服务器的选择,这显然是不合理的.这时候需要设计一套双方都遵守的接口.WSGI是Python Web Server Gateway Interface的简称.WSGI标准在PEP 333中定义并被许多框架实现,它规定了一种在Web服务器之间具有可移植性.在后来的PEP 3333中添加了Pyt

随机推荐