重庆分公司,新征程启航

为企业提供网站建设、域名注册、服务器等服务

自定义socket实现HTTP

自定义socket实现HTTP

Web服务的本质3
之前已经带过一点了,下面使用socket发一个请求并且接收返回的数据。去掉模块的封装,从比较底层的层面了解一下其中的过程。

员工经过长期磨合与沉淀,具备了协作精神,得以通过团队的力量开发出优质的产品。创新互联坚持“专注、创新、易用”的产品理念,因为“专注所以专业、创新互联网站所以易用所以简单”。公司专注于为企业提供成都网站制作、网站建设、外贸网站建设、微信公众号开发、电商网站开发,成都微信小程序,软件按需定制网站等一站式互联网企业服务。

HTTP/1.0

用socket自定义http请求:

import socket
from bs4 import BeautifulSoup

client = socket.socket()
# 连接
client.connect(('edu.51cto.com', 80))
# 发送
header = b'GET / HTTP/1.0\r\nHost: edu.51cto.com\r\n\r\n'
client.sendall(header)
# 接收
data = client.recv(1024)
content = b''
while data:
    content += data
    data = client.recv(1024)
head, body = content.split(b'\r\n\r\n')
print(head)
print(len(body), body)
soup = BeautifulSoup(body.decode(), features='html.parser')
title = soup.find('title')
print(title)

这里发送的请求头里有HTTP的版本 "HTTP/1.0" ,所以返回的响应头里有这个 “Connection: close” ,这是一个短连接,接收数据就是上面的方式可以判断服务端是否传完了。接收数据到最后会收到一个空,就表示收完了。这个空应该是socket连接断开时发送的。

HTTP/1.1

如果发的HTTP请求版本是 “HTTP/1.1” ,返回的响应头里会有这些 “Transfer-Encoding: chunked\r\nConnection: keep-alive\r\n” 。然后在响应头和响应体之间会是这个 “\r\n\r\n2b0\r\n” ,前面的 “\r\n\r\n” 是响应头和响应体的分隔符,关键是中间的数字,这个是之后要发送的16进制字节数。也就是说这里的数据是分段发送的。每段数据都是前面是字节数,后面是数据,并且这里的分隔符也是 “\r\n” 。大概是这个样子的:

响应头
\r\n\r\n
2b0\r\n(0x2b0个字符)\r\n
27e4\r\n(0x27e4个字符)\r\n
1c7c\r\n(0x1c7c个字符)\r\n
0\r\n\r\n

像上面这样,最后是会发一个空的,所以是以 "\r\n0\r\n\r\n" 结尾。下面是自己写的实现拼接head和body的方法:

import socket
from bs4 import BeautifulSoup

client = socket.socket()
# 连接
client.connect(('edu.51cto.com', 80))
# 发送
header = b'GET / HTTP/1.1\r\nHost: edu.51cto.com\r\n\r\n'
client.sendall(header)
# 接收
data = client.recv(1024)
body = b''

# 获取请求头
while len(data.split(b'\r\n\r\n', 1)) != 2:
    data += client.recv(1024)
head, data = data.split(b'\r\n\r\n', 1)
print('HEAD', head)

# 拼接body
while data != b'0\r\n\r\n':
    while len(data.split(b'\r\n', 1)) != 2:
        data += client.recv(1024)
    l, b = data.split(b'\r\n', 1)
    length = int(l, base=16)
    if len(b) <= length:
        body += b
        length -= len(b)
        # 一下子把整段数据剩余的部分都读完
        # length 可能会很长,但是可能一次收不全,所以得用循环和计数直到收完
        while length:
            b = client.recv(length)
            body += b
            length -= len(b)
        data = b''
    else:
        body += b[:length]
        data = b[length:]

    # 把下一段body开头的 b'\r\n' 切掉
    while len(data) < 2:
        data += client.recv(1024)
    else:
        # 这个断言可以验证之前的逻辑是否有问题
        assert data[0:2] == b'\r\n', b'data error: %d %b' % (len(data), data)
        data = data[2:]

