Python: Python网络协议和编程

import datetime
import time
import socket
import logging

FORMAT = "%(asctime)s %(threadName)s %(thread)d %(message)s"
logging.basicConfig(format=FORMAT, level=logging.INFO)

server = socket.socket(type=socket.SOCK_DGRAM)
addr = '127.0.0.1', 9999
server.bind(addr)
print(server)

data = server.recv(1024) #收，阻塞
print(type(data), data)
data, info= server.recvfrom(1024) #阻塞 (value, (ip, port))
print(type(data), data, info)

msg = "your msg={}".format(data.decode('utf-8')).encode()
server.sendto(msg, info) #发

server.close()

#客户端连接
PS D:\py> nc -u 127.0.0.1 9999
1111
2222
your msg=2222

#服务端返回
<socket.socket fd=336, family=2, type=2, proto=0, laddr=('127.0.0.1', 9999)>
<class 'bytes'> b'1111\n'
<class 'bytes'> b'2222\n' ('127.0.0.1', 60892)

import datetime
import time
import socket
import logging

FORMAT = "%(asctime)s %(threadName)s %(thread)d %(message)s"
logging.basicConfig(format=FORMAT, level=logging.INFO)

client = socket.socket(type=socket.SOCK_DGRAM)
addr = '127.0.0.1', 9999

#绑定本地ip和端口
print(client)
client.connect(addr) #laddr 自动获取本地可以用作通信的ip和port; raddr 对端ip和端口也确定了
print(client) #返回... laddr=('127.0.0.1', 49347), raddr=('127.0.0.1', 9999)>
client.send(b'test 111') #需要提前知道对端raddr地址

client.sendto(b'test 222', addr) #laddr 自动获取本地的ip和port用来发送,发送给指定的对端ip和端口
# print(client) #返回... laddr=('0.0.0.0', 55447)>

#收数据，阻塞，只要是知道本地ip和端口的对端都能发送数据。 有不可靠性
data = client.recv(1024) # 阻塞等待数据。 在本地端口上等待数据
print(data)
data, info = client.recvfrom(1024) # 阻塞等待数据(value, (ip, port))
print(data, info)

client.close()

#客户端返回
<socket.socket fd=288, family=2, type=2, proto=0>
<socket.socket fd=288, family=2, type=2, proto=0, laddr=('127.0.0.1', 49739), raddr=('127.0.0.1', 9999)>
b'333\n'
b'444\n' ('127.0.0.1', 9999)

#服务端返回
PS D:\py> nc -ulvp 9999
listening on [any] 9999 ...
connect to [127.0.0.1] from SKY-20240801BMX [127.0.0.1] 49739
test 111test 222333
444

bind方法    #可以指定本地地址和端口laddr，会立即占用
connect方法 #可以立即占用本地地址和端口laddr，填充远程地址和端口raddr
sendto方法  #可以立即占用本地地址和端口laddr，并把数据发往指定远端。
只有有了本地绑定端口，sendto就可以向任何远端发送数据

send方法    #需要和connect方法配合， 可以使用已经从本地端口把数据发往raddr指定的远端
recv方法    #要求一定要在占用了本地端口后，返回接受的数据
recvfrom方法  #要求一定要占用了本地端口后，返回接受的数据和对端地址的二元祖

# BaseServer代码
class BaseServer:
    def __init__(self, server_address, RequestHandlerClass):
        """Constructor. May be extended, do not override."""
        self.server_address = server_address
        self.RequestHandlerClass = RequestHandlerClass
        self.__is_shut_down = threading.Event()
        self.__shutdown_request = False

    def finish_request(self, request, client_address): # 处理请求的方法
        """Finish one request by instantiating RequestHandlerClass."""
        self.RequestHandlerClass(request, client_address, self) # RequestHandlerClass构造

# BaseRequestHandler要子类覆盖的方法
class BaseRequestHandler:
    def __init__(self, request, client_address, server):
        self.request = request
        self.client_address = client_address
        self.server = server
        self.setup()
        try:
            self.handle()
        finally:
            self.finish()

    def setup(self): # 每一个连接初始化
        pass

    def handle(self): # 每一次请求处理
        pass

    def finish(self): # 每一个连接清理
        pass

