摘要：前言通過了解異步設(shè)計(jì)的由來，來深入理解異步事件機(jī)制。代碼地址什么是異步同步并發(fā)線程多路復(fù)用異步回調(diào)參考文獻(xiàn)什么是異步為了深入理解異步的概念，就必須先了解異步設(shè)計(jì)的由來。使得維護(hù)這個(gè)列表更加容易，它會(huì)幫你在合適的位置插入新的定時(shí)器事件組。

通過了解異步設(shè)計(jì)的由來，來深入理解異步事件機(jī)制。

代碼地址

什么是異步

同步

并發(fā)(Concurrency)

線程(Thread)

I/O多路復(fù)用

異步(Asynchronous)

回調(diào)(Callback)

參考文獻(xiàn)

為了深入理解異步的概念，就必須先了解異步設(shè)計(jì)的由來。

顯然易見的是，同步的概念隨著我們學(xué)習(xí)第一個(gè)輸出Hello World的程序，就已經(jīng)深入人心。

然而我們也很容易忘記一個(gè)事實(shí)：一個(gè)現(xiàn)代編程語言(如Python)做了非常多的工作，來指導(dǎo)和約束你如何去構(gòu)建你自己的一個(gè)程序。

def f():
    print("in f()")
def g():
    print("in g()")
f()
g()

你知道in g()一定輸出在in f()之后，即函數(shù)f完成前函數(shù)g不會(huì)執(zhí)行。這即為同步。在現(xiàn)代編程語言的幫助下，這一切顯得非常的自然，從而也讓我們可以將我們的程序分解成
松散耦合的函數(shù)：一個(gè)函數(shù)并不需要關(guān)心誰調(diào)用了它，它甚至可以沒有返回值，只是完成一些操作。

當(dāng)然關(guān)于這些是怎么具體實(shí)現(xiàn)的就不探究了，然而隨著一個(gè)程序的功能的增加，同步設(shè)計(jì)的開發(fā)理念并不足以實(shí)現(xiàn)一些復(fù)雜的功能。

寫一個(gè)程序每隔3秒打印“Hello World”，同時(shí)等待用戶命令行的輸入。用戶每輸入一個(gè)自然數(shù)n，就計(jì)算并打印斐波那契函數(shù)的值F(n)，之后繼續(xù)等待下一個(gè)輸入

由于等待用戶輸入是一個(gè)阻塞的操作，如果按照同步的設(shè)計(jì)理念：如果用戶未輸入，則意味著接下來的函數(shù)并不會(huì)執(zhí)行，自然沒有辦法做到一邊輸出“Hello World”，
一邊等待用戶輸入。為了讓程序能解決這樣一個(gè)問題，就必須引入并發(fā)機(jī)制，即讓程序能夠同時(shí)做很多事，線程是其中一種。

具體代碼在example/hello_threads.py中。

from threading import Thread
from time import sleep
from time import time
from fib import timed_fib
def print_hello():
    while True:
        print("{} - Hello world!".format(int(time())))
        sleep(3)
def read_and_process_input():
    while True:
        n = int(input())
        print("fib({}) = {}".format(n, timed_fib(n)))
def main():
    # Second thread will print the hello message. Starting as a daemon means
    # the thread will not prevent the process from exiting.
    t = Thread(target=print_hello)
    t.daemon = True
    t.start()
    # Main thread will read and process input
    read_and_process_input()
if __name__ == "__main__":
    main()

對(duì)于之前那樣的問題，引入線程機(jī)制就可以解決這種簡單的并發(fā)問題。而對(duì)于線程我們應(yīng)該有一個(gè)簡單的認(rèn)知：

一個(gè)線程可以理解為指令的序列和CPU執(zhí)行的上下文的集合。

一個(gè)同步的程序即進(jìn)程，有且只會(huì)在一個(gè)線程中運(yùn)行，所以當(dāng)線程被阻塞，也就意味著整個(gè)進(jìn)程被阻塞

