利用Python如何實(shí)現(xiàn)異步IO操作

本篇文章給大家分享的是有關(guān)利用Python如何實(shí)現(xiàn)異步IO操作，小編覺(jué)得挺實(shí)用的，因此分享給大家學(xué)習(xí)，希望大家閱讀完這篇文章后可以有所收獲，話(huà)不多說(shuō)，跟著小編一起來(lái)看看吧。

前言

　　用阻塞 API 寫(xiě)同步代碼最簡(jiǎn)單，但一個(gè)線程同一時(shí)間只能處理一個(gè)請(qǐng)求，有限的線程數(shù)導(dǎo)致無(wú)法實(shí)現(xiàn)萬(wàn)級(jí)別的并發(fā)連接，過(guò)多的線程切換也搶走了 CPU 的時(shí)間，從而降低了每秒能夠處理的請(qǐng)求數(shù)量。為了達(dá)到高并發(fā)，你可能會(huì)選擇一個(gè)異步框架，用非阻塞 API 把業(yè)務(wù)邏輯打亂到多個(gè)回調(diào)函數(shù)，通過(guò)多路復(fù)用與事件循環(huán)的方式實(shí)現(xiàn)高并發(fā)。

磁盤(pán) IO 為例，描述了多線程中使用阻塞方法讀磁盤(pán)，2 個(gè)線程間的切換方式。那么，怎么才能實(shí)現(xiàn)高并發(fā)呢？

把上圖中本來(lái)由內(nèi)核實(shí)現(xiàn)的請(qǐng)求切換工作，交由用戶(hù)態(tài)的代碼來(lái)完成就可以了，異步化編程通過(guò)應(yīng)用層代碼實(shí)現(xiàn)了請(qǐng)求切換，降低了切換成本和內(nèi)存占用空間。異步化依賴(lài)于 IO 多路復(fù)用機(jī)制，比如 Linux 的 epoll 或者 Windows 上的 iocp，同時(shí)，必須把阻塞方法更改為非阻塞方法，才能避免內(nèi)核切換帶來(lái)的巨大消耗。Nginx、Redis 等高性能服務(wù)都依賴(lài)異步化實(shí)現(xiàn)了百萬(wàn)量級(jí)的并發(fā)。

下圖描述了異步 IO 的非阻塞讀和異步框架結(jié)合后，是如何切換請(qǐng)求的。

然而，寫(xiě)異步化代碼很容易出錯(cuò)。因?yàn)樗凶枞瘮?shù)，都需要通過(guò)非阻塞的系統(tǒng)調(diào)用拆分成兩個(gè)函數(shù)。雖然這兩個(gè)函數(shù)共同完成一個(gè)功能，但調(diào)用方式卻不同。第一個(gè)函數(shù)由你顯式調(diào)用，第二個(gè)函數(shù)則由多路復(fù)用機(jī)制調(diào)用。

這種方式違反了軟件工程的內(nèi)聚性原則，函數(shù)間同步數(shù)據(jù)也更復(fù)雜。特別是條件分支眾多、涉及大量系統(tǒng)調(diào)用時(shí)，異步化的改造工作會(huì)非常困難。

Python如何實(shí)現(xiàn)異步調(diào)用

from flask import Flask
import time
app = Flask(__name__)


@app.route('/bar')
def bar():
  time.sleep(1)
  return '<h2>bar!</h2>'

@app.route('/foo')
def foo():
  time.sleep(1)
  return '<h2>foo!</h2>'
if __name__ == '__main__':
  app.run(host='127.0.0.1',port=5555,debug=True)

采用同步的方式調(diào)用

import requests
import time

starttime = time.time()
print(requests.get('http://127.0.0.1:5555/bar').content)
print(requests.get('http://127.0.0.1:5555/foo').content)
print("消耗時(shí)間: ",time.time() -starttime)

b'<h2>bar!</h2>'
b'<h2>foo!</h2>'
消耗時(shí)間:  2.015509605407715

采樣異步的方式調(diào)用：

重點(diǎn)：

1.將阻塞io改為非阻塞io；

2.多路復(fù)用io監(jiān)聽(tīng)內(nèi)核事件，事件觸發(fā)通過(guò)回調(diào)函數(shù)；

3.用戶(hù)態(tài)代碼采取事件循環(huán)的方式獲取事件，執(zhí)行事件的回調(diào)函數(shù)；

import selectors
import socket
import time
# from asynrequest import ParserHttp
class asynhttp:
  def __init__(self):
    self.selecter = selectors.DefaultSelector()

  def get(self,url,optiondict = None):
    global reqcount
    reqcount += 1
    s = socket.socket()
    s.setblocking(False)
    try:
      s.connect(('127.0.0.1',5555))
    except BlockingIOError:
      pass
    requset = 'GET %s HTTP/1.0\r\n\r\n' % url
    callback = lambda : self.send(s,requset)
    self.selecter.register(s.fileno(),selectors.EVENT_WRITE,callback)

  def send(self,s,requset):
    self.selecter.unregister(s.fileno())
    s.send(requset.encode())
    chunks = []
    callback = lambda: self.recv(s,chunks)
    self.selecter.register(s.fileno(),selectors.EVENT_READ,callback)

  def recv(self,s,chunks):
    self.selecter.unregister(s.fileno())
    chunk = s.recv(1024)
    if chunk:
      chunks.append(chunk)
      callback = lambda: self.recv(s,chunks)
      self.selecter.register(s.fileno(), selectors.EVENT_READ, callback)
    else:
      global reqcount
      reqcount -= 1
      request_first,request_headers,request_content,_ = ParserHttp.parser(b''.join(chunks))
      print("解析數(shù)據(jù)：",request_first,request_headers,request_content)
      print((b''.join(chunks)).decode())
      return (b''.join(chunks)).decode()

starttime = time.time()
reqcount = 0
asynhttper = asynhttp()
asynhttper.get('/bar')
asynhttper.get('/foo')
while reqcount:
  events = asynhttper.selecter.select()
  for event,mask in events:
    func = event.data
    func()
print("消耗時(shí)間：" ,time.time() - starttime)

HTTP/1.0 200 OK
Content-Type: text/html; charset=utf-8
Content-Length: 13
Server: Werkzeug/1.0.1 Python/3.7.7
Date: Thu, 15 Oct 2020 03:28:16 GMT

<h2>bar!</h2>
HTTP/1.0 200 OK
Content-Type: text/html; charset=utf-8
Content-Length: 13
Server: Werkzeug/1.0.1 Python/3.7.7
Date: Thu, 15 Oct 2020 03:28:16 GMT

<h2>foo!</h2>
消耗時(shí)間： 1.0127637386322021

以上就是利用Python如何實(shí)現(xiàn)異步IO操作，小編相信有部分知識(shí)點(diǎn)可能是我們?nèi)粘９ぷ鲿?huì)見(jiàn)到或用到的。希望你能通過(guò)這篇文章學(xué)到更多知識(shí)。更多詳情敬請(qǐng)關(guān)注億速云行業(yè)資訊頻道。