from socketserver import ThreadingTCPServer, BaseRequestHandler
import threading

#socketserver模块把socket的套路封装在一起了，只需要提供自定义的RequestHandlerClass
class MyHandler(BaseRequestHandler):
    pass

server = ThreadingTCPServer(('127.0.0.1', 9999), MyHandler)
# self.RequestHandlerClass(request, client_address, self) #这3个参数是server注入的
print( server)

server.handle_request() #处理请求
print('=' * 30)
server.server_close()

#客户端。 连接瞬间断开
PS D:\> nc 127.0.0.1 9999
PS D:\>

#服务端返回
<socketserver.ThreadingTCPServer object at 0x0000026A10DF6900>
==============================

from socketserver import ThreadingTCPServer, BaseRequestHandler
import threading

#socketserver模块把socket的套路封装在一起了，只需要提供自定义的RequestHandlerClass
class MyHandler(BaseRequestHandler):
    def handle(self): #处理请求, 提供了一个处理线程、handler实例 recv
        print(self.client_address) # 客户端的
        print(id(self.server), self.server) 
        print(type(self.request), self.request) #和客户端通信的new socket对象

        print(threading.enumerate())

server = ThreadingTCPServer(('127.0.0.1', 9999), MyHandler)
# self.RequestHandlerClass(request, client_address, self) #这3个参数是server注入的 实例化
print(id(server), server)
print(server.socket)
print(server.__dict__)

# server.handle_request() #处理请求
server.serve_forever() # 循环的调用handle_request
print('=' * 30)
server.server_close()

#客户端。 连接瞬间断开
PS D:\> nc 127.0.0.1 9999
PS D:\>

#服务端返回
1872387008768 <socketserver.ThreadingTCPServer object at 0x000001B3F2F66900>
<socket.socket fd=324, family=2, type=1, proto=0, laddr=('127.0.0.1', 9999)>
{'server_address': ('127.0.0.1', 9999), 'RequestHandlerClass': <class '__main__.MyHandler'>, '_BaseServer__is_shut_down': <threading.Event at 0x1b3f31539d0: unset>, '_BaseServer__shutdown_request': False, 'socket': <socket.socket fd=324, family=2, type=1, proto=0, laddr=('127.0.0.1', 9999)>}
('127.0.0.1', 59978)
1872387008768 <socketserver.ThreadingTCPServer object at 0x000001B3F2F66900>
<class 'socket.socket'> <socket.socket fd=332, family=2, type=1, proto=0, laddr=('127.0.0.1', 9999), raddr=('127.0.0.1', 59978)>
[<_MainThread(MainThread, started 28292)>, <Thread(Thread-1 (process_request_thread), started 4512)>]

from socketserver import ThreadingTCPServer, BaseRequestHandler
import threading

#socketserver模块把socket的套路封装在一起了，只需要提供自定义的RequestHandlerClass
class MyHandler(BaseRequestHandler):
    def handle(self): #处理请求, 提供了一个处理线程、handler实例 recv
        super().handle() # 做为一个好习惯，虽然父类handle什么都没有做
        print(self.client_address) # 客户端的
        print(id(self.server), self.server) 
        print(type(self.request), self.request) #和客户端通信的new socket对象

        while True:
            data = self.request.recv(1024) #阻塞的
            print(data, '%%%%')
            msg = 'msg = {}'.format(data.decode())
            self.request.send(msg.encode())
        print(threading.enumerate())

server = ThreadingTCPServer(('127.0.0.1', 9999), MyHandler) #ThreadingTCPServer 异步; TCPServer 主线程
# self.RequestHandlerClass(request, client_address, self) #这3个参数是server注入的 实例化
print(id(server), server)
print(server.socket)
print(server.__dict__)

# server.handle_request() #处理请求
server.serve_forever() # 循环的调用handle_request
print('=' * 30)
server.server_close()

#客户端
PS D:\> nc 127.0.0.1 9999
aaaa
msg = aaaa

PS C:\Users\Administrator> nc 127.0.0.1 9999
bbbb
msg = bbbb

