Python 网络编程

category: 编程语言

Python 网络编程全面指南

Python提供了强大的网络编程能力，从底层的socket接口到高级的HTTP客户端/服务器实现。以下是Python网络编程的详细说明。

1. 网络编程基础

OSI模型与TCP/IP协议栈

Python主要工作在传输层(TCP/UDP)和应用层(HTTP/FTP等)

基本概念

• IP地址：设备的网络标识（如192.168.1.1）
• 端口：应用程序的通信端点（0-65535）
• 协议：通信规则（TCP/UDP/HTTP等）

2. Socket编程

创建Socket

import socket

# 创建TCP socket
tcp_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# 创建UDP socket
udp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

TCP服务器示例

import socket

def tcp_server():
    # 创建TCP socket
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    
    # 绑定地址和端口
    server_address = ('localhost', 8000)
    server_socket.bind(server_address)
    
    # 监听连接
    server_socket.listen(5)
    print("TCP服务器启动，等待连接...")
    
    while True:
        # 接受连接
        client_socket, client_address = server_socket.accept()
        print(f"接收到来自 {client_address} 的连接")
        
        try:
            # 接收数据
            data = client_socket.recv(1024)
            print(f"接收到的数据: {data.decode('utf-8')}")
            
            # 发送响应
            response = "消息已收到"
            client_socket.sendall(response.encode('utf-8'))
            
        finally:
            # 关闭连接
            client_socket.close()

if __name__ == '__main__':
    tcp_server()

TCP客户端示例

import socket

def tcp_client():
    # 创建TCP socket
    client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    
    # 连接服务器
    server_address = ('localhost', 8000)
    client_socket.connect(server_address)
    
    try:
        # 发送数据
        message = "Hello, Server!"
        client_socket.sendall(message.encode('utf-8'))
        
        # 接收响应
        data = client_socket.recv(1024)
        print(f"服务器响应: {data.decode('utf-8')}")
        
    finally:
        client_socket.close()

if __name__ == '__main__':
    tcp_client()

UDP服务器示例

import socket

def udp_server():
    # 创建UDP socket
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    
    # 绑定地址和端口
    server_address = ('localhost', 9000)
    server_socket.bind(server_address)
    print("UDP服务器启动，等待消息...")
    
    while True:
        # 接收数据和客户端地址
        data, client_address = server_socket.recvfrom(4096)
        print(f"接收到来自 {client_address} 的消息: {data.decode('utf-8')}")
        
        # 发送响应
        response = "UDP消息已收到"
        server_socket.sendto(response.encode('utf-8'), client_address)

if __name__ == '__main__':
    udp_server()

UDP客户端示例

import socket

def udp_client():
    # 创建UDP socket
    client_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    
    # 服务器地址
    server_address = ('localhost', 9000)
    
    try:
        # 发送数据
        message = "Hello, UDP Server!"
        client_socket.sendto(message.encode('utf-8'), server_address)
        
        # 接收响应
        data, _ = client_socket.recvfrom(4096)
        print(f"服务器响应: {data.decode('utf-8')}")
        
    finally:
        client_socket.close()

if __name__ == '__main__':
    udp_client()

3. 多线程/多进程服务器

多线程TCP服务器

import socket
import threading

def handle_client(client_socket, address):
    print(f"处理来自 {address} 的连接")
    try:
        while True:
            data = client_socket.recv(1024)
            if not data:
                break
            print(f"来自 {address} 的数据: {data.decode('utf-8')}")
            client_socket.sendall(data)
    finally:
        client_socket.close()
    print(f"与 {address} 的连接已关闭")

def tcp_threaded_server():
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server_address = ('localhost', 8000)
    server_socket.bind(server_address)
    server_socket.listen(5)
    print("多线程TCP服务器启动...")
    
    try:
        while True:
            client_socket, client_address = server_socket.accept()
            thread = threading.Thread(
                target=handle_client,
                args=(client_socket, client_address)
            )
            thread.start()
            print(f"活跃连接数: {threading.active_count() - 1}")
    finally:
        server_socket.close()

if __name__ == '__main__':
    tcp_threaded_server()

4. 异步IO (asyncio)

asyncio TCP服务器