一個(gè)進(jìn)程可以有多個(gè)線程，同一個(gè)進(jìn)程中的線程共享了進(jìn)程的一些資源，比如說內(nèi)存，地址空間，文件描述符等。

線程是由操作系統(tǒng)的調(diào)度器來調(diào)度的， 調(diào)度器統(tǒng)一負(fù)責(zé)管理調(diào)度進(jìn)程中的線程。

系統(tǒng)的調(diào)度器決定什么時(shí)候會(huì)把當(dāng)前線程掛起，并把CPU的控制器交個(gè)另一個(gè)線程。這個(gè)過程稱之為稱上下文切換，包括對(duì)于當(dāng)前線程上下文的保存、對(duì)目標(biāo)線程上下文的加載。

上下文切換會(huì)對(duì)性能產(chǎn)生影響，因?yàn)樗旧硪残枰狢PU的周期來執(zhí)行

而隨著現(xiàn)實(shí)問題的復(fù)雜化，如10K問題。

在Nginx沒有流行起來的時(shí)候，常被提到一個(gè)詞 10K（并發(fā)1W）。在互聯(lián)網(wǎng)的早期，網(wǎng)速很慢、用戶群很小需求也只是簡單的頁面瀏覽，
所以最初的服務(wù)器設(shè)計(jì)者們使用基于進(jìn)程/線程模型，也就是一個(gè)TCP連接就是分配一個(gè)進(jìn)程(線程)。誰都沒有想到現(xiàn)在Web 2.0時(shí)候用戶群里和復(fù)雜的頁面交互問題，
而現(xiàn)在即時(shí)通信和實(shí)在實(shí)時(shí)互動(dòng)已經(jīng)很普遍了。那么你設(shè)想如果每一個(gè)用戶都和服務(wù)器保持一個(gè)（甚至多個(gè)）TCP連接才能進(jìn)行實(shí)時(shí)的數(shù)據(jù)交互，別說BAT這種量級(jí)的網(wǎng)站，
就是豆瓣這種比較小的網(wǎng)站，同時(shí)的并發(fā)連接也要過億了。進(jìn)程是操作系統(tǒng)最昂貴的資源，一臺(tái)機(jī)器無法創(chuàng)建很多進(jìn)程。如果要?jiǎng)?chuàng)建10K個(gè)進(jìn)程，那么操作系統(tǒng)是無法承受的。
就算我們不討論隨著服務(wù)器規(guī)模大幅上升帶來復(fù)雜度幾何級(jí)數(shù)上升的問題，采用分布式系統(tǒng)，只是維持1億用戶在線需要10萬臺(tái)服務(wù)器，成本巨大，也只有FLAG、BAT這樣公司才有財(cái)力購買如此多的服務(wù)器。

而同樣存在一些原因，讓我們避免考慮多線程的方式：

線程在計(jì)算和資源消耗的角度來說是比較昂貴的。

線程并發(fā)所帶來的問題，比如因?yàn)楣蚕淼膬?nèi)存空間而帶來的死鎖和競態(tài)條件。這些又會(huì)導(dǎo)致更加復(fù)雜的代碼，在編寫代碼的時(shí)候需要時(shí)不時(shí)地注意一些線程安全的問題。

為了解決這一問題，出現(xiàn)了「用同一進(jìn)程/線程來同時(shí)處理若干連接」的思路，也就是I/O多路復(fù)用。

以Linux操作系統(tǒng)為例，Linux操作系統(tǒng)給出了三種監(jiān)聽文件描述符的機(jī)制，具體實(shí)現(xiàn)可參考：

select: 每個(gè)連接對(duì)應(yīng)一個(gè)描述符（socket），循環(huán)處理各個(gè)連接，先查下它的狀態(tài)，ready了就進(jìn)行處理，不ready就不進(jìn)行處理。但是缺點(diǎn)很多：

每次調(diào)用select，都需要把fd集合從用戶態(tài)拷貝到內(nèi)核態(tài)，這個(gè)開銷在fd很多時(shí)會(huì)很大