#服务端返回
2244418103552 <socketserver.ThreadingTCPServer object at 0x0000020A91BE6900>
<socket.socket fd=332, family=2, type=1, proto=0, laddr=('127.0.0.1', 9999)>
{'server_address': ('127.0.0.1', 9999), 'RequestHandlerClass': <class '__main__.MyHandler'>, '_BaseServer__is_shut_down': <threading.Event at 0x20a91de39d0: unset>, '_BaseServer__shutdown_request': False, 'socket': <socket.socket fd=332, family=2, type=1, proto=0, laddr=('127.0.0.1', 9999)>}
('127.0.0.1', 61660)
2244418103552 <socketserver.ThreadingTCPServer object at 0x0000020A91BE6900>
<class 'socket.socket'> <socket.socket fd=336, family=2, type=1, proto=0, laddr=('127.0.0.1', 9999), raddr=('127.0.0.1', 61660)>
b'aaaa\n' %%%%
('127.0.0.1', 61703)
2244418103552 <socketserver.ThreadingTCPServer object at 0x0000020A91BE6900>
<class 'socket.socket'> <socket.socket fd=344, family=2, type=1, proto=0, laddr=('127.0.0.1', 9999), raddr=('127.0.0.1', 61703)>
b'bbbb\n' %%%%

import datetime
import time
import socket
import logging
from socketserver import ThreadingTCPServer, StreamRequestHandler
import threading

FORMAT = "%(asctime)s %(threadName)s %(thread)d %(message)s"
logging.basicConfig(format=FORMAT, level=logging.INFO)

class EchoHandler(StreamRequestHandler):
    def handle(self):
        print(self.request)
        print(self.rfile, self.wfile) #rb rw

        print('handle ~~~~')
        while True:
            data = self.rfile.readline()
            # data = self.rfile.read1(1024)
            print(data, '********')
            msg = 'Your msg = {}'.format(data.decode()).encode()
            self.wfile.write(msg)
        print('++++++++++')

server = ThreadingTCPServer(('127.0.0.1', 9999), EchoHandler)
threading.Thread(target=server.serve_forever, name='serve').start()

while True:
    time.sleep(1)
    cmd = input('>>>')
    if cmd == 'quit':
        server.server_close()
        break
    print(threading.enumerate())

#客户端
PS D:\> nc  127.0.0.1 9999
1112333
Your msg = 1112333
lssfdfs
Your msg = lssfdfs

PS D:\> nc  127.0.0.1 9999
222
Your msg = 222

Your msg =

#服务端返回
>>><socket.socket fd=312, family=2, type=1, proto=0, laddr=('127.0.0.1', 9999), raddr=('127.0.0.1', 59637)>
<_io.BufferedReader name=312> <socketserver._SocketWriter object at 0x00000229C27B3A90>
handle ~~~~
b'1112333\n' ********
b'lssfdfs\n' ********
<socket.socket fd=328, family=2, type=1, proto=0, laddr=('127.0.0.1', 9999), raddr=('127.0.0.1', 59776)>
<_io.BufferedReader name=328> <socketserver._SocketWriter object at 0x00000229C2CF0640>
handle ~~~~
b'222\n' ********
b'\n' ********

import datetime
import time
import socket
import logging
from socketserver import ThreadingTCPServer, StreamRequestHandler
import threading

FORMAT = "%(asctime)s %(threadName)s %(thread)d %(message)s"
logging.basicConfig(format=FORMAT, level=logging.INFO)

class ChatHandler(StreamRequestHandler):
    clients = {}  #1 线程安全问题，加锁
    lock = threading.Lock()

    def setup(self):
        super().setup() #类属性
        self.event = threading.Event()
        with self.lock:
            self.clients[self.client_address] = self.wfile

    def handle(self): #rfile wfile
        super().handle()
        while not self.event.is_set():
            try:
                data = self.rfile.readline()
            except Exception as e:
                logging.error(e)
                data = b''
            # data = self.rfile.read1(1024)
            if not data or data == b'quit':
                #模块内部会自动关闭连接
                break
            msg = '{:%H%M%S} From {}:{} msg = {}'.format(
                datetime.datetime.now(),
                *self.client_address,
                data.decode()
            ).encode()

            with self.lock:
                for w in self.clients.values():
                    w.write(msg)
                    w.flush()

    def finish(self):
        super().finish()
        self.event.set()
        with self.lock:
            self.clients.pop(self.client_address)

