47異步編程_selectors

?

目錄

異步編程：... 1

同步、異步：... 1

阻塞、非阻塞：... 2

同步、異步與阻塞、非阻塞區別：... 2

聯系：... 2

IO模型... 2

py中的IO ?multiplexing——selectors：... 5

?

?

?

異步編程：

?

同步、異步：

你做完了，我才能做，同步，串行；我讓你打飯，你不打好給我，我不走開，直到你打飯給了我；

你做你的，我做我的，異步，步調不一致；我讓你打飯，你正打著，我不等你，但我會盯著你，你打完我過來拿走，異步并不保證多長時間打完飯；

?

函數或方法調用的時候，被調用者是否得到最終結果；

直接得到最終結果的，就是同步調用；

不直接得到最終結果的，就是異步調用；

某一方法是否在這個方法調用完時，是否能獲得最終結果；

?

同步、異步區別：

同步，一直要執行到返回結果；

異步，直接返回了，但不是最終結果，調用者不能通過這種調用得到結果，還要通過被調用者，使用其它方式通知調用者，來取回最終結果；

?

?

阻塞、非阻塞：

函數或方法調用時，是否立刻返回；

不立即返回，就是阻塞調用；

立即返回，就是非阻塞調用；

time.sleep()和event.wait()，event.wait()要優于time.sleep()，wait會主動讓出時間片，其它線程可以被調度，而sleep會占用時間片不讓出；

?

?

同步、異步與阻塞、非阻塞區別：

它們不相關；

同步、異步，強調的是結果；

阻塞、非阻塞強調的是時間，是否等待；

?

?

聯系：

注：調用者、被調用者；

同步阻塞，我啥事不干，就等你打飯給我，打飯是結果，而且我啥事不干一直等；

同步非阻塞，我等著你打飯給我，但我可以玩手機、看電視，打飯是結果，但我不一直等；

異步阻塞，我要打飯，你說等叫號，并沒有返回飯給我，我啥事不干，就等著飯好了你叫我，叫號；

異步非阻塞，我要打飯，你說等叫號，并沒有返回飯給我，我在旁邊玩手機、看電視，飯打好了你叫我；

?

?

IO模型

同步IO、異步IO、IO多路復用：

IO過程分2階段：

數據準備階段，內核從輸入設備讀寫數據（淘米、放鍋里煮）；

內核空間復制回用戶進程緩沖區階段，進程從內核復制數據（盛飯，從內核這個飯鍋里把飯裝到碗里來）；

?

注：

sendfile，disk緩沖區-->內核緩沖區（網絡緩沖區)-->發出；

0拷貝，網絡緩沖區上有disk緩沖區的快捷方式，直接從disk緩沖區發出；

?

sync同步IO模型：

阻塞IO、非阻塞IO、IO多路復用；

?

阻塞IO，sync blocking IO：

進程等待（阻塞），直到讀寫完成，全程等待；

?

非阻塞IO，sync non-blocking IO：

進程調用read操作，如果IO設備沒有準備好，立即返回ERROR，進程不阻塞，用戶可以再次發起系統調用，如果內核已準備好，就阻塞，然后復制數據到用戶空間；

第1階段數據沒有準備好，就先忙別的，等會再來看看，檢查數據是否準備好了的過程是非阻塞；systemcall后返回的是異常，直到return OK；

第2階段是阻塞（sync blocking）的，即內核空間和用戶空間之間復制數據是阻塞的；

淘米、煮飯我不等，我去玩會，盛飯過程我等著你裝好飯，但是要等到盛好飯才算完事，這是同步的，結果就是盛好飯；

?

IO多路復用，multiplexing：

同時監控多個IO，有一個準備好了，就不需要等了，開始處理，提高同時處理IO的能力；同時多路在等，而不是等完一個再等下一個；2個階段均是sync blocking；

select，所有平臺支持，poll是select的升級；

epoll，linux kernel2.5+支持，對select和poll的增強，在監視的基礎上，增加回調機制（通知機制）；BSD、MAC的kqueue；win的iocp；

通常用multiplexing；

以select為例，將關注的IO操作告訴select函數并調用，進程阻塞，內核監視select關注的fd文件描述符，被關注的任何一個fd對應的IO準備好了數據，select返回，在使用read將數據復制到用戶進程；

例，食堂供應很多菜（眾多的IO），你需要吃某三菜一湯，大師傅（OS）說要現做，需要等，你只好等待，其中一樣菜好了，大師傅叫你過來，你得自己找找看哪一樣菜好了，請服務員把做好的菜打給你，而epoll是有菜準備好了，大師傅喊你去幾號窗口直接打菜，不用自己找菜了；

?

async異步IO：

