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

通過了解異步設計的由來，來深入理解異步事件機制。

代碼地址

什么是異步

同步

并發(Concurrency)

線程(Thread)

I/O多路復用

異步(Asynchronous)

回調(Callback)

參考文獻

為了深入理解異步的概念，就必須先了解異步設計的由來。

顯然易見的是，同步的概念隨著我們學習第一個輸出Hello World的程序，就已經深入人心。

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

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

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

當然關于這些是怎么具體實現的就不探究了，然而隨著一個程序的功能的增加，同步設計的開發理念并不足以實現一些復雜的功能。

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

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

具體代碼在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()

對于之前那樣的問題，引入線程機制就可以解決這種簡單的并發問題。而對于線程我們應該有一個簡單的認知：

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

一個同步的程序即進程，有且只會在一個線程中運行，所以當線程被阻塞，也就意味著整個進程被阻塞

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

線程是由操作系統的調度器來調度的， 調度器統一負責管理調度進程中的線程。

系統的調度器決定什么時候會把當前線程掛起，并把CPU的控制器交個另一個線程。這個過程稱之為稱上下文切換，包括對于當前線程上下文的保存、對目標線程上下文的加載。

上下文切換會對性能產生影響，因為它本身也需要CPU的周期來執行

而隨著現實問題的復雜化，如10K問題。

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

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

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

線程并發所帶來的問題，比如因為共享的內存空間而帶來的死鎖和競態條件。這些又會導致更加復雜的代碼，在編寫代碼的時候需要時不時地注意一些線程安全的問題。

為了解決這一問題，出現了「用同一進程/線程來同時處理若干連接」的思路，也就是I/O多路復用。

以Linux操作系統為例，Linux操作系統給出了三種監聽文件描述符的機制，具體實現可參考：

select: 每個連接對應一個描述符（socket），循環處理各個連接，先查下它的狀態，ready了就進行處理，不ready就不進行處理。但是缺點很多：

每次調用select，都需要把fd集合從用戶態拷貝到內核態，這個開銷在fd很多時會很大

同時每次調用select都需要在內核遍歷傳遞進來的所有fd，這個開銷在fd很多時也很大

select支持的文件描述符數量太小了，默認是1024

poll: 本質上和select沒有區別，但是由于它是基于鏈表來存儲的，沒有最大連接數的限制。缺點是：

大量的的數組被整體復制于用戶態和內核地址空間之間，而不管這樣的復制是不是有意義。

poll的特點是「水平觸發(只要有數據可以讀，不管怎樣都會通知)」，如果報告后沒有被處理，那么下次poll時會再次報告它。

epoll: 它使用一個文件描述符管理多個描述符，將用戶關系的文件描述符的事件存放到內核的一個事件表中，這樣在用戶空間和內核空間的copy只需一次。epoll支持水平觸發和邊緣觸發，最大的特點在于「邊緣觸發」，它只告訴進程哪些剛剛變為就緒態，并且只會通知一次。使用epoll的優點很多：

沒有最大并發連接的限制，能打開的fd的上限遠大于1024（1G的內存上能監聽約10萬個端口）

效率提升，不是輪詢的方式，不會隨著fd數目的增加效率下降

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

綜上所述，通過epoll的機制，給現代高級語言提供了高并發、高性能解決方案的基礎。而同樣FreeBSD推出了kqueue，Windows推出了IOCP，Solaris推出了/dev/poll。

而在Python3.4中新增了selectors模塊，用于封裝各個操作系統所提供的I/O多路復用的接口。
那么之前同樣的問題，我們可以通過I/O多路復用的機制實現并發。

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

通過最基礎的輪詢機制(poll)，輪詢標準輸入(stdin)是否變為可讀的狀態，從而當標準輸入能被讀取時，去執行計算Fibonacci數列。然后判斷時間是否過去三秒鐘，從而是否輸出"Hello World!".
具體代碼在example/hello_selectors_poll.py中。

注意：在Windows中并非一切都是文件，所以該實例代碼無法在Windows平臺下運行。

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()

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

通過之前的探究，不難發現一個同步的程序也能通過操作系統的接口實現“并發”，而這種“并發”的行為即可稱之為異步。

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

通過對事件本身的抽象，以及其對應的處理函數(handler)，可以實現如下算法：

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

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

注意：在Windows中并非一切都是文件，所以該實例代碼無法在Windows平臺下運行。

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 開源異步并發框架的未來

Understanding Asyncio Node.js Python3.4

使用Python進行并發編程-asyncio篇(一)

select、poll、epoll之間的區別總結[整理]