Python 高级编程（十二）协程

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124

import selectors
import socket
import time
import urllib.parse

selector = selectors.DefaultSelector()  # 根据不同平台自动选择最好的 io 多路复用方式

urls = ['http://www.baidu.com'] * 20
stop = False


class Fetcher:
    def __init__(self):
        self.host = None
        self.path = None
        self.client = None
        self.cur_url = None
        self.data = b''

    def get_url(self, url):
        self.cur_url = url
        url = urllib.parse.urlparse(url)
        self.host = url.netloc
        self.path = url.path if url.path != '' else '/'

        # 建立 socket 连接
        self.client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.client.setblocking(False)

        try:
            self.client.connect((self.host, 80))
        except BlockingIOError as e:
            pass

        # 把 socket 注册到 selector 上
        selector.register(self.client.fileno(), selectors.EVENT_WRITE, self.connected)  # 回调模式，当这个 socket 上可写时，调用

    def connected(self, key):
        selector.unregister(key.fd)
        self.client.send(
            'GET {} HTTP/1.1\r\nHost:{}\r\nConnection:Close\r\n\r\n'.format(self.path, self.host).encode('utf8'))
        selector.register(self.client.fileno(), selectors.EVENT_READ, self.readable)

    def readable(self, key):  # 准备好一段读一段，该函数可能有多次 EVENT_READ 多次被调用
        d = self.client.recv(1024)
        if d:
            self.data += d
        else:
            selector.unregister(key.fd)

            data = self.data.decode('utf8')
            html = data.split('\r\n\r\n')[1]
            print(html)
            self.client.close()
            urls.remove(self.cur_url)
            if not urls:
                global stop
                stop = True


def loop():
    # 事件循环，不停的请求 socket 的状态，并调用对应的回调函数
    # twisted、gevent、asyncio 本质上来讲都是这种模式：回调 + 事件循环 + select\poll\epoll
    # 1. select 本身不支持 register 模式(回调)的
    # 2. socket 状态变化以后的回调是由我们的程序完成的
    while not stop:
        ready = selector.select()
        for key, mask in ready:
            call_back = key.data
            call_back(key)


def get_url(url):
    url = urllib.parse.urlparse(url)
    host = url.netloc
    path = url.path if url.path != '' else '/'

    client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    client.setblocking(False)
    try:
        client.connect((url.netloc, 80))
    except BlockingIOError as e:
        pass

    while True:
        try:
            client.send('GET {} HTTP/1.1\r\nHost:{}\r\nConnection:Close\r\n\r\n'.format(path, host).encode('utf8'))
        except OSError as e:
            pass
        else:
            break

    data = b''
    while True:
        try:
            d = client.recv(1024)
        except BlockingIOError as e:
            continue
        if d:
            data += d
        else:
            break

    data = data.decode('utf8')
    html = data.split('\r\n\r\n')[1]
    print(html)
    client.close()


if __name__ == '__main__':
    # 异步
    start = time.time()
    for url in urls:
        fetcher = Fetcher()
        fetcher.get_url(url)
    loop()
    print('select:', time.time() - start)

    # 同步
    urls = ['http://www.baidu.com'] * 20
    start = time.time()
    for url in urls:
        get_url(url)
    print('for   :', time.time() - start)