一、网络编程基础

1.1 Socket编程

Python内置的socket模块提供了底层的网络通信能力：

python

Copy

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# 服务端示例
import socket

server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.bind(('localhost', 8000))
server_socket.listen(1)

print("等待客户端连接...")
client_socket, addr = server_socket.accept()
print(f"连接来自: {addr}")

data = client_socket.recv(1024)
print(f"接收数据: {data.decode()}")

client_socket.send("Hello, Client!".encode())
client_socket.close()

python

Copy

1
2
3
4
5
6
7
8
9
10
11
# 客户端示例
import socket

client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client_socket.connect(('localhost', 8000))

client_socket.send("Hello, Server!".encode())
data = client_socket.recv(1024)
print(f"接收数据: {data.decode()}")

client_socket.close()

1.2 多线程/多进程服务器

处理多个客户端连接的基本模式：

python

Copy

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
from threading import Thread

def handle_client(client_socket, addr):
    try:
        while True:
            data = client_socket.recv(1024)
            if not data:
                break
            print(f"来自{addr}的消息: {data.decode()}")
            client_socket.send(data)
    finally:
        client_socket.close()

server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.bind(('0.0.0.0', 8000))
server_socket.listen(5)

while True:
    client_sock, addr = server_socket.accept()
    print(f"接受来自 {addr} 的连接")
    client_handler = Thread(target=handle_client, args=(client_sock, addr))
    client_handler.start()

二、HTTP协议与Web开发

2.1 HTTP服务器

Python内置http.server模块：

python

Copy

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from http.server import HTTPServer, SimpleHTTPRequestHandler

class CustomHandler(SimpleHTTPRequestHandler):
    def do_GET(self):
        if self.path == '/hello':
            self.send_response(200)
            self.send_header('Content-type', 'text/html')
            self.end_headers()
            self.wfile.write(b"<h1>Hello, World!</h1>")
        else:
            super().do_GET()

server = HTTPServer(('localhost', 8000), CustomHandler)
print("服务器运行在 http://localhost:8000")
server.serve_forever()

2.2 主流Web框架对比

框架 特点 适用场景
Flask 微框架，灵活轻量 小型项目、API服务、快速原型
Django 全功能，自带ORM和Admin 企业级应用、内容管理系统
FastAPI 高性能，自动文档 现代API开发、微服务
Tornado 异步非阻塞 长连接应用、WebSockets

三、Flask实战

3.1 基本应用结构

python

Copy

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/')
def home():
    return "Welcome to my Flask App!"

@app.route('/api/data', methods=['GET', 'POST'])
def handle_data():
    if request.method == 'POST':
        data = request.get_json()
        return jsonify({"status": "success", "received": data})
    return jsonify({"message": "Send me some JSON!"})

if __name__ == '__main__':
    app.run(debug=True)

3.2 蓝图(Blueprints)组织大型应用

python

Copy

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# auth/views.py
from flask import Blueprint

auth_bp = Blueprint('auth', __name__)

@auth_bp.route('/login')
def login():
    return "Login Page"

# app.py
from auth.views import auth_bp

app = Flask(__name__)
app.register_blueprint(auth_bp, url_prefix='/auth')

四、异步Web开发

4.1 FastAPI示例

python

Copy

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float

@app.get("/")
async def read_root():
    return {"message": "Hello World"}

@app.post("/items/")
async def create_item(item: Item):
    return {"item_name": item.name, "item_price": item.price}

4.2 WebSocket通信

python

Copy

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# aiohttp WebSocket示例
from aiohttp import web
import asyncio

async def websocket_handler(request):
    ws = web.WebSocketResponse()
    await ws.prepare(request)
    
    async for msg in ws:
        if msg.type == web.WSMsgType.TEXT:
            print(f"收到消息: {msg.data}")
            await ws.send_str(f"你发送了: {msg.data}")
        elif msg.type == web.WSMsgType.ERROR:
            print(f"WebSocket错误: {ws.exception()}")
    
    return ws

app = web.Application()
app.add_routes([web.get('/ws', websocket_handler)])

web.run_app(app)

五、网络爬虫开发

5.1 requests + BeautifulSoup

python

Copy

1
2
3
4
5
6
7
8
9
import requests
from bs4 import BeautifulSoup

url = "https://example.com"
response = requests.get(url)
soup = BeautifulSoup(response.text, 'html.parser')