#ChatHandler(3个参数) ，每一个请求都会产生一个hander实例
server = ThreadingTCPServer(('127.0.0.1', 9999), ChatHandler)
server.daemon_threads = True #工作线程都设置为daemon， server_forever() 是none-daemon线程。
threading.Thread(target=server.serve_forever, name='serve').start()

while True:
    time.sleep(1)
    cmd = input('>>>')
    if cmd == 'quit':
        server.server_close()
        break
    print(threading.enumerate())

#客户端
PS D:\> nc  127.0.0.1 9999
1111
231710 From 127.0.0.1:4334 msg = 1111
231717 From 127.0.0.1:4326 msg = 333

PS D:\> nc  127.0.0.1 9999
231710 From 127.0.0.1:4334 msg = 1111
333
231717 From 127.0.0.1:4326 msg = 333
PS D:\>


#服务端返回
>>>2025-03-20 23:17:27,777 Thread-1 (process_request_thread) 14504 [WinError 10054] 远程主机强迫关闭了一个现有的连接。

[<_MainThread(MainThread, started 20424)>, <Thread(serve, started 21144)>, <Thread(Thread-2 (process_request_thread), started 8792)>]
>>>

# Choose the best implementation, roughly:
# epoll|kqueue|devpoll > poll > select.
# select() also can't accept a FD > FD_SETSIZE (usually around 1024)
if 'KqueueSelector' in globals():
    DefaultSelector = KqueueSelector
elif 'EpollSelector' in globals():
    DefaultSelector = EpollSelector
elif 'DevpollSelector' in globals():
    DefaultSelector = DevpollSelector
elif 'PollSelector' in globals():
    DefaultSelector = PollSelector
else:
    DefaultSelector = SelectSelector

class SelectSelector(_BaseSelectorImpl): 
    """Select-based selector."""
    def register(fileobj, events, data=None) -> SelectorKey: pass

import time
import threading
import socket
import selectors

#TCP Echo Server
server = socket.socket()
server.bind(('127.0.0.1', 9999))
server.listen()
server.setblocking(False) #设置为非阻塞模式
print(server)

# server.accept()

selector = selectors.DefaultSelector()
key = selector.register(server, selectors.EVENT_READ, 12345) #注册待监控的文件对象或socket对象
print(type(key), key) #key.fileobj key.events, key.fd, key.data
print('-' * 30)
events = selector.select() #阻塞到监控的IO满足事件, 返回list  一批 [(key, event), (key1, 1), (key2, 2)]
print(type(events), events)

selector.close()

#客户端
PS D:\> nc  127.0.0.1 9999
PS D:\>

#服务端返回
<socket.socket fd=296, family=2, type=1, proto=0, laddr=('127.0.0.1', 9999)>
<class 'selectors.SelectorKey'> SelectorKey(fileobj=<socket.socket fd=296, family=2, type=1, proto=0, laddr=('127.0.0.1', 9999)>, fd=296, events=1, data=12345)
------------------------------
<class 'list'> [(SelectorKey(fileobj=<socket.socket fd=296, family=2, type=1, proto=0, laddr=('127.0.0.1', 9999)>, fd=296, events=1, data=12345), 1)]

import time
import threading
import socket
import selectors

#TCP Echo Server
server = socket.socket()
server.bind(('127.0.0.1', 9999))
server.listen()

# 官方建议采用非阻塞ID
server.setblocking(False) #设置为非阻塞模式
print(server)

# 回调函数，sock的读事件
# 形参自定义
# server.accept()
def accept(sock:socket.socket, mask:int):
    newsock, raddr = sock.accept() #第二阶段
    newsock.setblocking(False) #非阻塞
    print(newsock)
    print(raddr)

    key = selector.register(newsock, selectors.EVENT_READ, recv) #