print(len(body), body)
soup = BeautifulSoup(body.decode(), features='html.parser')
title = soup.find('title')
print(title)

验证body接收是否正确,可以和上面的HTTP/1.0的结果对比一下,看一下body的长度。

自定义异步IO实现HTTP

先补充点 selector 模块的知识,再用异步的 socket 实现 HTTP 请求。

补充知识(fileno() 方法)

select 和 selectors 模块里,需要把创建的socket实例放到监听的列表里。这里,可以添加到监听列表里的可以不是原生的socket实例。这里可以是 fd 也可以是一个拥有 fileno() 方法的对象。
fd : 文件描述符,是一个整数,它是文件对象的 fileno() 方法的返回值。
这里,我们不仅要把socket对象加到监听列表里,还需要给它绑定一些别的属性。这就需要对socket封装一下。写一个自己类,加一写自己的属性以及一个socket对象的实例属性。关键是在类里实现一个 fileno() 方法,该方法原样返回 socket 实例的 fileno() 方法就可以了:

class HttpResponse(object):
    """接收一个实例化好的socket对象,在封装一些别的数据"""

    def __init__(self, sk, item):
        self.sk = sk
        self.item = item

    def fileno(self):
        """请求sockect对象的文件描述符,用于select监听"""
        return self.sk.fileno()

selector 模块

官方文档:https://docs.python.org/3/library/selectors.html

模块定义了一个 BaseSelector 的抽象基类,以及它的子类,包括:SelectSelector,PollSelector,EpollSelector,DevpollSelector,KqueueSelector。
另外还有一个DefaultSelector类,它其实是以上其中一个子类的别名而已,它自动选择为当前环境中最有效的Selector,所以平时用 DefaultSelector类就可以了,其它用不着。

# 用之前先创建实例
sel = selectors.DefaultSelector()

模块定义了两个常量,在注册事件的时候定义响应哪类事件:

  • EVENT_READ : 表示可读的; 它的值其实是1
  • EVENT_WRITE : 表示可写的; 它的值其实是2

上面两个常量是位掩码,是这样定义的:

# generic events, that must be mapped to implementation-specific ones
EVENT_READ = (1 << 0)
EVENT_WRITE = (1 << 1)

所以应该也可以同时监听两个事件, EVENT_READ+EVENT_WRITE ,也就是3。

抽象基类中的注册事件的方法
fileobj上一小节讲了,传入socket对象或者是其他实现了 fileno() 方法对象。events参数就是上面的两个常量。data参数在select方法里会返回:

register(fileobj, events, data=None)  # 注册一个文件对象
unregister(fileobj)  # 注销一个已经注册过的文件对象
modify(fileobj, events, data=None)  # 用于修改一个注册过的文件对象,比如从监听可读变为监听可写。

一个文件对象只能注册一个事件。注册的事件可以调用上面的 unregister 方法注销。
另外如果要改变文件对象监听的 event ,则调用上面的 modify 方法。它其实就是 register + unregister,但是使用modify更高效。
关于data对象,可以传入任何东西。selector本身应该不会操作data,只是事件注册时传入的data,等到该事件返回的时候,data也一起返回了。一种做法是传入回调函数,等事件返回的时候获取data调用一下做处理。还有一个做法是传入数据,等到事件返回的时候,再对对应的数据做处理。也可以以上2个都有,自己想办法吧所有的东西封装一下就是了。

抽象基类中的其他方法
select(timeout=None) :用于选择满足我们监听的event的文件对象。
这个方法如果不设置参数就是阻塞的。返回1个元组 (key, mask)
key 就是一个SelectorKey类的实例,
key.fileobj 就是注册方法的第一个参数,也就是传入的文件对象,比如socket对象。
key.data 就是注册方式的第三个参数,一般可以把回调函数传进去。
mask 就是 EVENT 事件的常量,1、2或者也可能是3。

close() : 关闭 selector。

get_key(fileobj) : 返回注册文件对象的 key,返回的是 SelectorKey 类的实例,同select方法里的key。

get_map() 方法
这个也是上面的抽象类中的方法,不过单独讲。该方法返回的所有注册的对象。返回的类型如下:

不过基本上就是个字典。可以遍历,也有 .keys()、.values()、.items() 这些方法。
字典的key,就是文件描述符fd的数字
字典的value,里面有4个属性,注册时传入的内容都在这里:

  • fileobj : 注册方法传进来的fileobj参数,就是socket对象(或者是其他实现了fileno方法的对象)
  • fd : 就是文件描述符的数字
  • events : 注册方法传进来的events参数
  • data : 注册方法传进来的data参数。

由于主要内容都在value里,所以可以遍历 sel.get_map().values() 进行操作。但是要注意遍历的时候不能操作被遍历的对象,所以不能在for循环里做注册或者注销。
可以判断是否有注册对象,还有没有注册任何对象:

if sel.get_map():  # 有注册方法
    pass

while sel.get_map():  # 可以用这个逻辑,在没有任何注册事件的时候退出循环
    pass

还可以检查注册了多少方法,控制注册事件的数量。事件太多了也就不能再注册了:

print(len(sel.get_map()))

实现异步 IO 的 HTTP

import selectors
import socket
from bs4 import BeautifulSoup

url_list = [
    {'host': 'edu.51cto.com', 'port': 80, },
    {'host': 'www.baidu.com', 'port': 80, },
    {'host': 'www.python-requests.org', 'port': 80, 'url': '/en/master/'},
    {'host': 'open-falcon.org', 'port': 80, 'url': '/'},
    {'host': 'www.jetbrains.com', 'port': 80},
]

class HttpSocket(object):
    """接收一个实例化好的socket对象,在封装一些别的数据"""

    def __init__(self, sk, item):
        self.sk = sk
        self.item = item
        self.host = self.item.get('host')
        self.port = self.item.get('port', 80)
        self.method = self.item.get('method', 'GET')
        self.url = self.item.get('url', '/')
        self.body = self.item.get('body', '')
        self.callback = self.item.get('callback')
        self.buffer = []  # 请求的返回值记录在这里

    def fileno(self):
        """请求sockect对象的文件描述符,用于select监听"""
        return self.sk.fileno()

    def create_request_header(self):
        """创建请求信息"""
        request = '%s %s HTTP/1.0\r\nHost: %s\r\n\r\n%s' % (self.method.upper(), self.url, self.host, self.body)
        return request.encode('utf-8')

    def write(self, data):
        """把接收到的数据写入 self.buffer"""
        self.buffer.append(data)

    def finish(self):
        """接收完毕后执行的函数"""
        content = b''.join(self.buffer)
        head, body = content.split(b'\r\n\r\n', 1)
        print(head)
        print(len(body), body)
        soup = BeautifulSoup(body.decode(), features='html.parser')
        title = soup.find('title')
        print(title)

class AsyncRequest(object):

    def __init__(self):
        self.sel = selectors.DefaultSelector()

    def add_request(self, item):
        """创建连接请求"""
        host = item.get('host')
        port = item.get('port')
        client = socket.socket()
        client.setblocking(False)
        try:
            client.connect((host, port))
        except BlockingIOError as e:
            pass  # 至此,已经向服务器发出连接请求了
        hsk = HttpSocket(client, item)
        self.sel.register(hsk, selectors.EVENT_WRITE, self.connect)
        # 不同同时注册2个事件,下面的注册要等到连接建立之后执行
        # self.sel.register(sk, selectors.EVENT_READ, self.accept)

    def connect(self, hsk, mask):
        """建立连接后的回调函数
        发送请求,然后注册 EVENT_READ 事件
        """
        print("连接成功:", hsk.item)
        content = hsk.create_request_header()
        print("发送请求:", content)
        hsk.sk.sendall(content)
        self.sel.modify(hsk, selectors.EVENT_READ, self.accept)

    def accept(self, hsk, mask):
        """接收请求返回的内容"""
        # print("返回信息:", hsk.item)
        data = hsk.sk.recv(1024)
        if data:
            hsk.write(data)
        else:
            print("接收完毕", hsk.item)
            hsk.finish()
            self.sel.unregister(hsk)

    def run(self):
        """主函数"""
        while self.sel.get_map():
            events = self.sel.select()
            for key, mask in events:
                callback = key.data  # key.data就是sel.register里的第三个参数
                callback(key.fileobj, mask)  # key.fileobj就是sel.register里第一次参数

