Socket（套接字）是网络通信的基础编程接口，Python 中通过 `socket` 模块可轻松实现客户端-服务器通信

Socket（套接字）是网络通信的基础编程接口，Python 中通过 socket 模块可轻松实现客户端-服务器通信。以下是一个贴近生活的 Python Socket 实例：

生活实例：家庭智能灯远程控制小程序
设想你在家用树莓派（作为服务器）连接一盏LED灯，手机或另一台电脑（作为客户端）通过Wi-Fi发送“on”/“off”指令，远程开关灯。

✅ 服务端（树莓派运行，监听指令）：

import socket

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind(('0.0.0.0', 8888)) # 监听所有网卡的8888端口
server.listen(1)
print("💡 智能灯服务器已启动，等待指令…")

while True:
conn, addr = server.accept()
print(f"收到设备 {addr} 的连接")
data = conn.recv(1024).decode().strip()
if data == "on":
print("✅ 灯已打开（实际可控制GPIO点亮LED）")
conn.send(b"Light ON")
elif data == "off":
print("❌ 灯已关闭")
conn.send(b"Light OFF")
else:
conn.send(b"Unknown command. Use 'on' or 'off'")
conn.close()

✅ 客户端（手机终端/笔记本运行，模拟APP发送指令）：

import socket

client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.connect(('192.168.1.100', 8888)) # 替换为树莓派实际IP
cmd = input("请输入指令 (on/off): ")
client.send(cmd.encode())
print("响应:", client.recv(1024).decode())
client.close()

💡 这个例子体现了 Socket 在物联网（IoT）中的典型应用：轻量、实时、低开销的设备间指令通信，无需复杂框架，适合家庭自动化、远程家电控制等场景。

要在 Python Socket 中实现多客户端同时连接（例如多个手机/设备并发控制同一盏智能灯），核心是解决“单线程阻塞”问题。默认 socket.accept() 和 conn.recv() 都是阻塞的，一次只能服务一个客户端。以下是三种主流、实用且渐进式的实现方式，从易到难，均附可运行代码示例：

✅ 方案1：多线程（最简单直观，适合初学者 & 小规模场景 ≤ 50客户端）
每个新连接分配一个独立线程处理，主线程继续监听新连接。

import socket
import threading

def handle_client(conn, addr):
print(f"🔌 新客户端接入: {addr}")
while True:
try:
data = conn.recv(1024).decode().strip()
if not data:
break
print(f"[{addr}] 指令: {data}")
# ✅ 实际控制逻辑（如调用GPIO、发MQTT等）
if data in ["on", "off"]:
response = f"✅ 灯已{data}"
# 🌟 这里可广播给其他在线客户端（见方案3扩展）
else:
response = "❌ 仅支持 'on' 或 'off'"
conn.send(response.encode())
except ConnectionResetError:
break
except OSError:
break
print(f"👋 客户端 {addr} 已断开")
conn.close()

# 主服务器
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind(('0.0.0.0', 8888))
server.listen(5)
print("🏠 智能灯服务器启动（多线程版），端口 8888…")

while True:
conn, addr = server.accept()
# 启动新线程处理该客户端
client_thread = threading.Thread(target=handle_client, args=(conn, addr), daemon=True)
client_thread.start()

⚠️ 注意：daemon=True 确保主线程退出时子线程自动结束；生产环境建议加线程池（concurrent.futures.ThreadPoolExecutor）防资源耗尽。

✅ 方案2：select + 单线程（轻量、跨平台、无GIL限制，适合中等并发）
使用 I/O 多路复用，单线程轮询所有 socket，避免创建大量线程。

import socket
import select

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind(('0.0.0.0', 8888))
server.listen(5)
server.setblocking(False) # 关键：设为非阻塞！

inputs = [server] # 待读取的socket列表
clients = {} # {conn: addr}

print("⚡ 单线程 select 版服务器启动…")

while True:
readable, _, _ = select.select(inputs, [], [], 1.0) # 超时1秒，防忙等
for sock in readable:
if sock is server:
# 新连接
conn, addr = server.accept()
conn.setblocking(False)
inputs.append(conn)
clients[conn] = addr
print(f"➕ 接入 {addr}")
else:
# 已有客户端发来数据
try:
data = sock.recv(1024)
if not data:
raise ConnectionResetError
cmd = data.decode().strip()
print(f"[{clients[sock]}] → {cmd}")
resp = "✅ 灯已" + cmd if cmd in ["on","off"] else "❌ 无效指令"
sock.send(resp.encode())
except (ConnectionResetError, ConnectionAbortedError, OSError):
print(f"➖ 客户端 {clients[sock]} 断开")
inputs.remove(sock)
del clients[sock]
sock.close()

✅ 方案3：asyncio（现代推荐，高并发、异步原生、可扩展性强）
Python 3.7+ 原生异步支持，轻松支撑数千连接，代码简洁且易维护。

import asyncio

# 全局灯状态（可替换为Redis/MQ共享状态）
light_state = "off"

async def handle_client(reader, writer):
addr = writer.get_extra_info('peername')
print(f"🌐 异步接入: {addr}")

while True:
try:
data = await reader.read(1024)
if not data:
break
cmd = data.decode().strip()
print(f"[{addr}] 指令: {cmd}")

global light_state
if cmd == "on":
light_state = "on"
response = "✅ 灯已打开"
elif cmd == "off":
light_state = "off"
response = "✅ 灯已关闭"
elif cmd == "status":
response = f"💡 当前状态: {light_state}"
else:
response = "❌ 支持指令: on / off / status"

writer.write(response.encode())
await writer.drain()
except (ConnectionResetError, OSError):
break
print(f"👋 {addr} 断开连接")
writer.close()
await writer.wait_closed()

# 启动异步服务器
async def main():
server = await asyncio.start_server(handle_client, '0.0.0.0', 8888)
print(f"🚀 asyncio 服务器运行中 — {server.sockets[0].getsockname()}")
async with server:
await server.serve_forever()

if __name__ == '__main__':
asyncio.run(main())

📌 补充建议：

• 若需广播通知所有客户端（如某手机开灯后，其他手机实时收到状态更新），可在 light_state 变更时遍历所有活跃 writer 发送消息（asyncio 方案最易实现）；
• 生产环境务必添加超时、心跳检测（如每30秒 ping/pong）、指令限频、输入校验，防DoS攻击；
• 对于真实硬件控制（如树莓派GPIO），建议将控制逻辑封装为独立函数，并加锁（threading.Lock 或 asyncio.Lock）避免并发冲突。