socket 客户端和服务端

socket 通讯类

"""用于与 Socket 通讯"""
import socket
import logging
class SocketConnectClass:
    """Socket 连接通讯类"""
    def __init__(self, server_ip, server_port, recv_size=1024, agreement='TCP'):
        self._socket = None
        self.server_ip = server_ip
        self.server_port = server_port
        self.recv_size = recv_size
        self.agreement = agreement.upper()
        self.agreement_connect_dict = {'TCP':self.__connect_tcp, 'UDP':self.__connect_udp}
        self.agreement_send_command_dict = {'TCP':self.__send_tcp, 'UDP':self.__send_udp}

    def creat_connect(self):
        """连接 Socket"""
        logging.info('Socket 通讯协议：%s', self.agreement)
        connect_func = self.agreement_connect_dict.get(self.agreement)
        if connect_func:
            return connect_func()

    def send_command(self, command):
        """发送 Socket 命令"""
        send_command_func = self.agreement_send_command_dict.get(self.agreement)
        if send_command_func:
            return send_command_func(command)

    def __connect_tcp(self):
        """TCP 协议连接"""
        try:
            self._socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self._socket.connect((self.server_ip, self.server_port))
            return True
        except Exception:
            logging.exception('Socket 连接出错')

    def __connect_udp(self):
        """UDP 协议连接"""
        self._socket = socket.socket(socket.AF_INET,socket.SOCK_DGRAM)
        return True

    def __send_tcp(self, command):
        """TCP 发送命令"""
        try:
            self._socket.send(command)
            logging.info("Socket 发送命令为: %s",command)
            recv_data = self._socket.recv(self.recv_size)
            logging.info('Socket 接收数据为: %s', recv_data)
            return recv_data
        except Exception:
            logging.exception('Socket 发送命令出错')
    
    def __send_udp(self, command):
        """UDP 发送命令"""
        # 发送数据 字节
        try:
            self._socket.sendto(command,(self.server_ip,self.server_port))
            logging.info("Socket 发送命令为: %s",command)
            return True
        except Exception as e:
            logging.exception('Socket 发送命令出错')

    def close_socket(self):
        """关闭 Socket 连接"""
        if self._socket:
            self._socket.close()
        self._socket = None
        logging.info('Socket 连接已关闭')

socket 并发服务器

"""接受 Socket 信号"""
import logging
import socketserver

class SockeListenServer(socketserver.BaseRequestHandler):
    """Socket 服务器"""

    def handle(self):
        while True:
            try:
                recv_data = self.request.recv(1024)
                if len(recv_data) == 0:
                    logging.info('客户端断开连接')
                    break
                print("监听到 %s 信号：%s",self.client_address,recv_data.decode('utf-8'))
                logging.info("监听到 %s 信号：%s",self.client_address,recv_data.decode('utf-8'))
                self.request.send(b'OK
')
            except Exception:
                logging.exception('通讯报错，本次通讯结束')
                break
        self.request.close()

    def server_shutdown(self):
        # 关闭服务器（可选操作）
        self.server.shutdown()        

class SocketStartServerClass:
    """Socket 监听信号类"""
    def __init__(self, server_ip, server_port, socketservercls):
        self.server_ip = server_ip
        self.server_port = server_port
        self.socketservercls = socketservercls
    
    def start_server(self):
        try:
            logging.info('服务器地址：%s:%s', self.server_ip, self.server_port)
            server = socketserver.ThreadingTCPServer((self.server_ip,self.server_port),self.socketservercls)
            server.serve_forever()
        except Exception:
            logging.exception('启动 Socket 监听服务出错')

if __name__ == '__main__':
    socketstartserver = SocketStartServerClass('127.0.0.1',8000,SockeListenServer)
    socketstartserver.start_server()