ThreadingTCPServer实现简单的Tcp服务器

通过ThreadingTCPServer实现一个简单的Tcp服务器

# !/usr/bin/env python
# -*- encoding: utf-8 -*-
'''
@File         :tcp_server.py
@Description  :
@Time         :2022/11/17 20:01:15
@Author       :tccxy
@Version      :1.0
'''

import socketserver
import time


class TcpServer(socketserver.BaseRequestHandler):

    def handle(self) -> None:
        conn = self.request
        client_address = self.client_address
        conn.sendall(bytes("hello test ", encoding="utf-8"))
        print(client_address, '--> client has connect ')
        flag = True
        while flag:
            try:
                data = conn.recv(1024)

                data_str = ""
                for byte in data:
                    try:
                        byte_char = chr(byte)
                    except Exception:
                        byte_char = f"{byte:02x} "
                    finally:
                        data_str += byte_char
                if(len(data_str) != 0):
                    # %Y-%m-%d 
                    print(
                        f'{time.strftime("%H:%M:%S", time.localtime(time.time()))} {client_address} {data_str}')
                else:
                    print("client has exit", client_address)
                    flag = False
            except Exception as recv_ex:
                print('client has exit ', recv_ex, client_address)
        return super().handle()


if __name__ == '__main__':
    server = socketserver.ThreadingTCPServer(('127.0.0.1', 9009), TcpServer)
    try:
        server.serve_forever()
    except KeyboardInterrupt as recv_ex:
        print('sys exit ', recv_ex)
        exit()

原文地址：http://www.cnblogs.com/tccxy/p/16904026.html