第6-12章:数据编码、函数、类与对象、元编程、模块、网络、并发
一、导读
本章合并了 Python Cookbook 剩余章节的核心内容,涵盖数据编码与处理、函数编程、类与对象设计、元编程技巧、模块与包组织、网络与 Web 编程、并发编程。
二、核心概念详解
2.1 数据编码(struct、二进制)
python
import struct
# struct: 二进制数据打包
data = struct.pack('>ihb', 1, 2, 3) # big-endian: int, int, byte
print(len(data)) # 7 bytes
values = struct.unpack('>ihb', data)
print(values) # (1, 2, 3)
# 读取二进制文件
with open('data.bin', 'rb') as f:
header = f.read(32)
magic, version, count = struct.unpack('<4sII', header[:12])2.2 函数编程
python
from functools import partial, lru_cache
# 闭包
def make_multiplier(factor):
def multiplier(x):
return x * factor
return multiplier
double = make_multiplier(2)
print(double(5)) # 10
# partial: 固定部分参数
def power(base, exponent):
return base ** exponent
square = partial(power, exponent=2)
cube = partial(power, exponent=3)
print(square(5)) # 25
# 装饰器
def memoize(func):
cache = {}
def wrapper(*args):
if args not in cache:
cache[args] = func(*args)
return cache[args]
return wrapper
# lru_cache: 内置缓存装饰器
@lru_cache(maxsize=128)
def fibonacci(n):
if n < 2: return n
return fibonacci(n-1) + fibonacci(n-2)2.3 类与对象
python
class LazyProperty:
"""延迟计算的属性描述符"""
def __init__(self, func):
self.func = func
self.name = func.__name__
def __get__(self, instance, owner):
if instance is None:
return self
value = self.func(instance)
instance.__dict__[self.name] = value
return value
class DataProcessor:
def __init__(self, data):
self.data = data
@LazyProperty
def statistics(self):
print("Computing statistics...")
return {
'mean': sum(self.data) / len(self.data),
'min': min(self.data),
'max': max(self.data),
}
dp = DataProcessor([1, 2, 3, 4, 5])
print(dp.statistics) # 第一次: 计算
print(dp.statistics) # 第二次: 缓存2.4 模块与包
mypackage/
__init__.py
module1.py
module2.py
subpackage/
__init__.py
module3.pypython
# __init__.py: 控制包的公共接口
from .module1 import func1
from .module2 import func2
__all__ = ['func1', 'func2']
# 延迟导入
def heavy_function():
import expensive_module # 只在调用时导入
return expensive_module.compute()2.5 网络编程
python
import socket
import urllib.request
# TCP 客户端
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.connect(('example.com', 80))
s.sendall(b'GET / HTTP/1.0\r\nHost: example.com\r\n\r\n')
response = s.recv(4096)
# HTTP 请求
with urllib.request.urlopen('https://httpbin.org/get') as response:
data = response.read()
# 简单 HTTP 服务器
from http.server import HTTPServer, SimpleHTTPRequestHandler
server = HTTPServer(('localhost', 8080), SimpleHTTPRequestHandler)
server.serve_forever()2.6 并发编程
python
import threading
import multiprocessing
import asyncio
# 线程
def worker(n):
print(f"Thread {n}: working")
threads = [threading.Thread(target=worker, args=(i,)) for i in range(5)]
for t in threads: t.start()
for t in threads: t.join()
# 进程池
with multiprocessing.Pool(4) as pool:
results = pool.map(lambda x: x*x, range(10))
# asyncio
async def fetch(url):
await asyncio.sleep(1)
return f"Response from {url}"
async def main():
tasks = [fetch(f"url{i}") for i in range(5)]
results = await asyncio.gather(*tasks)
asyncio.run(main())三、本章小结
struct 处理二进制数据
闭包和装饰器是函数编程的核心
描述符控制属性访问
包结构组织大型项目
socket 实现网络通信
threading/multiprocessing/asyncio 提供多种并发模型
四、思考与练习
何时选择 threading vs multiprocessing vs asyncio?
如何设计一个线程安全的缓存?