同時(shí)每次調(diào)用select都需要在內(nèi)核遍歷傳遞進(jìn)來的所有fd，這個(gè)開銷在fd很多時(shí)也很大

select支持的文件描述符數(shù)量太小了，默認(rèn)是1024

poll: 本質(zhì)上和select沒有區(qū)別，但是由于它是基于鏈表來存儲(chǔ)的，沒有最大連接數(shù)的限制。缺點(diǎn)是：

大量的的數(shù)組被整體復(fù)制于用戶態(tài)和內(nèi)核地址空間之間，而不管這樣的復(fù)制是不是有意義。

poll的特點(diǎn)是「水平觸發(fā)(只要有數(shù)據(jù)可以讀，不管怎樣都會(huì)通知)」，如果報(bào)告后沒有被處理，那么下次poll時(shí)會(huì)再次報(bào)告它。

epoll: 它使用一個(gè)文件描述符管理多個(gè)描述符，將用戶關(guān)系的文件描述符的事件存放到內(nèi)核的一個(gè)事件表中，這樣在用戶空間和內(nèi)核空間的copy只需一次。epoll支持水平觸發(fā)和邊緣觸發(fā)，最大的特點(diǎn)在于「邊緣觸發(fā)」，它只告訴進(jìn)程哪些剛剛變?yōu)榫途w態(tài)，并且只會(huì)通知一次。使用epoll的優(yōu)點(diǎn)很多：

沒有最大并發(fā)連接的限制，能打開的fd的上限遠(yuǎn)大于1024（1G的內(nèi)存上能監(jiān)聽約10萬個(gè)端口）

效率提升，不是輪詢的方式，不會(huì)隨著fd數(shù)目的增加效率下降

內(nèi)存拷貝，利用mmap()文件映射內(nèi)存加速與內(nèi)核空間的消息傳遞；即epoll使用mmap減少復(fù)制開銷

綜上所述，通過epoll的機(jī)制，給現(xiàn)代高級(jí)語言提供了高并發(fā)、高性能解決方案的基礎(chǔ)。而同樣FreeBSD推出了kqueue，Windows推出了IOCP，Solaris推出了/dev/poll。

而在Python3.4中新增了selectors模塊，用于封裝各個(gè)操作系統(tǒng)所提供的I/O多路復(fù)用的接口。
那么之前同樣的問題，我們可以通過I/O多路復(fù)用的機(jī)制實(shí)現(xiàn)并發(fā)。

寫一個(gè)程序每隔3秒打印“Hello World”，同時(shí)等待用戶命令行的輸入。用戶每輸入一個(gè)自然數(shù)n，就計(jì)算并打印斐波那契函數(shù)的值F(n)，之后繼續(xù)等待下一個(gè)輸入

通過最基礎(chǔ)的輪詢機(jī)制(poll)，輪詢標(biāo)準(zhǔn)輸入(stdin)是否變?yōu)榭勺x的狀態(tài)，從而當(dāng)標(biāo)準(zhǔn)輸入能被讀取時(shí)，去執(zhí)行計(jì)算Fibonacci數(shù)列。然后判斷時(shí)間是否過去三秒鐘，從而是否輸出"Hello World!".
具體代碼在example/hello_selectors_poll.py中。

注意：在Windows中并非一切都是文件，所以該實(shí)例代碼無法在Windows平臺(tái)下運(yùn)行。

import selectors
import sys
from time import time
from fib import timed_fib
def process_input(stream):
    text = stream.readline()
    n = int(text.strip())
    print("fib({}) = {}".format(n, timed_fib(n)))
def print_hello():
    print("{} - Hello world!".format(int(time())))
def main():
    selector = selectors.DefaultSelector()
    # Register the selector to poll for "read" readiness on stdin
    selector.register(sys.stdin, selectors.EVENT_READ)
    last_hello = 0  # Setting to 0 means the timer will start right away
    while True:
        # Wait at most 100 milliseconds for input to be available
        for event, mask in selector.select(0.1):
            process_input(event.fileobj)
        if time() - last_hello > 3:
            last_hello = time()
            print_hello()
