Python 网络编程：socket、HTTP 客户端与服务端开发完全指南

import socket
import threading

def handle_client(conn, addr):
    """处理客户端连接"""
    print(f"客户端 {addr} 已连接")
    with conn:
        while True:
            data = conn.recv(1024)
            if not data:
                break
            print(f"收到: {data.decode()}")
            conn.sendall(b"Echo: " + data)
    print(f"客户端 {addr} 已断开")

def start_server(host="0.0.0.0", port=8888):
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        s.bind((host, port))
        s.listen(5)
        print(f"服务器启动于 {host}:{port}")
        
        while True:
            conn, addr = s.accept()
            thread = threading.Thread(target=handle_client, args=(conn, addr))
            thread.start()

if __name__ == "__main__":
    start_server()

import socket

def send_message(message, host="127.0.0.1", port=8888):
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.connect((host, port))
        s.sendall(message.encode())
        response = s.recv(1024)
        print(f"响应: {response.decode()}")

send_message("Hello, Server!")

import selectors
import socket

sel = selectors.DefaultSelector()

def accept(sock):
    conn, addr = sock.accept()
    conn.setblocking(False)
    sel.register(conn, selectors.EVENT_READ, read)

def read(conn):
    data = conn.recv(1024)
    if data:
        conn.send(data)
    else:
        sel.unregister(conn)
        conn.close()

server = socket.socket()
server.bind(("0.0.0.0", 8888))
server.listen()
server.setblocking(False)
sel.register(server, selectors.EVENT_READ, accept)

while True:
    events = sel.select()
    for key, mask in events:
        callback = key.data
        callback(key.fileobj)

import socket

# UDP 服务端
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
    s.bind(("0.0.0.0", 9999))
    print("UDP 服务器启动")
    while True:
        data, addr = s.recvfrom(1024)
        print(f"来自 {addr}: {data.decode()}")
        s.sendto(b"Received: " + data, addr)

# UDP 客户端
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
    s.sendto(b"Hello UDP", ("127.0.0.1", 9999))
    data, _ = s.recvfrom(1024)
    print(data.decode())

import requests

# GET
resp = requests.get("https://api.github.com/users/python")
print(resp.status_code)
print(resp.json()["public_repos"])

# 带参数
resp = requests.get("https://httpbin.org/get", params={"key": "value"})

# POST JSON
resp = requests.post(
    "https://httpbin.org/post",
    json={"name": "Alice", "age": 25},
    headers={"Authorization": "Bearer token123"},
)

# 下载文件
resp = requests.get("https://example.com/large.zip", stream=True)
with open("large.zip", "wb") as f:
    for chunk in resp.iter_content(chunk_size=8192):
        f.write(chunk)

# Session（保持 cookies/连接池）
with requests.Session() as s:
    s.headers.update({"User-Agent": "MyApp/1.0"})
    s.get("https://example.com/login")

import urllib.request
import urllib.parse

# GET
with urllib.request.urlopen("https://httpbin.org/get") as resp:
    print(resp.read().decode())

# POST
data = urllib.parse.urlencode({"key": "value"}).encode()
req = urllib.request.Request("https://httpbin.org/post", data=data, method="POST")
with urllib.request.urlopen(req) as resp:
    print(resp.read().decode())

# pip install httpx
import httpx

# 同步 API（兼容 requests）
resp = httpx.get("https://api.example.com/items")

# 异步 API（原生支持 async/await）
async with httpx.AsyncClient() as client:
    resp = await client.get("https://api.example.com/items")
    print(resp.json())

# HTTP/2 支持
client = httpx.Client(http2=True)

# pip install aiohttp
import aiohttp
import asyncio

async def fetch_all(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_one(session, url) for url in urls]
        return await asyncio.gather(*tasks)

async def fetch_one(session, url):
    async with session.get(url) as resp:
        return await resp.text()

urls = ["https://httpbin.org/get"] * 5
results = asyncio.run(fetch_all(urls))
print(f"获取 {len(results)} 个页面")

# 一行命令启动文件服务器
# python -m http.server 8000

# 程序化使用
from http.server import HTTPServer, SimpleHTTPRequestHandler

server = HTTPServer(("0.0.0.0", 8000), SimpleHTTPRequestHandler)
print("Serving on port 8000")
server.serve_forever()

from http.server import HTTPServer, BaseHTTPRequestHandler
import json

class APIHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path == "/api/users":
            data = [{"id": 1, "name": "Alice"}]
            self._send_json(data)
        else:
            self.send_error(404)
    
    def do_POST(self):
        content_length = int(self.headers['Content-Length'])
        body = self.rfile.read(content_length)
        data = json.loads(body)
        
        self.send_response(201)
        self._send_json({"created": True, "data": data})
    
    def _send_json(self, data):
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(json.dumps(data).encode())

server = HTTPServer(("0.0.0.0", 8000), APIHandler)
server.serve_forever()

# pip install websockets
import asyncio
import websockets

async def handler(websocket, path):
    async for message in websocket:
        print(f"收到: {message}")
        await websocket.send(f"Echo: {message}")

async def main():
    async with websockets.serve(handler, "localhost", 8765):
        print("WebSocket server on ws://localhost:8765")
        await asyncio.Future()   # 永远运行

asyncio.run(main())

# 客户端
async def ws_client():
    async with websockets.connect("ws://localhost:8765") as ws:
        await ws.send("Hello WebSocket!")
        response = await ws.recv()
        print(response)

asyncio.run(ws_client())

# 1. 查看 DNS 解析
import socket
print(socket.gethostbyname("example.com"))
print(socket.getaddrinfo("example.com", 80))

# 2. 获取本机 IP
hostname = socket.gethostname()
print(socket.gethostbyname(hostname))

# 3. 端口是否开放
sock = socket.socket()
result = sock.connect_ex(("localhost", 5432))
print("开放" if result == 0 else "关闭")