import asyncio

async def handle_client(reader, writer):
    addr = writer.get_extra_info('peername')
    print(f"接收到来自 {addr} 的连接")
    
    while True:
        data = await reader.read(100)
        if not data:
            break
        message = data.decode()
        print(f"收到来自 {addr} 的消息: {message}")
        
        writer.write(data)
        await writer.drain()
    
    print(f"关闭 {addr} 的连接")
    writer.close()

async def tcp_async_server():
    server = await asyncio.start_server(
        handle_client, 'localhost', 8000)
    
    addr = server.sockets[0].getsockname()
    print(f"异步TCP服务器在 {addr} 上运行")
    
    async with server:
        await server.serve_forever()

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

5. HTTP客户端

使用urllib

from urllib.request import urlopen
from urllib.parse import urlencode

# GET请求
with urlopen('https://www.example.com') as response:
    print(response.status)  # 200
    print(response.read().decode('utf-8'))  # HTML内容

# POST请求
data = urlencode({'key1': 'value1', 'key2': 'value2'}).encode()
with urlopen('https://httpbin.org/post', data=data) as response:
    print(response.read().decode('utf-8'))

使用requests库

import requests

# GET请求
response = requests.get('https://www.example.com')
print(response.status_code)  # 200
print(response.text)  # HTML内容

# POST请求
data = {'key1': 'value1', 'key2': 'value2'}
response = requests.post('https://httpbin.org/post', data=data)
print(response.json())  # JSON响应

# 带headers和参数
headers = {'User-Agent': 'MyApp/1.0'}
params = {'q': 'python'}
response = requests.get(
    'https://www.google.com/search',
    headers=headers,
    params=params
)
print(response.url)  # 查看完整URL

6. HTTP服务器

使用http.server

from http.server import HTTPServer, SimpleHTTPRequestHandler

def run_http_server():
    server_address = ('', 8000)  # 所有接口，端口8000
    httpd = HTTPServer(server_address, SimpleHTTPRequestHandler)
    print("HTTP服务器在 http://localhost:8000 运行")
    httpd.serve_forever()

if __name__ == '__main__':
    run_http_server()

自定义请求处理

from http.server import BaseHTTPRequestHandler, HTTPServer

class MyHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        self.send_response(200)
        self.send_header('Content-type', 'text/html')
        self.end_headers()
        
        message = f"<h1>路径: {self.path}</h1>"
        self.wfile.write(message.encode('utf-8'))

def run_custom_server():
    server_address = ('', 8000)
    httpd = HTTPServer(server_address, MyHandler)
    print("自定义HTTP服务器运行中...")
    httpd.serve_forever()

if __name__ == '__main__':
    run_custom_server()

7. Web框架 (Flask示例)

基本Flask应用

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/')
def home():
    return "<h1>欢迎来到Flask应用</h1>"

@app.route('/api/data', methods=['GET', 'POST'])
def handle_data():
    if request.method == 'POST':
        data = request.json
        return jsonify({"status": "success", "received": data})
    else:
        return jsonify({"message": "GET请求成功"})

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, debug=True)

8. 网络协议实现

SMTP客户端

import smtplib
from email.mime.text import MIMEText

def send_email():
    sender = '[email protected]'
    receivers = ['[email protected]']
    
    message = MIMEText('这是邮件正文内容')
    message['Subject'] = '测试邮件'
    message['From'] = sender
    message['To'] = ', '.join(receivers)
    
    try:
        with smtplib.SMTP('smtp.example.com', 587) as server:
            server.starttls()
            server.login('username', 'password')
            server.sendmail(sender, receivers, message.as_string())
        print("邮件发送成功")
    except smtplib.SMTPException as e:
        print(f"邮件发送失败: {e}")

if __name__ == '__main__':
    send_email()

FTP客户端

from ftplib import FTP

def ftp_operations():
    with FTP('ftp.example.com') as ftp:
        # 登录
        ftp.login('username', 'password')
        print(ftp.getwelcome())
        
        # 列出目录
        ftp.dir()
        
        # 下载文件
        with open('local_file.txt', 'wb') as f:
            ftp.retrbinary('RETR remote_file.txt', f.write)
        
        # 上传文件
        with open('local_file.txt', 'rb') as f:
            ftp.storbinary('STOR remote_file.txt', f)
        
        # 退出
        ftp.quit()