if __name__ == '__main__':
    obj = AsyncRequest()
    for url_dic in url_list:
        obj.add_request(url_dic)
    obj.run()

接收完毕之后,最后执行的函数,这里是调用finish函数。这个函数最好可以自定义,那么就需要在搞一个callback参数。思路大概是这样的,最后就在finish函数里先可以做一些处理。然后判断一下,如果有callback,则调用callback。否则继续之后finish里之后的代码。
这个callback参数在哪里设置似乎在实现上都没问题:
可以在url_list里加,在HttpSocket的构造函数里提取出来。
或者是先给 AsyncRequest 类的构造函数,然后在add_request方法里实例化HttpSocket的时候再传过去。
再或者给add_request再加个参数,也是在add_request方法里实例化HttpSocket的时候再传过去。

socket 扫描端口

巩固 selector 模块的知识,又写了一个端口扫描的程序:

import socket
import selectors
import time

class ScanTask(object):
    def __init__(self, host, start, end):
        self.host = host
        self.start = start
        self.end = end
        self.port = start

    def __str__(self):
        return "%s:(%s-%s)" % (self.host, self.start, self.end)

class AsyncScanPort(object):
    def __init__(self, start=1, end=65535, timeout=1, interval=0.01, pool=100):
        self.sel = selectors.DefaultSelector()
        self.start = start
        self.end = end
        self.timeout = timeout
        self.interval = interval
        self.pool = pool
        self.scan_list = []
        self.scan_index = 0

    def add_scan_task(self, host, start=None, end=None):
        start = start or self.start
        end = end or self.end
        obj = ScanTask(host, start, end)
        self.scan_list.append(obj)

    def check_timeout(self):
        for i in self.sel.get_map().values():
            t = i.data['time']
            if time.time() > t + self.timeout:
                self.sel.unregister(i.fileobj)
                return True
            # 只要判断第一个就好了
            return False

    def add_to_scan(self):
        if len(self.sel.get_map()) >= self.pool:
            res = self.check_timeout()
            if not res:
                return False
        if self.scan_index >= len(self.scan_list):
            return False
        obj = self.scan_list[self.scan_index]
        if obj.port > obj.end:
            self.scan_index += 1
            print("\r单个地址加载完毕:", obj)
            return self.add_to_scan()
        client = socket.socket()
        client.setblocking(False)
        try:
            client.connect((obj.host, obj.port))
        except BlockingIOError as e:
            pass
        self.sel.register(client, selectors.EVENT_WRITE, {'obj': obj, 'port': obj.port, 'time': time.time()})
        print('\r正在扫描 ==> %s:%s' % (obj.host, obj.port), end='', flush=True)
        obj.port += 1
        return True

    def run(self):
        """主函数"""
        self.add_to_scan()
        while self.sel.get_map():
            events = self.sel.select(self.interval)
            if events:
                for key, mask in events:
                    self.callback(key.fileobj, mask, key.data)  # key.fileobj就是sel.register里第一次参数
            self.add_to_scan()
        print("扫描完毕")
        self.sel.close()

    def callback(self, sk, mask, data):
        try:
            sk.getpeername()
        except OSError as e:
            # 没有扫描到端口
            self.sel.unregister(sk)
            return
        self.sel.unregister(sk)
        print("\r扫描到端口:", data['obj'], data['port'])

if __name__ == '__main__':
    hosts = ['192.168.1.1', '192.168.1.2', '192.168.1.3']
    obj = AsyncScanPort()
    for h in hosts:
        obj.add_scan_task(h)
    obj.run()

文章题目:自定义socket实现HTTP
URL地址:http://cqcxhl.com/article/pohgeg.html

其他资讯

在线咨询
服务热线
服务热线:028-86922220
TOP