linux的aio系統調用，從kernel version2.6開始支持；

2階段都是異步，高并發時用；

進程發起異步IO請求，立即返回，內核完成IO的兩個階段，內核給進程發一個信號；

例，來打飯，跟大師傅說飯好了叫我，飯菜準備好了，窗口服務員把飯盛好了打電話叫你，兩階段都是異步的，在整個過程中，進程都可以忙別的，等好了才過來；

例，今天不想出去到飯店吃了，點外賣，飯菜在飯店做好了（第1階段），快遞員把飯送到你家門口（第2階段）；

?

?

?

py中的IO multiplexing——selectors：

select庫：

實現了select、poll系統調用，部分實現了epoll；

是底層的IO多路復用模塊，類似網絡編程的socket；

?

開發中選擇：

1、完全跨平臺，使用select、poll，但性能較差；

2、針對不同OS自行選擇支持的技術，這樣做會提高IO處理的性能；

?

selectors庫：

py3.4提供，高級IO復用庫；

?

類層次結構：

BaseSelector

???????? SelectSelector

???????? PollSelector

???????? EpollSelector

???????? DevpollSelector

???????? KqueueSelector

?

selectors.DefaultSelector()，返回當前平臺最有效、性能最高的實現；

但，此庫沒有實現win的iocp，將自動退化為select；

?

?

class BaseSelector(metaclass=ABCMeta):

??? @abstractmethod

??? def register(self, fileobj, events, data=None):?? #fileobj，被監視文件對象，如socket對象；events，該文件對象必須等待的事件；data，可選，與此文件對象相關聯的不透明數據，如，可用來存儲每個客戶端的會話ID，本例中關聯方法，就是讓selector干什么事，可以是任意對象，如果是函數類似回調

?

events常量：

EVENT_READ，可讀0x01，內核已準備好輸入輸出設備，可以開始讀了；

EVENT_WRITE，可寫0x10，內核已準備好，可以往里寫了；

?

注：

# generic events, that must be mapped to implementation-specific ones

EVENT_READ = (1 << 0)

EVENT_WRITE = (1 << 1)

?

單線程+selectors，單線程下用selectors而不用多線程，條件不滿足不能轉為就緒態，則不會被cpu調度；

多線程+阻塞，在OS中多了線程資源而已；

selectors可解決網絡IO、文件IO；

?

?

例：

selector = selectors.DefaultSelector()

?

def accept(sock: socket.socket):

??? conn, addrinfo = sock.accept()

??? conn.setblocking(False)

??? selector.register(conn, selectors.EVENT_READ, recv)

?

def recv(conn: socket.socket):

??? data = conn.recv(1024).strip().decode()

??? print(data)

??? msg = 'ack: {}'.format(data)

??? conn.send(msg.encode())

?

sock = socket.socket()

addr = ('127.0.0.1', 9999)

sock.bind(addr)

sock.listen()

sock.setblocking(False)

?

e = threading.Event()

?

key = selector.register(sock, selectors.EVENT_READ, accept)

print(key)

?

while not e.is_set():

??? events = selector.select()

??? if events:

??????? print(type(events))

?????????????????? print(events)

??? for key,mask in events:

??????? print(key, mask)

??????? callback = key.data

??????? callback(key.fileobj)

輸出：

SelectorKey(fileobj=<socket.socket fd=172, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=0, laddr=('127.0.0.1', 9999)>, fd=172, events=1, data=<function accept at 0x00000000011C1488>)

<class 'list'>

[(SelectorKey(fileobj=<socket.socket fd=172, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=0, laddr=('127.0.0.1', 9999)>, fd=172, events=1, data=<function accept at 0x00000000011C1488>), 1)]

SelectorKey(fileobj=<socket.socket fd=172, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=0, laddr=('127.0.0.1', 9999)>, fd=172, events=1, data=<function accept at 0x00000000011C1488>) 1

<class 'list'>

[(SelectorKey(fileobj=<socket.socket fd=216, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=0, laddr=('127.0.0.1', 9999), raddr=('127.0.0.1', 1570)>, fd=216, events=1, data=<function recv at 0x00000000012AAE18>), 1)]

SelectorKey(fileobj=<socket.socket fd=216, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=0, laddr=('127.0.0.1', 9999), raddr=('127.0.0.1', 1570)>, fd=216, events=1, data=<function recv at 0x00000000012AAE18>) 1

nimeide

?

?

例：

ChatServer改為IO多路復用方式；

?

注：

if mask == selectors.EVENT_READ:?? #不嚴格

if (mask & selectors.EVENT_READ) == selectors.EVENT_READ:?? #嚴格寫法，如果r和w同時滿足，進來3則不作處理

?

?

?

?