# 回调函数
def recv(conn:socket.socket, mask:int):
    data = conn.recv(1024)
    print(data, '++++')
    msg = 'msg = {}'.format(data.decode())
    conn.send(msg.encode())

selector = selectors.DefaultSelector()
key = selector.register(server, selectors.EVENT_READ, accept) #注册待监控的文件对象或socket对象
print(type(key), key)
print('-' * 30)
while True: #IO 多路复用
    events = selector.select() #阻塞到监控的IO满足事件, 返回list  一批 [(key, event), (key1, 1), (key2, 2)]
    # print(type(events), events)
    for key, mask in events:
        print(key, mask) #key.fileobj:server key.events=1, key.fd=fd, key.data=accept
        key.data(key.fileobj, mask)    #调用accept(server)
    print('=' * 30)
    print(*selector.get_map().items(), sep='\n')
    print('=' * 30)

selector.close()

#客户端
PS D:\> nc  127.0.0.1 9999
12345
msg = 12345

#服务端
<socket.socket fd=316, family=2, type=1, proto=0, laddr=('127.0.0.1', 9999)>
<class 'selectors.SelectorKey'> SelectorKey(fileobj=<socket.socket fd=316, family=2, type=1, proto=0, laddr=('127.0.0.1', 9999)>, fd=316, events=1, data=<function accept at 0x00000289F038B880>)
------------------------------
SelectorKey(fileobj=<socket.socket fd=316, family=2, type=1, proto=0, laddr=('127.0.0.1', 9999)>, fd=316, events=1, data=<function accept at 0x00000289F038B880>) 1
<socket.socket fd=296, family=2, type=1, proto=0, laddr=('127.0.0.1', 9999), raddr=('127.0.0.1', 14797)>
('127.0.0.1', 14797)
==============================
(316, SelectorKey(fileobj=<socket.socket fd=316, family=2, type=1, proto=0, laddr=('127.0.0.1', 9999)>, fd=316, events=1, data=<function accept at 0x00000289F038B880>))
(296, SelectorKey(fileobj=<socket.socket fd=296, family=2, type=1, proto=0, laddr=('127.0.0.1', 9999), raddr=('127.0.0.1', 14797)>, 
fd=296, events=1, data=<function recv at 0x00000289F045A5C0>))
==============================
SelectorKey(fileobj=<socket.socket fd=296, family=2, type=1, proto=0, laddr=('127.0.0.1', 9999), raddr=('127.0.0.1', 14797)>, fd=296, events=1, data=<function recv at 0x00000289F045A5C0>) 1
b'12345\n' ++++
==============================
(316, SelectorKey(fileobj=<socket.socket fd=316, family=2, type=1, proto=0, laddr=('127.0.0.1', 9999)>, fd=316, events=1, data=<function accept at 0x00000289F038B880>))
(296, SelectorKey(fileobj=<socket.socket fd=296, family=2, type=1, proto=0, laddr=('127.0.0.1', 9999), raddr=('127.0.0.1', 14797)>, 
fd=296, events=1, data=<function recv at 0x00000289F045A5C0>))
==============================

#客户端
PS D:\tmp> nc  127.0.0.1 9999
222
msg = 222

