17.1 网络编程基础

17.1.1 网络协议栈

17.1.2 Socket基础

Socket通信流程：

# 服务端
import socket
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.bind(('localhost', 12345))
s.listen()
conn, addr = s.accept()  # 阻塞等待连接

# 客户端
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect(('localhost', 12345))

17.2 TCP/UDP编程

17.2.1 TCP服务端/客户端

# TCP服务端
def tcp_server():
    with socket.socket() as s:
        s.bind(('0.0.0.0', 8888))
        s.listen()
        print("Server started")
        while True:
            conn, addr = s.accept()
            with conn:
                print(f"Connected by {addr}")
                while True:
                    data = conn.recv(1024)
                    if not data:
                        break
                    conn.sendall(data.upper())

# TCP客户端
def tcp_client():
    with socket.socket() as s:
        s.connect(('localhost', 8888))
        s.sendall(b"hello world")
        data = s.recv(1024)
        print(f"Received: {data.decode()}")

17.2.2 UDP通信

# UDP服务端
def udp_server():
    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
        s.bind(('0.0.0.0', 9999))
        while True:
            data, addr = s.recvfrom(1024)
            print(f"From {addr}: {data.decode()}")
            s.sendto(data.upper(), addr)

# UDP客户端
def udp_client():
    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
        s.sendto(b"hello", ('localhost', 9999))
        data, addr = s.recvfrom(1024)
        print(f"From server: {data.decode()}")

17.3 HTTP编程

17.3.1 http.client

from http.client import HTTPConnection

conn = HTTPConnection("example.com")
conn.request("GET", "/")
resp = conn.getresponse()
print(resp.status, resp.reason)
print(resp.read().decode()[:100])

17.3.2 requests库

import requests

# GET请求
resp = requests.get("https://api.github.com/events")
print(resp.json()[0]['id'])

# POST请求
resp = requests.post("https://httpbin.org/post", 
                    data={'key': 'value'},
                    headers={'X-Test': 'true'})
print(resp.json())

17.4 WebSocket编程

17.4.1 websockets库

# 服务端
import asyncio
import websockets

async def echo(websocket):
    async for message in websocket:
        await websocket.send(f"Echo: {message}")

async def ws_server():
    async with websockets.serve(echo, "localhost", 8765):
        await asyncio.Future()  # 永久运行

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

# 启动服务端和客户端
async def main():
    server = asyncio.create_task(ws_server())
    await asyncio.sleep(1)  # 等待服务端启动
    await ws_client()
    server.cancel()

asyncio.run(main())

17.5 异步网络编程

17.5.1 asyncio TCP

import asyncio

async def handle_echo(reader, writer):
    data = await reader.read(100)
    message = data.decode()
    writer.write(data.upper())
    await writer.drain()
    writer.close()

async def async_tcp_server():
    server = await asyncio.start_server(
        handle_echo, '127.0.0.1', 8888)
    async with server:
        await server.serve_forever()

async def async_tcp_client():
    reader, writer = await asyncio.open_connection(
        '127.0.0.1', 8888)
    writer.write(b"hello world")
    data = await reader.read(100)
    print(f"Received: {data.decode()}")
    writer.close()

async def main():
    server = asyncio.create_task(async_tcp_server())
    await asyncio.sleep(1)
    await async_tcp_client()
    server.cancel()

asyncio.run(main())

17.6 应用举例

案例1：简易HTTP服务器

from http.server import HTTPServer, BaseHTTPRequestHandler
import json

class APIHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path == '/api/data':
            self.send_response(200)
            self.send_header('Content-Type', 'application/json')
            self.end_headers()
            data = {'status': 'ok', 'data': [1,2,3]}
            self.wfile.write(json.dumps(data).encode())
        else:
            self.send_error(404)

def run_server():
    server = HTTPServer(('localhost', 8000), APIHandler)
    print("Server started on http://localhost:8000")
    server.serve_forever()

if __name__ == '__main__':
    run_server()

案例2：网络监控工具

import socket
import time
from concurrent.futures import ThreadPoolExecutor

def check_port(host, port, timeout=1):
    """检查端口是否开放"""
    try:
        with socket.create_connection((host, port), timeout=timeout):
            return True
    except (socket.timeout, ConnectionRefusedError):
        return False

def network_scanner(host, ports, max_workers=50):
    """多线程端口扫描器"""
    open_ports = []
    
    def worker(port):
        if check_port(host, port):
            open_ports.append(port)
            print(f"Port {port} is open")
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        executor.map(worker, ports)
    
    return sorted(open_ports)

if __name__ == '__main__':
    host = 'example.com'
    ports = range(1, 1025)  # 扫描常用端口
    start = time.time()
    open_ports = network_scanner(host, ports)
    print(f"扫描完成，耗时 {time.time()-start:.2f}秒")
    print(f"{host} 开放端口: {open_ports}")

17.7 知识图谱

持续更新Python编程学习日志与技巧，敬请关注！

#编程# #学习# #在头条记录我的2025# #python#