if __name__ == "__main__":
    main()

從上面解決問題的設(shè)計(jì)方案演化過程，從同步到并發(fā)，從線程到I/O多路復(fù)用。可以看出根本思路去需要程序本身高效去阻塞，
讓CPU能夠執(zhí)行核心任務(wù)。意味著將數(shù)據(jù)包處理，內(nèi)存管理，處理器調(diào)度等任務(wù)從內(nèi)核態(tài)切換到應(yīng)用態(tài)，操作系統(tǒng)只處理控制層，
數(shù)據(jù)層完全交給應(yīng)用程序在應(yīng)用態(tài)中處理。極大程度的減少了程序在應(yīng)用態(tài)和內(nèi)核態(tài)之間切換的開銷，讓高性能、高并發(fā)成為了可能。

通過之前的探究，不難發(fā)現(xiàn)一個(gè)同步的程序也能通過操作系統(tǒng)的接口實(shí)現(xiàn)“并發(fā)”，而這種“并發(fā)”的行為即可稱之為異步。

之前通過I/O復(fù)用的所提供的解決方案，進(jìn)一步抽象，即可抽象出最基本的框架事件循環(huán)(Event Loop)，而其中最容易理解的實(shí)現(xiàn)，
則是回調(diào)(Callback).

通過對(duì)事件本身的抽象，以及其對(duì)應(yīng)的處理函數(shù)(handler)，可以實(shí)現(xiàn)如下算法：

維護(hù)一個(gè)按時(shí)間排序的事件列表，最近需要運(yùn)行的定時(shí)器在最前面。這樣的話每次只需要從頭檢查是否有超時(shí)的事件并執(zhí)行它們。

bisect.insort使得維護(hù)這個(gè)列表更加容易，它會(huì)幫你在合適的位置插入新的定時(shí)器事件組。
具體代碼在example/hello_event_loop_callback.py中。

注意：在Windows中并非一切都是文件，所以該實(shí)例代碼無法在Windows平臺(tái)下運(yùn)行。

from bisect import insort
from fib import timed_fib
from time import time
import selectors
import sys
class EventLoop(object):
    """
    Implements a callback based single-threaded event loop as a simple
    demonstration.
    """
    def __init__(self, *tasks):
        self._running = False
        self._stdin_handlers = []
        self._timers = []
        self._selector = selectors.DefaultSelector()
        self._selector.register(sys.stdin, selectors.EVENT_READ)
    def run_forever(self):
        self._running = True
        while self._running:
            # First check for available IO input
            for key, mask in self._selector.select(0):
                line = key.fileobj.readline().strip()
                for callback in self._stdin_handlers:
                    callback(line)
            # Handle timer events
            while self._timers and self._timers[0][0] < time():
                handler = self._timers[0][1]
                del self._timers[0]
                handler()
    def add_stdin_handler(self, callback):
        self._stdin_handlers.append(callback)
    def add_timer(self, wait_time, callback):
        insort(self._timers, (time() + wait_time, callback))
    def stop(self):
        self._running = False
def main():
    loop = EventLoop()
    def on_stdin_input(line):
        if line == "exit":
            loop.stop()
            return
        n = int(line)
        print("fib({}) = {}".format(n, timed_fib(n)))
    def print_hello():
        print("{} - Hello world!".format(int(time())))
        loop.add_timer(3, print_hello)
    def f(x):
        def g():
            print(x)
        return g
    loop.add_stdin_handler(on_stdin_input)
    loop.add_timer(0, print_hello)
    loop.run_forever()
if __name__ == "__main__":
    main()

Some thoughts on asynchronous API design in a post-async/await world

Python 開源異步并發(fā)框架的未來

Understanding Asyncio Node.js Python3.4

使用Python進(jìn)行并發(fā)編程-asyncio篇(一)

select、poll、epoll之間的區(qū)別總結(jié)[整理]