#服务端
SelectorKey(fileobj=<socket.socket fd=316, family=2, type=1, proto=0, laddr=('127.0.0.1', 9999)>, fd=316, events=1, data=<function accept at 0x00000289F038B880>) 1
<socket.socket fd=320, family=2, type=1, proto=0, laddr=('127.0.0.1', 9999), raddr=('127.0.0.1', 19082)>
('127.0.0.1', 19082)
==============================
(316, SelectorKey(fileobj=<socket.socket fd=316, family=2, type=1, proto=0, laddr=('127.0.0.1', 9999)>, fd=316, events=1, data=<function accept at 0x00000289F038B880>))
(296, SelectorKey(fileobj=<socket.socket fd=296, family=2, type=1, proto=0, laddr=('127.0.0.1', 9999), raddr=('127.0.0.1', 14797)>, 
fd=296, events=1, data=<function recv at 0x00000289F045A5C0>))
(320, SelectorKey(fileobj=<socket.socket fd=320, family=2, type=1, proto=0, laddr=('127.0.0.1', 9999), raddr=('127.0.0.1', 19082)>, 
fd=320, events=1, data=<function recv at 0x00000289F045A5C0>))
==============================
SelectorKey(fileobj=<socket.socket fd=320, family=2, type=1, proto=0, laddr=('127.0.0.1', 9999), raddr=('127.0.0.1', 19082)>, fd=320, events=1, data=<function recv at 0x00000289F045A5C0>) 1
b'222\n' ++++
==============================
(316, SelectorKey(fileobj=<socket.socket fd=316, family=2, type=1, proto=0, laddr=('127.0.0.1', 9999)>, fd=316, events=1, data=<function accept at 0x00000289F038B880>))
(296, SelectorKey(fileobj=<socket.socket fd=296, family=2, type=1, proto=0, laddr=('127.0.0.1', 9999), raddr=('127.0.0.1', 14797)>, 
fd=296, events=1, data=<function recv at 0x00000289F045A5C0>))
(320, SelectorKey(fileobj=<socket.socket fd=320, family=2, type=1, proto=0, laddr=('127.0.0.1', 9999), raddr=('127.0.0.1', 19082)>, 
fd=320, events=1, data=<function recv at 0x00000289F045A5C0>))
==============================

import time
import socket
import selectors
import logging

FORMAT = '%(asctime)s %(threadName)s %(thread)d %(message)s'
logging.basicConfig(format=FORMAT, level=logging.INFO)

class ChatServer:
    def __init__(self, ip='127.0.0.1', port=9999):
        self.addr = ip, port
        self.server = socket.socket()

    def start(self):
        self.server.bind(self.addr)
        self.server.listen()
        self.server.setblocking(False)

        #?

    def stop(self):
        self.server.close()

import time
import socket
import selectors
import logging
import threading

FORMAT = '%(asctime)s %(threadName)s %(thread)d %(message)s'
logging.basicConfig(format=FORMAT, level=logging.INFO)

class ChatServer:
    def __init__(self, ip='127.0.0.1', port=9999):
        self.addr = ip, port
        self.server = socket.socket()
        self.selector = selectors.DefaultSelector()
        self.event = threading.Event()

    def start(self):
        self.server.bind(self.addr)
        self.server.listen()
        self.server.setblocking(False)

        self.selector.register(self.server, selectors.EVENT_READ, self.accept)

        #select
        while not self.event.is_set():
            events = self.selector.select() #阻塞到某一批IO就绪
            for key, mask in events:
                print(key, mask)
                key.data(key.fileobj, mask)

    def accept(self, sock, mask):
        print('#' * 30)
        new_client_conn, client_raddr = self.server.accept()
        print(new_client_conn)
        print(client_raddr)
        new_client_conn.setblocking(False)
        print('#' * 30)

    def stop(self):
        self.event.set()
        self.server.close()

def run():
    cs = ChatServer()
    cs.start() #阻塞了

if __name__ == '__main__':
    run()

#客户端
PS D:\> nc -p 51100  127.0.0.1 9999
PS D:\>

#服务端
SelectorKey(fileobj=<socket.socket fd=328, family=2, type=1, proto=0, laddr=('127.0.0.1', 9999)>, fd=328, events=1, data=<bound method ChatServer.accept of <__main__.ChatServer object at 0x000001BB57A59A90>>) 1
##############################
<socket.socket fd=184, family=2, type=1, proto=0, laddr=('127.0.0.1', 9999), raddr=('127.0.0.1', 51100)>
('127.0.0.1', 51100)
##############################

#客户端1
PS D:\> nc -p 51100  127.0.0.1 9999
1111
2025-03-24 10:27:45.862748 From 127.0.0.1:51100. msg=1111