for link in soup.find_all('a'):
    print(link.get('href'))

5.2 Scrapy框架

python

Copy

1
2
3
4
5
6
7
8
9
10
11
12
13
# 定义Scrapy爬虫
import scrapy

class BlogSpider(scrapy.Spider):
    name = 'blogspider'
    start_urls = ['https://example.com/blog']

    def parse(self, response):
        for title in response.css('.post-title'):
            yield {'title': title.css('::text').get()}
        
        for next_page in response.css('a.next-page'):
            yield response.follow(next_page, self.parse)

六、网络工具与实用库

6.1 网络请求库

python

Copy

1
2
3
4
5
6
7
8
9
10
11
12
13
# requests高级用法
import requests

# 带会话保持
session = requests.Session()
session.get('https://example.com/login', auth=('user', 'pass'))

# 超时设置
response = requests.get('https://example.com', timeout=3)

# 文件上传
files = {'file': open('report.xls', 'rb')}
r = requests.post('https://example.com/upload', files=files)

6.2 网络测试工具

python

Copy

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# 端口扫描示例
import socket

def scan_port(host, port):
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(1)
        result = sock.connect_ex((host, port))
        sock.close()
        return result == 0
    except:
        return False

for port in range(80, 100):
    if scan_port('localhost', port):
        print(f"Port {port} is open")

七、安全最佳实践

7.1 常见安全防护

python

Copy

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# Flask安全中间件示例
from flask import Flask
from flask_talisman import Talisman

app = Flask(__name__)
Talisman(app, 
         force_https=True,
         strict_transport_security=True,
         session_cookie_secure=True)

# 密码哈希
from werkzeug.security import generate_password_hash, check_password_hash

password = 'secret'
hashed = generate_password_hash(password)
check_password_hash(hashed, password)  # 返回True/False

7.2 JWT认证

python

Copy

1
2
3
4
5
6
7
8
9
10
11
12
# Flask-JWT-Extended示例
from flask import Flask, jsonify
from flask_jwt_extended import JWTManager, create_access_token

app = Flask(__name__)
app.config['JWT_SECRET_KEY'] = 'super-secret'
jwt = JWTManager(app)

@app.route('/login', methods=['POST'])
def login():
    access_token = create_access_token(identity='username')
    return jsonify(access_token=access_token)

八、部署与性能优化

8.1 生产环境部署

1
2
3
4
5
# 使用Gunicorn部署Flask应用
gunicorn -w 4 -b 0.0.0.0:8000 app:app

# 使用uvicorn部署FastAPI
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

8.2 性能优化技巧

  1. 数据库连接池：

    1
    2
    3
    # SQLAlchemy连接池配置
    SQLALCHEMY_POOL_SIZE = 20
    SQLALCHEMY_MAX_OVERFLOW = 10

  2. 缓存策略：

    1
    2
    3
    # Flask-Caching示例
    from flask_caching import Cache
    cache = Cache(config={'CACHE_TYPE': 'SimpleCache'})

  3. 异步任务队列：

    1
    2
    3
    # Celery配置
    app.config['CELERY_BROKER_URL'] = 'redis://localhost:6379/0'
    app.config['CELERY_RESULT_BACKEND'] = 'redis://localhost:6379/0'

九、项目实战：简易聊天室

9.1 技术选型

  • 前端：HTML5 + JavaScript + WebSocket
  • 后端：Python + Flask + Flask-SocketIO
  • 数据库：Redis（消息暂存）

9.2 核心代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# server.py
from flask import Flask, render_template
from flask_socketio import SocketIO, emit

app = Flask(__name__)
socketio = SocketIO(app)

@app.route('/')
def index():
    return render_template('index.html')

@socketio.on('message')
def handle_message(data):
    print('收到消息:', data)
    emit('broadcast', data, broadcast=True)

if __name__ == '__main__':
    socketio.run(app, debug=True)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
<!-- templates/index.html -->
<script src="https://cdn.socket.io/4.0.1/socket.io.min.js"></script>
<script>
    const socket = io();
    socket.on('connect', () => {
        console.log('已连接到服务器');
    });
    
    socket.on('broadcast', (data) => {
        const messages = document.getElementById('messages');
        messages.innerHTML += `<div>${data}</div>`;
    });
    
    function sendMessage() {
        const input = document.getElementById('messageInput');
        socket.emit('message', input.value);
        input.value = '';
    }
</script>