if __name__ == '__main__':
    ftp_operations()

9. 网络安全

SSL/TLS加密通信

import ssl
import socket

def secure_client():
    hostname = 'www.example.com'
    context = ssl.create_default_context()
    
    with socket.create_connection((hostname, 443)) as sock:
        with context.wrap_socket(sock, server_hostname=hostname) as ssock:
            print(f"SSL版本: {ssock.version()}")
            ssock.sendall(b"GET / HTTP/1.1\r\nHost: www.example.com\r\n\r\n")
            response = ssock.recv(4096)
            print(response.decode())

if __name__ == '__main__':
    secure_client()

10. 高级主题

WebSocket客户端

import websockets
import asyncio

async def websocket_client():
    uri = "ws://localhost:8765"
    async with websockets.connect(uri) as websocket:
        await websocket.send("Hello, Server!")
        response = await websocket.recv()
        print(f"收到服务器响应: {response}")

if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(websocket_client())

DNS查询

import socket

def dns_lookup():
    hostname = 'www.example.com'
    
    # 获取IP地址
    ip_address = socket.gethostbyname(hostname)
    print(f"{hostname} 的IP地址: {ip_address}")
    
    # 获取完整DNS信息
    host_info = socket.gethostbyname_ex(hostname)
    print(f"DNS信息: {host_info}")
    
    # 反向DNS查询
    try:
        hostname, aliases, addresses = socket.gethostbyaddr(ip_address)
        print(f"反向查询结果: {hostname}")
    except socket.herror:
        print("反向查询失败")

if __name__ == '__main__':
    dns_lookup()

11. 性能优化

连接池管理

from urllib3 import PoolManager

http = PoolManager(num_pools=5)

def make_request(url):
    response = http.request('GET', url)
    return response.data

if __name__ == '__main__':
    data = make_request('https://www.example.com')
    print(len(data))

异步HTTP客户端

import aiohttp
import asyncio

async def fetch_url(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

async def main():
    urls = [
        'https://www.example.com',
        'https://www.python.org',
        'https://www.google.com'
    ]
    tasks = [fetch_url(url) for url in urls]
    pages = await asyncio.gather(*tasks)
    for url, content in zip(urls, pages):
        print(f"{url} 返回 {len(content)} 字节")

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

12. 实际应用示例

网络端口扫描器

import socket
from concurrent.futures import ThreadPoolExecutor

def scan_port(host, port):
    try:
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            s.settimeout(1)
            result = s.connect_ex((host, port))
            if result == 0:
                print(f"端口 {port} 开放")
    except socket.error:
        pass

def port_scanner(host, start_port, end_port, max_threads=100):
    print(f"扫描 {host} 从端口 {start_port} 到 {end_port}")
    with ThreadPoolExecutor(max_threads) as executor:
        for port in range(start_port, end_port + 1):
            executor.submit(scan_port, host, port)

if __name__ == '__main__':
    port_scanner('localhost', 80, 100)

简易聊天服务器

import socket
import threading

clients = []

def broadcast(message, sender=None):
    for client in clients:
        if client != sender:
            try:
                client.send(message.encode('utf-8'))
            except:
                clients.remove(client)

def handle_client(client_socket):
    clients.append(client_socket)
    try:
        while True:
            message = client_socket.recv(1024).decode('utf-8')
            if not message:
                break
            print(f"收到消息: {message}")
            broadcast(message, client_socket)
    finally:
        clients.remove(client_socket)
        client_socket.close()

def chat_server():
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_socket.bind(('localhost', 9999))
    server_socket.listen(5)
    print("聊天服务器启动...")
    
    try:
        while True:
            client_socket, _ = server_socket.accept()
            thread = threading.Thread(target=handle_client, args=(client_socket,))
            thread.start()
    finally:
        server_socket.close()

if __name__ == '__main__':
    chat_server()

13. 总结

Python网络编程提供了从底层到高级的全套工具：

掌握这些工具和技术，可以开发从简单的网络工具到复杂的分布式系统的各种网络应用。