#服务端
SelectorKey(fileobj=<socket.socket fd=312, family=2, type=1, proto=0, laddr=('127.0.0.1', 9999)>, fd=312, events=1, data=<bound method ChatServer.accept of <__main__.ChatServer object at 0x000001E964B99A90>>) 1
##############################
<socket.socket fd=324, family=2, type=1, proto=0, laddr=('127.0.0.1', 9999), raddr=('127.0.0.1', 51100)>
('127.0.0.1', 51100)
##############################
SelectorKey(fileobj=<socket.socket fd=324, family=2, type=1, proto=0, laddr=('127.0.0.1', 9999), raddr=('127.0.0.1', 51100)>, fd=324, events=1, data=<bound method ChatServer.recv of <__main__.ChatServer object at 0x000001E964B99A90>>) 1
b'1111\n' ++++

#客户端2  ctrl+c 主动断开
PS D:\> nc -p 51101  127.0.0.1 9999
2222
2025-03-24 10:29:26.373504 From 127.0.0.1:51101. msg=2222
PS D:\>


#服务端
SelectorKey(fileobj=<socket.socket fd=312, family=2, type=1, proto=0, laddr=('127.0.0.1', 9999)>, fd=312, events=1, data=<bound method ChatServer.accept of <__main__.ChatServer object at 0x000001E964B99A90>>) 1
##############################
<socket.socket fd=328, family=2, type=1, proto=0, laddr=('127.0.0.1', 9999), raddr=('127.0.0.1', 51101)>
('127.0.0.1', 51101)
##############################
SelectorKey(fileobj=<socket.socket fd=328, family=2, type=1, proto=0, laddr=('127.0.0.1', 9999), raddr=('127.0.0.1', 51101)>, fd=328, events=1, data=<bound method ChatServer.recv of <__main__.ChatServer object at 0x000001E964B99A90>>) 1
b'2222\n' ++++
SelectorKey(fileobj=<socket.socket fd=328, family=2, type=1, proto=0, laddr=('127.0.0.1', 9999), raddr=('127.0.0.1', 51101)>, fd=328, events=1, data=<bound method ChatServer.recv of <__main__.ChatServer object at 0x000001E964B99A90>>) 1
Traceback (most recent call last):
  File "d:\project\pyproj\trae\t2.py", line 63, in <module>
    run()
    ~~~^^
  File "d:\project\pyproj\trae\t2.py", line 60, in run
    cs.start() #阻塞了
    ~~~~~~~~^^
  File "d:\project\pyproj\trae\t2.py", line 30, in start
    key.data(key.fileobj, mask)
    ~~~~~~~~^^^^^^^^^^^^^^^^^^^
  File "d:\project\pyproj\trae\t2.py", line 44, in recv
    data = conn.recv(1024)
ConnectionResetError: [WinError 10054] 远程主机强迫关闭了一个现有的连接。

if not data or data == b'quit':
    self.selector.unregister(conn) #注销， 不再监控了，反注册
    conn.close() #?
    logging.info('{} leaving. bye...'.format(conn))
    return #结束当前线程

import time
import socket
import selectors
import logging
import threading
import datetime

FORMAT = '%(asctime)s %(threadName)s %(thread)d %(message)s'
logging.basicConfig(format=FORMAT, level=logging.INFO)

class ChatServer:
    def __init__(self, ip='127.0.0.1', port=9999):
        self.addr = ip, port
        self.server = socket.socket()
        self.selector = selectors.DefaultSelector()
        self.event = threading.Event()

    def start(self):
        self.server.bind(self.addr)
        self.server.listen()
        self.server.setblocking(False)

        self.selector.register(self.server, selectors.EVENT_READ, self.accept)

        #select
        while not self.event.is_set():
            events = self.selector.select() #阻塞到某一批IO就绪
            for key, mask in events:
                print(key, mask)
                key.data(key.fileobj, mask)

    def accept(self, sock, mask):
        print('#' * 30)
        new_client_conn, client_raddr = self.server.accept()  #第二阶段 阻塞
        # new_client_conn, client_raddr = sock.accept() #这样与上一行效果一样
        print(new_client_conn)
        print(client_raddr)
        new_client_conn.setblocking(False)
        print('#' * 30)

        key = self.selector.register(new_client_conn, selectors.EVENT_READ, self.recv)

    def recv(self, conn:socket.socket, mask):
        data = conn.recv(1024)
        print(data, '++++')
        if not data or data == b'quit':
            self.selector.unregister(conn) #注销， 不再监控了，反注册
            conn.close() #?
            logging.info('{} leaving. bye...'.format(conn))
            return #结束当前线程
        msg = "{} From {}:{}. msg={}".format(
            datetime.datetime.now(),
            *conn.getpeername(), #对端地址
            data.decode()
        ).encode()
        # conn.send(msg)
        # print(*self.selector.get_map().items(), sep='\n')
        for fd, key in self.selector.get_map().items():
            # print('-' * 30)
            # print(key.fileobj)
            # print(key.data)
            # print('-' * 30)
            # if key.data is self.recv: #绑定对象id不一样
            if key.data == self.recv: #绑定对象内容一样
                key.fileobj.send(msg) #发消息给每个客户端

    def stop(self):
        self.event.set()
        self.server.close()

