创建一个TCP socket

server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

AF_INET表示IPv4，SOCK_STREAM表示TCP

这个设计真的反人类：地址重用必须放在bind之前

server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

绑定到本地地址和端口

server_socket.bind(('0.0.0.0', 8888)) # 0.0.0.0监听所有网卡
server_socket.listen(5) # 最大等待连接数

print(f"服务器启动，监听端口 8888")

while True:
client_socket, addr = server_socket.accept()
print(f"收到来自 {addr} 的连接")

data = client_socket.recv(1024)
print(f"收到数据: {data.decode('utf-8')}")

这段代码看起来简单，但我第一次运行时，bind()就报错"Address already in use"。后来发现是上一个程序没有正常关闭socket，端口被占了。setsockopt那行就是解决这个问题的——设置SO_REUSEADDR让端口可以立即重用。

另一个坑是recv()的阻塞问题。上面这个例子，如果客户端不发数据，recv()就会一直卡在那里，整个服务器都被冻住了。这在生产环境中绝对不行，所以需要把socket改成非阻塞模式：

server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server_socket.bind(('0.0.0.0', 8888))
server_socket.listen(5)

设置非阻塞

server_socket.setblocking(False)

用select来管理所有socket

sockets_list = [server_socket]

print("非阻塞服务器启动，监听端口 8888")

while True:
# select会返回可读的socket列表
read_sockets, _, exception_sockets = select.select(sockets_list, [], sockets_list)

用select的好处很明显：不会阻塞，而且可以同时处理多个连接。但是代码量翻倍了，而且select本身有性能上限——最多监控1024个socket。如果要处理更大量连接，得用epoll（Linux）或kqueue（macOS）。

还有个技巧：如果需要处理多个客户端并保持连接，更常见的方式是用多线程。我写了一个简单的多线程Echo服务器，性能从单线程的3.2秒降到0.8秒：

def handle_client(client_socket, addr):
print(f"新线程处理 {addr}")
try:
while True:
data = client_socket.recv(1024)
if not data:
break
message = data.decode('utf-8')
print(f"从 {addr} 收到: {message}")
# echo回客户端
response = f"服务器回复: {message}"
client_socket.send(response.encode('utf-8'))
except ConnectionResetError:
print(f"{addr} 强制断开连接")
finally:
client_socket.close()
print(f"{addr} 连接已关闭")

server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server_socket.bind(('0.0.0.0', 8888))
server_socket.listen(5)

print("多线程服务器启动，监听端口 8888")

多线程版本最大的坑是线程安全问题。如果多个线程同时修改共享数据（比如全局计数器），必须加锁。我踩过一次：没有加锁，结果两个客户端同时发送消息，计数器的值乱掉了。

现在来写个简易聊天机器人，结合前面学的所有技巧：

简单的回复库

RESPONSES = [
"嗯，有意思，继续说",
"这个我知道，但你先说说你的看法",
"哈哈，你太有趣了",
"别急，让我想想...",
"真的吗？我不信",
]

class ChatBot:
def __init__(self, host='0.0.0.0', port=8888):
self.server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self.server.bind((host, port))
self.server.listen(5)
self.clients = {} # 保存所有客户端

def broadcast(self, message, sender_socket=None):
"""向所有客户端广播消息"""
for client in self.clients:
if client != sender_socket:
try:
client.send(message.encode('utf-8'))
except:
pass

def handle_client(self, client_socket, addr):
self.clients[client_socket] = addr
welcome_msg = f"欢迎 {addr} 加入聊天室！"
self.broadcast(welcome_msg)

try:
while True:
data = client_socket.recv(1024)
if not data:
break

message = data.decode('utf-8')
print(f"[{addr}] {message}")

# 处理特殊命令
if message.strip() == '/help':
response = "可用命令: /help, /quit, /list"
client_socket.send(response.encode('utf-8'))
elif message.strip() == '/list':
users = [str(addr) for addr in self.clients.values()]
response = f"在线用户: {', '.join(users)}"
client_socket.send(response.encode('utf-8'))
else:
# 随机回复
reply = random.choice(RESPONSES)
client_socket.send(reply.encode('utf-8'))

except ConnectionResetError:
pass
finally:
del self.clients[client_socket]
client_socket.close()
leave_msg = f"{addr} 离开了聊天室"
self.broadcast(leave_msg)

def start(self):
print("聊天机器人启动，监听端口 8888")
while True:
client_socket, addr = self.server.accept()
thread = threading.Thread(target=self.handle_client, args=(client_socket, addr))
thread.start()

这个聊天机器人支持多客户端同时连接，有广播功能，还有简单的命令系统。但还有几个问题可以改进：比如没有用户昵称、消息没有时间戳、不能持久化聊天记录。

总结一下，你可以立刻用的三个点：