一、基于事件循环的非阻塞框架

代码

import socket,select

class ConnectSocketSever(object):
    def __init__(self):
        self.socket_list = []  # 用于存储所有需要连接的socket对象
        self.conn_list = []  # 用于存储所有未连接成功的socket对象

    def add_request(self, ip_port_func):
        conn = socket.socket()
        conn.setblocking(False)  # 非阻塞
        try:
            conn.connect((ip_port_func[0], ip_port_func[1]))
        except BlockingIOError as e:
            pass

        self.socket_list.append(conn)
        self.conn_list.append(conn)

    def run(self):
        """
        检测self.socket_list中的socket对象是否连接成功
        :return:
        """
        while True:
            # select.select中的4个参数和3个返回值的解释：
            #       参数1：list类型，存储着所有需要连接的对象
            #       参数2：list类型，存储着所有未连接成功的socket对象
            #       参数3：
            #       参数4：监测时长
            #       返回值r：list类型，具体是哪些socket获取到结果
            #       返回值w：list类型，具体是哪些socket连接成功
            r,w,e = select.select(self.socket_list, self.conn_list, [], 0.05)  # IO多路复用，检测socket的连接状态（是否连接成功的状态和是否获取响应结果的状态）
            for sock in w:
                sock.send(bytes.fromhex("EB9A"))  # 向socket服务端发送两个字节
                self.conn_list.remove(sock)

            for sock in r:
                data = bytes.hex(sock.recv(1024)).upper()  # socket服务端的响应结果，最多1次性接收1024字节
                print(data)
                sock.close()
                self.socket_list.remove(sock)

            if not self.socket_list:  # socket_list为空退出循环
                break

if __name__ == '__main__':
    CSSObject = ConnectSocketSever()
    urls = [
        ("127.0.0.1", 5000),
        ("127.0.0.1", 5001)
    ]
    for url in urls:
        CSSObject.add_request(url)

    CSSObject.run()

运行结果

二、基于事件循环的异步非阻塞框架

import socket,select

class ConnectSocketSever(object):
    def __init__(self):
        self.socket_list = []  # 用于存储所有需要连接的socket对象
        self.conn_list = []  # 用于存储所有未连接成功的socket对象

        self.conn_func_dict = {}

    def add_request(self, ip_port_func):
        conn = socket.socket()
        conn.setblocking(False)
        try:
            conn.connect((ip_port_func[0], ip_port_func[1]))
        except BlockingIOError as e:
            pass

        self.conn_func_dict[conn] = ip_port_func[2]  # 向conn_func_dict添加键值对，键就是conn，值就是回调函数，这样就可以将socket对象和对应的回调函数保存在字典中

        self.socket_list.append(conn)
        self.conn_list.append(conn)

    def run(self):
        """
        检测self.socket_list中的socket对象是否连接成功
        :return:
        """
        while True:
            # select.select中的4个参数和3个返回值的解释：
            #       参数1：list类型，存储着所有需要连接的对象
            #       参数2：list类型，存储着所有未连接成功的socket对象
            #       参数3：
            #       参数4：监测时长
            #       返回值r：list类型，具体是哪些socket获取到结果
            #       返回值w：list类型，具体是哪些socket连接成功
            r,w,e = select.select(self.socket_list, self.conn_list, [], 0.05)  # IO多路复用，检测socket的连接状态（是否连接成功的状态和是否获取响应结果的状态）
            for sock in w:
                sock.send(bytes.fromhex("EB9A"))  # 向socket服务端发送信息
                self.conn_list.remove(sock)

            for sock in r:
                data = bytes.hex(sock.recv(1024)).upper()  # socket服务端的响应结果
                # print(data)

                func = self.conn_func_dict[sock]  # 获取到这个sock对象对应的回调函数
                func(data)

                sock.close()
                self.socket_list.remove(sock)
            if not self.socket_list:  # socket_list为空退出循环
                break

def callbackFunction_1(data):  # 通过第1个ip和port 获取到数据后执行的回调函数
    print(data)

def callbackFunction_2(data):  # 通过第2个ip和port 获取到数据后执行的回调函数
    print(data)

if __name__ == '__main__':
    CSSObject = ConnectSocketSever()
    urls = [
        ("127.0.0.1", 5000, callbackFunction_1),
        ("127.0.0.1", 5001, callbackFunction_2)
    ]
    for url in urls:
        CSSObject.add_request(url)

    CSSObject.run()

如果想一直接收服务端的数据，可以将如下两行注释掉

测试结果：

参考视频链接：https://www.bilibili.com/video/BV1tf4y1K7Ww?p=5&spm_id_from=pageDriver&vd_source=36a3e35639c44bb339f59760641390a8