def run():
    cs = ChatServer()
    cs.start() #阻塞了

if __name__ == '__main__':
    run()

import time
import logging
import threading

FORMAT = '%(asctime)s %(threadName)s %(thread)d %(message)s'
logging.basicConfig(format=FORMAT, level=logging.INFO)

def a():
    time.sleep(5)
    logging.info(threading.enumerate())
    logging.info('hello')
    time.sleep(3)

threading.Thread(target=a).start()

time.sleep(2)
raise TypeError('test')

#等2秒主线程异常结束了
#等5秒，还要等其他non-daemon线程结束
#进程的异常会导致进程状态码非0

#返回结果
# Traceback (most recent call last):
#   File "d:\project\pyproj\trae\t.py", line 17, in <module>
#     raise TypeError('test')
# TypeError: test
# 2025-03-24 11:41:45,823 Thread-1 (a) 25540 [<_MainThread(MainThread, stopped 27544)>, <Thread(Thread-1 (a), started 25540)>]
# 2025-03-24 11:41:45,823 Thread-1 (a) 25540 hello

import time
import socket
import selectors
import logging
import threading
import datetime

FORMAT = '%(asctime)s %(threadName)s %(thread)d %(message)s'
logging.basicConfig(format=FORMAT, level=logging.INFO)

class ChatServer:
    def __init__(self, ip='127.0.0.1', port=9999):
        self.addr = ip, port
        self.server = socket.socket()
        self.selector = selectors.DefaultSelector()
        self.event = threading.Event()

    def start(self):
        self.server.bind(self.addr)
        self.server.listen()
        self.server.setblocking(False)

        threading.Thread(target=self.select, name='select', daemon=False).start()

    def select(self):
        self.selector.register(self.server, selectors.EVENT_READ, self.accept)
        while not self.event.is_set():
            events = self.selector.select(0.5) #阻塞到某一批IO就绪
            for key, mask in events:
                key.data(key.fileobj, mask)

    def accept(self, sock, mask):
        new_client_conn, client_raddr = self.server.accept()  #第二阶段 阻塞
        new_client_conn.setblocking(False)
        key = self.selector.register(new_client_conn, selectors.EVENT_READ, self.recv)

    def recv(self, conn:socket.socket, mask):
        data = conn.recv(1024)
        print(data, '++++')
        if not data or data == b'quit':
            self.selector.unregister(conn) #注销， 不再监控了，反注册
            conn.close() #?
            logging.info('{} leaving. bye...'.format(conn))
            return #结束当前线程
        msg = "{} From {}:{}. msg={}".format(
            datetime.datetime.now(),
            *conn.getpeername(), #对端地址
            data.decode()
        ).encode()
        for fd, key in self.selector.get_map().items():
            if key.data == self.recv: #绑定对象内容一样
                key.fileobj.send(msg) #发消息给每个客户端

    def stop(self):
        self.event.set()
        # fileobjs = [key.fileobj for key in self.selector.get_map().values()]
        # for obj in fileobjs:
        #     self.selector.unregister(obj)
        #     obj.close()
        # self.selector.close()
        self.server.close()

def run():
    cs = ChatServer()
    cs.start() #阻塞了

    while True:
        cmd = input('>>>')
        if cmd == 'quit':
            cs.stop()
            break
        print(threading.enumerate())

if __name__ == '__main__':
    run()