面试题手册

NestJS GraphQL 集成详解GraphQL 概述GraphQL 是一种用于 API 的查询语言和运行时环境。NestJS 通过 @nestjs/graphql 包提供了完整的 GraphQL 支持，使开发者能够构建灵活、高效的 GraphQL API。安装依赖npm install @nestjs/graphql graphql apollo-server-expressnpm install -D @types/graphql基本配置配置 GraphQL 模块import { Module } from '@nestjs/common';import { GraphQLModule } from '@nestjs/graphql';import { ApolloDriver, ApolloDriverConfig } from '@nestjs/apollo';import { join } from 'path';@Module({ imports: [ GraphQLModule.forRoot<ApolloDriverConfig>({ driver: ApolloDriver, autoSchemaFile: join(process.cwd(), 'src/schema.gql'), sortSchema: true, playground: true, introspection: true, context: ({ req }) => ({ req }), }), ],})export class AppModule {}定义 Schema使用装饰器定义 Schemaimport { Field, Int, ObjectType, Resolver, Query, Mutation, Args, ID } from '@nestjs/graphql';import { User } from './entities/user.entity';@ObjectType()export class User { @Field(() => ID) id: number; @Field() name: string; @Field() email: string; @Field(() => Int) age: number; @Field() isActive: boolean; @Field() createdAt: Date; @Field() updatedAt: Date;}定义输入类型import { InputType, Field } from '@nestjs/graphql';@InputType()export class CreateUserInput { @Field() name: string; @Field() email: string; @Field() password: string; @Field(() => Int, { nullable: true }) age?: number;}@InputType()export class UpdateUserInput { @Field(() => ID) id: number; @Field({ nullable: true }) name?: string; @Field({ nullable: true }) email?: string; @Field(() => Int, { nullable: true }) age?: number;}创建 Resolver基本 Resolverimport { Resolver, Query, Mutation, Args, ID } from '@nestjs/graphql';import { UsersService } from './users.service';import { User } from './entities/user.entity';import { CreateUserInput, UpdateUserInput } from './dto/user.input';@Resolver(() => User)export class UsersResolver { constructor(private usersService: UsersService) {} @Query(() => [User]) async users(): Promise<User[]> { return this.usersService.findAll(); } @Query(() => User, { nullable: true }) async user(@Args('id', { type: () => ID }) id: number): Promise<User> { return this.usersService.findOne(id); } @Mutation(() => User) async createUser(@Args('input') input: CreateUserInput): Promise<User> { return this.usersService.create(input); } @Mutation(() => User, { nullable: true }) async updateUser(@Args('input') input: UpdateUserInput): Promise<User> { return this.usersService.update(input.id, input); } @Mutation(() => Boolean) async deleteUser(@Args('id', { type: () => ID }) id: number): Promise<boolean> { return this.usersService.remove(id); }}使用自定义装饰器import { createParamDecorator, ExecutionContext } from '@nestjs/common';export const CurrentUser = createParamDecorator( (data: unknown, ctx: ExecutionContext) => { const request = ctx.switchToHttp().getRequest(); return request.user; },);// 在 Resolver 中使用@Mutation(() => User)async createUser( @Args('input') input: CreateUserInput, @CurrentUser() user: any,): Promise<User> { return this.usersService.create(input, user.id);}关系和加载一对一关系@ObjectType()export class Profile { @Field(() => ID) id: number; @Field() bio: string; @Field(() => User) user: User;}@ObjectType()export class User { @Field(() => ID) id: number; @Field() name: string; @Field(() => Profile, { nullable: true }) profile?: Profile;}一对多关系@ObjectType()export class Post { @Field(() => ID) id: number; @Field() title: string; @Field() content: string; @Field(() => User) author: User;}@ObjectType()export class User { @Field(() => ID) id: number; @Field() name: string; @Field(() => [Post]) posts: Post[];}多对多关系@ObjectType()export class Tag { @Field(() => ID) id: number; @Field() name: string; @Field(() => [Post]) posts: Post[];}@ObjectType()export class Post { @Field(() => ID) id: number; @Field() title: string; @Field(() => [Tag]) tags: Tag[];}分页基于游标的分页import { ArgsType, Field, Int } from '@nestjs/graphql';@ArgsType()export class PaginationArgs { @Field(() => Int, { nullable: true }) first?: number; @Field(() => String, { nullable: true }) after?: string;}@ObjectType()export class PaginatedUsers { @Field(() => [User]) data: User[]; @Field(() => Boolean) hasNextPage: boolean; @Field(() => String, { nullable: true }) endCursor?: string;}@Resolver(() => User)export class UsersResolver { @Query(() => PaginatedUsers) async users(@Args() pagination: PaginationArgs): Promise<PaginatedUsers> { const { data, hasNextPage, endCursor } = await this.usersService.paginate(pagination); return { data, hasNextPage, endCursor }; }}基于偏移量的分页@ArgsType()export class OffsetPaginationArgs { @Field(() => Int, { nullable: true, defaultValue: 0 }) skip?: number; @Field(() => Int, { nullable: true, defaultValue: 10 }) take?: number;}@ObjectType()export class OffsetPaginatedUsers { @Field(() => [User]) data: User[]; @Field(() => Int) total: number; @Field(() => Int) page: number; @Field(() => Int) totalPages: number;}@Resolver(() => User)export class UsersResolver { @Query(() => OffsetPaginatedUsers) async users(@Args() pagination: OffsetPaginationArgs): Promise<OffsetPaginatedUsers> { const { data, total, page, totalPages } = await this.usersService.paginate(pagination); return { data, total, page, totalPages }; }}订阅（Subscriptions）配置订阅import { PubSub } from 'graphql-subscriptions';@Module({ imports: [ GraphQLModule.forRoot<ApolloDriverConfig>({ driver: ApolloDriver, installSubscriptionHandlers: true, context: ({ req, connection }) => { if (connection) { return { req: connection.context }; } return { req }; }, }), ],})export class AppModule {}创建订阅 Resolverimport { Resolver, Subscription, Root } from '@nestjs/graphql';import { PubSub } from 'graphql-subscriptions';@ObjectType()export class Message { @Field(() => ID) id: number; @Field() content: string; @Field() createdAt: Date;}@Resolver(() => Message)export class MessagesResolver { private pubSub: PubSub; constructor() { this.pubSub = new PubSub(); } @Subscription(() => Message, { filter: (payload, variables) => { return payload.chatId === variables.chatId; }, }) messageAdded(@Root() message: Message, @Args('chatId') chatId: number): Message { return message; } async publishMessage(message: Message) { await this.pubSub.publish('messageAdded', { messageAdded: message }); }}数据加载器（DataLoader）创建 DataLoaderimport { DataLoader } from 'dataloader';import { UsersService } from './users.service';export class UserDataLoader { private loader: DataLoader<number, User>; constructor(private usersService: UsersService) { this.loader = new DataLoader(async (ids) => { const users = await this.usersService.findByIds(ids); return ids.map(id => users.find(user => user.id === id)); }); } async load(id: number): Promise<User> { return this.loader.load(id); } async loadMany(ids: number[]): Promise<User[]> { return this.loader.loadMany(ids); }}在 Resolver 中使用 DataLoaderimport { Resolver, Query, Args, ID, Parent } from '@nestjs/graphql';import { UserDataLoader } from './user-dataloader';@Resolver(() => Post)export class PostsResolver { constructor(private userDataLoader: UserDataLoader) {} @Query(() => [Post]) async posts(): Promise<Post[]> { return this.postsService.findAll(); } @FieldResolver(() => User) async author(@Parent() post: Post): Promise<User> { return this.userDataLoader.load(post.authorId); }}验证和转换使用 class-validatorimport { IsEmail, IsString, MinLength } from 'class-validator';@InputType()export class CreateUserInput { @Field() @IsEmail() email: string; @Field() @IsString() @MinLength(6) password: string; @Field() @IsString() name: string;}自定义验证器import { registerDecorator, ValidationOptions, ValidationArguments } from 'class-validator';export function IsStrongPassword(validationOptions?: ValidationOptions) { return function (object: Object, propertyName: string) { registerDecorator({ name: 'isStrongPassword', target: object.constructor, propertyName: propertyName, options: validationOptions, validator: { validate(value: any) { const hasUpperCase = /[A-Z]/.test(value); const hasLowerCase = /[a-z]/.test(value); const hasNumber = /[0-9]/.test(value); return hasUpperCase && hasLowerCase && hasNumber; }, defaultMessage(args: ValidationArguments) { return 'Password must contain uppercase, lowercase, and numbers'; }, }, }); };}@InputType()export class CreateUserInput { @Field() @IsStrongPassword() password: string;}错误处理自定义错误类import { ApolloError } from 'apollo-server-express';export class UserNotFoundError extends ApolloError { constructor(id: number) { super(`User with ID ${id} not found`, 'USER_NOT_FOUND', { id, }); }}export class ValidationError extends ApolloError { constructor(message: string, fields: string[]) { super(message, 'VALIDATION_ERROR', { fields }); }}在 Resolver 中使用错误@Resolver(() => User)export class UsersResolver { constructor(private usersService: UsersService) {} @Query(() => User, { nullable: true }) async user(@Args('id', { type: () => ID }) id: number): Promise<User> { const user = await this.usersService.findOne(id); if (!user) { throw new UserNotFoundError(id); } return user; } @Mutation(() => User) async createUser(@Args('input') input: CreateUserInput): Promise<User> { try { return await this.usersService.create(input); } catch (error) { throw new ValidationError('Invalid input', error.fields); } }}权限控制创建权限守卫import { CanActivate, ExecutionContext, Injectable, UnauthorizedException } from '@nestjs/common';import { GqlExecutionContext } from '@nestjs/graphql';@Injectable()export class GqlAuthGuard implements CanActivate { canActivate(context: ExecutionContext): boolean { const ctx = GqlExecutionContext.create(context); const { req } = ctx.getContext(); if (!req.user) { throw new UnauthorizedException(); } return true; }}使用权限守卫import { Resolver, Mutation, UseGuards } from '@nestjs/graphql';import { GqlAuthGuard } from './guards/gql-auth.guard';@Resolver(() => User)export class UsersResolver { @Mutation(() => User) @UseGuards(GqlAuthGuard) async createUser(@Args('input') input: CreateUserInput): Promise<User> { return this.usersService.create(input); }}性能优化查询复杂度分析import { GraphQLModule } from '@nestjs/graphql';import { ApolloDriver, ApolloDriverConfig } from '@nestjs/apollo';GraphQLModule.forRoot<ApolloDriverConfig>({ driver: ApolloDriver, validationRules: [ queryComplexity({ maximumComplexity: 1000, variables: {}, onComplete: (complexity) => { console.log(`Query complexity: ${complexity}`); }, }), ],})查询深度限制import { depthLimit } from 'graphql-depth-limit';GraphQLModule.forRoot<ApolloDriverConfig>({ driver: ApolloDriver, validationRules: [ depthLimit(5), ],})最佳实践Schema 优先：优先使用 Schema 优先的方法类型安全：充分利用 TypeScript 的类型系统分页：始终实现分页以避免大量数据传输数据加载器：使用 DataLoader 解决 N+1 查询问题错误处理：提供清晰的错误信息权限控制：实现适当的权限控制机制性能监控：监控查询性能和复杂度文档化：使用 GraphQL Playground 和文档工具总结NestJS GraphQL 集成提供了：完整的 GraphQL 支持灵活的 Schema 定义强大的类型安全丰富的查询功能易于集成的生态系统掌握 NestJS GraphQL 集成是构建现代、灵活 API 的关键。通过合理使用 GraphQL 的特性，可以构建出高效、类型安全、易于维护的 API，满足前端对数据的精确需求。GraphQL 的查询语言特性使客户端能够精确获取所需数据，减少网络传输和提升性能。

​

Python 函数式编程详解函数式编程的基本概念函数式编程是一种编程范式，强调使用纯函数、避免可变状态和副作用。Python 虽然不是纯函数式语言，但提供了丰富的函数式编程工具。纯函数纯函数是指相同的输入总是产生相同的输出，并且没有任何副作用。# 纯函数示例def add(a, b): return a + bprint(add(2, 3)) # 5print(add(2, 3)) # 5 - 相同输入，相同输出# 非纯函数示例counter = 0def increment(): global counter counter += 1 return counterprint(increment()) # 1print(increment()) # 2 - 相同输入，不同输出（有副作用）不可变数据函数式编程倾向于使用不可变数据结构。# 不可变操作original_list = [1, 2, 3]new_list = original_list + [4, 5] # 创建新列表，不修改原列表print(original_list) # [1, 2, 3]print(new_list) # [1, 2, 3, 4, 5]# 可变操作（不推荐）original_list.append(4) # 修改原列表print(original_list) # [1, 2, 3, 4]高阶函数高阶函数是指接受函数作为参数或返回函数的函数。map 函数map 函数对可迭代对象的每个元素应用指定函数。# 基本用法numbers = [1, 2, 3, 4, 5]squared = list(map(lambda x: x ** 2, numbers))print(squared) # [1, 4, 9, 16, 25]# 使用命名函数def square(x): return x ** 2squared = list(map(square, numbers))print(squared) # [1, 4, 9, 16, 25]# 多个可迭代对象numbers1 = [1, 2, 3]numbers2 = [4, 5, 6]summed = list(map(lambda x, y: x + y, numbers1, numbers2))print(summed) # [5, 7, 9]filter 函数filter 函数根据条件过滤可迭代对象的元素。# 基本用法numbers = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]even_numbers = list(filter(lambda x: x % 2 == 0, numbers))print(even_numbers) # [2, 4, 6, 8, 10]# 使用命名函数def is_even(x): return x % 2 == 0even_numbers = list(filter(is_even, numbers))print(even_numbers) # [2, 4, 6, 8, 10]# 过滤字符串words = ["apple", "banana", "cherry", "date"]long_words = list(filter(lambda x: len(x) > 5, words))print(long_words) # ['banana', 'cherry']reduce 函数reduce 函数对可迭代对象的元素进行累积操作。from functools import reduce# 基本用法numbers = [1, 2, 3, 4, 5]sum_result = reduce(lambda x, y: x + y, numbers)print(sum_result) # 15# 计算乘积product = reduce(lambda x, y: x * y, numbers)print(product) # 120# 使用初始值sum_with_initial = reduce(lambda x, y: x + y, numbers, 10)print(sum_with_initial) # 25# 查找最大值max_value = reduce(lambda x, y: x if x > y else y, numbers)print(max_value) # 5sorted 函数sorted 函数对可迭代对象进行排序。# 基本排序numbers = [3, 1, 4, 1, 5, 9, 2, 6]sorted_numbers = sorted(numbers)print(sorted_numbers) # [1, 1, 2, 3, 4, 5, 6, 9]# 降序排序sorted_desc = sorted(numbers, reverse=True)print(sorted_desc) # [9, 6, 5, 4, 3, 2, 1, 1]# 按键排序students = [ {"name": "Alice", "age": 25}, {"name": "Bob", "age": 20}, {"name": "Charlie", "age": 30}]sorted_by_age = sorted(students, key=lambda x: x["age"])print(sorted_by_age)# [{'name': 'Bob', 'age': 20}, {'name': 'Alice', 'age': 25}, {'name': 'Charlie', 'age': 30}]Lambda 表达式Lambda 表达式是匿名函数，适用于简单的函数定义。基本语法# Lambda 表达式add = lambda x, y: x + yprint(add(3, 5)) # 8# 等价于def add(x, y): return x + y实际应用# 与高阶函数结合使用numbers = [1, 2, 3, 4, 5]squared = list(map(lambda x: x ** 2, numbers))print(squared) # [1, 4, 9, 16, 25]# 排序students = [("Alice", 25), ("Bob", 20), ("Charlie", 30)]sorted_students = sorted(students, key=lambda x: x[1])print(sorted_students) # [('Bob', 20), ('Alice', 25), ('Charlie', 30)]# 条件表达式get_grade = lambda score: "A" if score >= 90 else "B" if score >= 80 else "C"print(get_grade(95)) # Aprint(get_grade(85)) # Bprint(get_grade(75)) # CLambda 的限制# Lambda 只能包含表达式，不能包含语句# 错误示例# bad_lambda = lambda x: if x > 0: return x # 语法错误# 正确做法good_lambda = lambda x: x if x > 0 else 0print(good_lambda(5)) # 5print(good_lambda(-5)) # 0装饰器装饰器是高阶函数的一种应用，用于修改或增强函数的行为。基本装饰器def my_decorator(func): def wrapper(): print("函数执行前") func() print("函数执行后") return wrapper@my_decoratordef say_hello(): print("Hello!")say_hello()# 输出:# 函数执行前# Hello!# 函数执行后带参数的装饰器def repeat(times): def decorator(func): def wrapper(*args, **kwargs): for _ in range(times): result = func(*args, **kwargs) return result return wrapper return decorator@repeat(3)def greet(name): print(f"Hello, {name}!")greet("Alice")# 输出:# Hello, Alice!# Hello, Alice!# Hello, Alice!保留函数元数据from functools import wrapsdef logging_decorator(func): @wraps(func) def wrapper(*args, **kwargs): print(f"调用函数: {func.__name__}") return func(*args, **kwargs) return wrapper@logging_decoratordef calculate(x, y): """计算两个数的和""" return x + yprint(calculate.__name__) # calculateprint(calculate.__doc__) # 计算两个数的和偏函数偏函数固定函数的某些参数，创建新的函数。from functools import partial# 基本用法def power(base, exponent): return base ** exponentsquare = partial(power, exponent=2)cube = partial(power, exponent=3)print(square(5)) # 25print(cube(5)) # 125# 实际应用def greet(name, greeting, punctuation): return f"{greeting}, {name}{punctuation}"hello = partial(greet, greeting="Hello", punctuation="!")goodbye = partial(greet, greeting="Goodbye", punctuation=".")print(hello("Alice")) # Hello, Alice!print(goodbye("Bob")) # Goodbye, Bob.列表推导式与生成器表达式列表推导式# 基本用法numbers = [1, 2, 3, 4, 5]squared = [x ** 2 for x in numbers]print(squared) # [1, 4, 9, 16, 25]# 带条件even_squared = [x ** 2 for x in numbers if x % 2 == 0]print(even_squared) # [4, 16]# 嵌套matrix = [[1, 2, 3], [4, 5, 6], [7, 8, 9]]flattened = [item for row in matrix for item in row]print(flattened) # [1, 2, 3, 4, 5, 6, 7, 8, 9]生成器表达式# 基本用法numbers = (x ** 2 for x in range(10))print(list(numbers)) # [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]# 内存效率# 列表推导式 - 占用大量内存large_list = [x ** 2 for x in range(1000000)]# 生成器表达式 - 几乎不占用内存large_gen = (x ** 2 for x in range(1000000))实际应用场景1. 数据处理管道from functools import reduce# 处理数据管道data = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]# 过滤偶数even = filter(lambda x: x % 2 == 0, data)# 平方squared = map(lambda x: x ** 2, even)# 求和result = reduce(lambda x, y: x + y, squared)print(result) # 2202. 函数组合def compose(*functions): """组合多个函数""" def inner(arg): result = arg for func in reversed(functions): result = func(result) return result return inner# 定义函数def add_one(x): return x + 1def multiply_two(x): return x * 2def square(x): return x ** 2# 组合函数pipeline = compose(square, multiply_two, add_one)print(pipeline(3)) # ((3 + 1) * 2) ** 2 = 643. 柯里化def curry(func): """柯里化函数""" def curried(*args): if len(args) >= func.__code__.co_argcount: return func(*args) return lambda *more_args: curried(*(args + more_args)) return curried@currydef add(a, b, c): return a + b + cadd_1 = add(1)add_1_2 = add_1(2)result = add_1_2(3)print(result) # 6# 也可以链式调用result = add(1)(2)(3)print(result) # 64. 记忆化from functools import lru_cache# 使用 lru_cache 装饰器@lru_cache(maxsize=128)def fibonacci(n): if n < 2: return n return fibonacci(n-1) + fibonacci(n-2)print(fibonacci(100)) # 快速计算# 手动实现记忆化def memoize(func): cache = {} def wrapper(*args): if args not in cache: cache[args] = func(*args) return cache[args] return wrapper@memoizedef fibonacci_manual(n): if n < 2: return n return fibonacci_manual(n-1) + fibonacci_manual(n-2)print(fibonacci_manual(100)) # 快速计算函数式编程的优势1. 可预测性# 纯函数的行为是可预测的def calculate_discount(price, discount_rate): return price * (1 - discount_rate)print(calculate_discount(100, 0.2)) # 80.0print(calculate_discount(100, 0.2)) # 80.0 - 总是相同2. 可测试性# 纯函数易于测试def add(a, b): return a + b# 测试assert add(2, 3) == 5assert add(-1, 1) == 0assert add(0, 0) == 03. 并行性# 纯函数可以安全地并行执行from concurrent.futures import ThreadPoolExecutordef process_item(item): return item ** 2items = list(range(1000))with ThreadPoolExecutor(max_workers=4) as executor: results = list(executor.map(process_item, items))4. 代码简洁性# 函数式风格更简洁numbers = [1, 2, 3, 4, 5]# 命令式风格squared = []for num in numbers: squared.append(num ** 2)# 函数式风格squared = list(map(lambda x: x ** 2, numbers))最佳实践1. 优先使用纯函数# 好的做法 - 纯函数def calculate_total(price, tax_rate): return price * (1 + tax_rate)# 不好的做法 - 有副作用total = 0def add_to_total(amount): global total total += amount2. 避免过度使用 Lambda# 不好的做法 - 复杂的 Lambdacomplex_lambda = lambda x: x ** 2 if x > 0 else (x * 2 if x < 0 else 0)# 好的做法 - 使用命名函数def process_number(x): if x > 0: return x ** 2 elif x < 0: return x * 2 else: return 03. 合理使用列表推导式# 简单情况 - 使用列表推导式squared = [x ** 2 for x in range(10)]# 复杂情况 - 使用生成器或循环def complex_process(data): for item in data: # 复杂的处理逻辑 processed = item * 2 if processed > 10: yield processed4. 使用内置函数# 好的做法 - 使用内置函数numbers = [1, 2, 3, 4, 5]total = sum(numbers)maximum = max(numbers)minimum = min(numbers)# 不好的做法 - 手动实现total = 0for num in numbers: total += num总结Python 函数式编程的核心概念：纯函数：相同输入总是产生相同输出，无副作用不可变数据：避免修改原始数据，创建新数据高阶函数：接受或返回函数的函数（map, filter, reduce）Lambda 表达式：匿名函数，适用于简单操作装饰器：修改或增强函数行为偏函数：固定函数参数，创建新函数列表推导式：简洁地创建列表生成器表达式：惰性求值，节省内存函数式编程的优势：代码更简洁、更易读更容易测试和调试更好的并行性减少副作用和状态管理掌握函数式编程技巧，能够编写出更优雅、更高效的 Python 代码。

Python 内存管理机制详解Python 内存管理概述Python 使用自动内存管理，主要通过引用计数（Reference Counting）和垃圾回收（Garbage Collection）两种机制来管理内存。这种机制让开发者无需手动分配和释放内存，大大提高了开发效率。引用计数（Reference Counting）基本原理每个 Python 对象都有一个引用计数器，记录有多少个引用指向该对象。当引用计数降为 0 时，对象会被立即回收。引用计数示例import sysa = [1, 2, 3] # 引用计数 = 1print(sys.getrefcount(a)) # 2 (getrefcount 本身也会创建一个临时引用)b = a # 引用计数 = 2print(sys.getrefcount(a)) # 3c = b # 引用计数 = 3print(sys.getrefcount(a)) # 4del b # 引用计数 = 2print(sys.getrefcount(a)) # 3del c # 引用计数 = 1print(sys.getrefcount(a)) # 2del a # 引用计数 = 0，对象被回收引用计数的变化情况# 1. 赋值操作x = [1, 2, 3]y = x # 引用计数增加# 2. 函数调用def func(obj): passfunc(x) # 函数参数传递时引用计数增加# 3. 容器存储lst = [x, y] # 列表存储时引用计数增加# 4. 删除操作del x # 引用计数减少del y # 引用计数减少del lst # 引用计数减少引用计数的优缺点优点：实时回收：对象不再被引用时立即回收简单高效：无需复杂的标记-清除算法可预测性：内存回收时机明确缺点：无法处理循环引用维护引用计数需要额外开销多线程环境下需要加锁保护循环引用问题什么是循环引用当两个或多个对象相互引用，形成闭环时，即使没有外部引用，它们的引用计数也不会降为 0，导致内存泄漏。class Node: def __init__(self, value): self.value = value self.next = None# 创建循环引用node1 = Node(1)node2 = Node(2)node1.next = node2node2.next = node1 # 形成循环引用# 即使删除外部引用，对象也不会被回收del node1del node2# 此时两个对象的引用计数仍为 1（相互引用）循环引用的解决方案Python 的垃圾回收器专门处理循环引用问题。垃圾回收（Garbage Collection）分代回收机制Python 的垃圾回收器采用分代回收策略，将对象分为三代：第 0 代（Generation 0）：新创建的对象第 1 代（Generation 1）：经历过一次回收仍存活的对象第 2 代（Generation 2）：经历过多次回收仍存活的对象回收阈值import gc# 查看回收阈值print(gc.get_threshold()) # (700, 10, 10)# 含义：# - 700: 第 0 代对象数量达到 700 时触发回收# - 10: 第 0 代回收 10 次后触发第 1 代回收# - 10: 第 1 代回收 10 次后触发第 2 代回收# 设置回收阈值gc.set_threshold(1000, 15, 15)手动触发垃圾回收import gc# 手动触发垃圾回收gc.collect()# 禁用垃圾回收gc.disable()# 启用垃圾回收gc.enable()# 检查是否启用print(gc.isenabled())垃圾回收器工作原理import gcclass MyClass: def __del__(self): print(f"{self} 被回收")# 创建循环引用obj1 = MyClass()obj2 = MyClass()obj1.ref = obj2obj2.ref = obj1# 删除外部引用del obj1del obj2# 手动触发垃圾回收collected = gc.collect()print(f"回收了 {collected} 个对象")内存池机制小对象内存池（Pymalloc）Python 对小对象（小于 512 字节）使用专门的内存池管理，提高内存分配效率。import sys# 小对象使用内存池small_list = [1, 2, 3]print(f"小对象大小: {sys.getsizeof(small_list)} 字节")# 大对象直接使用系统内存large_list = list(range(10000))print(f"大对象大小: {sys.getsizeof(large_list)} 字节")内存池的优势减少内存碎片提高分配速度降低系统调用次数内存优化技巧1. 使用生成器替代列表# 不好的做法 - 使用列表def get_squares_list(n): return [i ** 2 for i in range(n)]# 好的做法 - 使用生成器def get_squares_generator(n): for i in range(n): yield i ** 22. 使用 slots 减少内存占用class Person: def __init__(self, name, age): self.name = name self.age = ageclass PersonWithSlots: __slots__ = ['name', 'age'] def __init__(self, name, age): self.name = name self.age = age# 对比内存占用import sysp1 = Person("Alice", 25)p2 = PersonWithSlots("Alice", 25)print(f"普通对象大小: {sys.getsizeof(p1)} 字节")print(f"使用 __slots__ 对象大小: {sys.getsizeof(p2)} 字节")3. 使用弱引用（Weak Reference）import weakrefclass Cache: def __init__(self): self.cache = weakref.WeakValueDictionary() def get(self, key): return self.cache.get(key) def set(self, key, value): self.cache[key] = value# 使用弱引用避免循环引用cache = Cache()obj = MyClass()cache.set("key", obj)del obj # 对象可以被回收4. 及时释放大对象# 处理大文件def process_large_file(filename): with open(filename, 'r') as f: data = f.read() # 读取大文件 result = process_data(data) del data # 及时释放内存 return result5. 使用适当的数据结构# 使用元组替代列表（不可变数据）coordinates = (1, 2, 3) # 比列表更节省内存# 使用集合替代列表（需要快速查找）unique_items = set(items) # 查找效率更高# 使用字典替代多个列表data = {'names': names, 'ages': ages} # 更好的组织方式内存分析工具1. 使用 sys 模块import sys# 获取对象大小obj = [1, 2, 3, 4, 5]print(f"对象大小: {sys.getsizeof(obj)} 字节")# 获取引用计数print(f"引用计数: {sys.getrefcount(obj)}")2. 使用 gc 模块import gc# 获取所有对象all_objects = gc.get_objects()print(f"对象总数: {len(all_objects)}")# 获取垃圾对象garbage = gc.garbageprint(f"垃圾对象数: {len(garbage)}")# 获取回收统计print(gc.get_stats())3. 使用 tracemalloc 模块import tracemalloc# 开始跟踪内存分配tracemalloc.start()# 执行代码data = [i for i in range(100000)]# 获取内存快照snapshot = tracemalloc.take_snapshot()# 显示内存分配统计top_stats = snapshot.statistics('lineno')for stat in top_stats[:10]: print(stat)# 停止跟踪tracemalloc.stop()4. 使用 memory_profiler# 安装: pip install memory-profilerfrom memory_profiler import profile@profiledef memory_intensive_function(): data = [i for i in range(1000000)] return sum(data)if __name__ == '__main__': memory_intensive_function()常见内存问题及解决方案1. 内存泄漏# 问题代码class Observer: def __init__(self, subject): self.subject = subject subject.observers.append(self) # 形成循环引用# 解决方案 1: 使用弱引用import weakrefclass Observer: def __init__(self, subject): self.subject = weakref.ref(subject) subject.observers.append(self)# 解决方案 2: 提供清理方法class Observer: def __init__(self, subject): self.subject = subject subject.observers.append(self) def cleanup(self): if self in self.subject.observers: self.subject.observers.remove(self)2. 大对象占用过多内存# 问题代码def load_all_data(): return [process_item(item) for item in large_dataset]# 解决方案: 使用生成器def load_data_generator(): for item in large_dataset: yield process_item(item)3. 缓存无限增长# 问题代码cache = {}def get_data(key): if key not in cache: cache[key] = expensive_operation(key) return cache[key]# 解决方案: 使用 LRU 缓存from functools import lru_cache@lru_cache(maxsize=128)def get_data(key): return expensive_operation(key)最佳实践1. 避免不必要的对象创建# 不好的做法def process_items(items): results = [] for item in items: temp = item * 2 results.append(temp) return results# 好的做法def process_items(items): return [item * 2 for item in items]2. 使用上下文管理器# 好的做法 - 自动释放资源with open('large_file.txt', 'r') as f: data = f.read() # 处理数据# 文件自动关闭，内存自动释放3. 及时清理不再需要的引用def process_data(): large_data = load_large_dataset() result = analyze(large_data) del large_data # 及时释放大对象 return result4. 使用适当的数据类型# 使用数组替代列表（数值数据）import arrayarr = array.array('i', [1, 2, 3, 4, 5]) # 更节省内存# 使用字节串替代字符串（二进制数据）data = b'binary data' # 比 str 更节省内存总结Python 的内存管理机制包括：引用计数：实时回收不再使用的对象垃圾回收：处理循环引用问题内存池：提高小对象分配效率分代回收：优化垃圾回收性能内存优化关键点使用生成器替代列表使用 __slots__ 减少对象内存占用使用弱引用避免循环引用及时释放大对象选择合适的数据结构使用缓存时设置大小限制理解 Python 的内存管理机制，有助于编写更高效、更稳定的程序，避免内存泄漏和性能问题。

Python 元编程详解元编程的基本概念元编程是指编写能够操作、生成或修改代码的代码。Python 提供了丰富的元编程工具，包括装饰器、元类、动态属性等。元编程的应用场景框架开发（如 Django ORM）代码生成和自动化动态属性和方法创建面向切面编程（AOP）序列化和反序列化元类（Metaclass）什么是元类元类是创建类的类，就像类是创建对象的模板一样，元类是创建类的模板。# 基本概念class MyClass: pass# MyClass 是 type 的实例print(type(MyClass)) # <class 'type'># obj 是 MyClass 的实例obj = MyClass()print(type(obj)) # <class '__main__.MyClass'>自定义元类class MyMeta(type): def __new__(cls, name, bases, namespace): # 在类创建时执行 print(f"Creating class: {name}") # 添加类属性 namespace['created_by'] = 'MyMeta' return super().__new__(cls, name, bases, namespace)class MyClass(metaclass=MyMeta): passprint(MyClass.created_by) # MyMeta元类的作用class SingletonMeta(type): """单例元类""" _instances = {} def __call__(cls, *args, **kwargs): if cls not in cls._instances: cls._instances[cls] = super().__call__(*args, **kwargs) return cls._instances[cls]class Singleton(metaclass=SingletonMeta): def __init__(self, value): self.value = values1 = Singleton(1)s2 = Singleton(2)print(s1 is s2) # Trueprint(s1.value) # 2元类的高级用法class ValidateMeta(type): """验证元类""" def __new__(cls, name, bases, namespace): # 确保类有特定的属性 if 'required_attr' not in namespace: raise TypeError(f"{name} must have 'required_attr'") # 验证方法 for attr_name, attr_value in namespace.items(): if callable(attr_value) and not attr_name.startswith('_'): if not hasattr(attr_value, '__annotations__'): raise TypeError(f"Method {attr_name} must have type hints") return super().__new__(cls, name, bases, namespace)class ValidatedClass(metaclass=ValidateMeta): required_attr = "value" def method(self, x: int) -> int: return x * 2# class InvalidClass(metaclass=ValidateMeta):# pass # TypeError: InvalidClass must have 'required_attr'动态属性和方法动态属性class DynamicAttributes: def __init__(self): self._data = {} def __getattr__(self, name): """访问不存在的属性时调用""" if name.startswith('get_'): attr_name = name[4:] return self._data.get(attr_name) raise AttributeError(f"'{self.__class__.__name__}' object has no attribute '{name}'") def __setattr__(self, name, value): """设置属性时调用""" if name.startswith('_'): super().__setattr__(name, value) else: self._data[name] = value def __delattr__(self, name): """删除属性时调用""" if name in self._data: del self._data[name] else: super().__delattr__(name)obj = DynamicAttributes()obj.name = "Alice"obj.age = 25print(obj.get_name) # Aliceprint(obj.get_age) # 25动态方法class DynamicMethods: def __init__(self): self.methods = {} def add_method(self, name, func): """动态添加方法""" self.methods[name] = func def __getattr__(self, name): """动态调用方法""" if name in self.methods: return self.methods[name] raise AttributeError(f"'{self.__class__.__name__}' object has no attribute '{name}'")obj = DynamicMethods()# 动态添加方法obj.add_method('greet', lambda self, name: f"Hello, {name}!")obj.add_method('calculate', lambda self, x, y: x + y)print(obj.greet("Alice")) # Hello, Alice!print(obj.calculate(3, 5)) # 8使用 types 模块创建方法import typesclass MyClass: passdef new_method(self): return "This is a dynamically added method"# 动态添加方法MyClass.new_method = new_methodobj = MyClass()print(obj.new_method()) # This is a dynamically added method# 使用 types.MethodTypedef another_method(self, value): return f"Value: {value}"obj.another_method = types.MethodType(another_method, obj)print(obj.another_method(42)) # Value: 42描述符（Descriptor）描述符协议描述符是实现 __get__、__set__ 和 __delete__ 方法的类，用于控制属性的访问。class Descriptor: def __init__(self, name=None): self.name = name def __get__(self, instance, owner): if instance is None: return self return instance.__dict__.get(self.name, f"No {self.name} set") def __set__(self, instance, value): instance.__dict__[self.name] = value def __delete__(self, instance): if self.name in instance.__dict__: del instance.__dict__[self.name]class Person: name = Descriptor('name') age = Descriptor('age')person = Person()person.name = "Alice"person.age = 25print(person.name) # Aliceprint(person.age) # 25描述符的应用class ValidatedAttribute: """验证属性描述符""" def __init__(self, validator=None, default=None): self.validator = validator self.default = default self.name = None def __set_name__(self, owner, name): self.name = f"_{name}" def __get__(self, instance, owner): if instance is None: return self return getattr(instance, self.name, self.default) def __set__(self, instance, value): if self.validator and not self.validator(value): raise ValueError(f"Invalid value for {self.name}: {value}") setattr(instance, self.name, value)class User: name = ValidatedAttribute(lambda x: isinstance(x, str) and len(x) > 0) age = ValidatedAttribute(lambda x: isinstance(x, int) and 0 <= x <= 150) email = ValidatedAttribute(lambda x: '@' in x)user = User()user.name = "Alice"user.age = 25user.email = "alice@example.com"print(user.name) # Aliceprint(user.age) # 25# user.age = -5 # ValueError: Invalid value for _age: -5属性装饰器@property 装饰器class Temperature: def __init__(self, celsius): self._celsius = celsius @property def celsius(self): """获取摄氏温度""" return self._celsius @celsius.setter def celsius(self, value): """设置摄氏温度""" if value < -273.15: raise ValueError("Temperature below absolute zero") self._celsius = value @property def fahrenheit(self): """获取华氏温度（只读）""" return self._celsius * 9/5 + 32temp = Temperature(25)print(temp.celsius) # 25print(temp.fahrenheit) # 77.0temp.celsius = 30print(temp.celsius) # 30# temp.fahrenheit = 100 # AttributeError: can't set attribute动态属性计算class Circle: def __init__(self, radius): self._radius = radius @property def radius(self): return self._radius @radius.setter def radius(self, value): if value <= 0: raise ValueError("Radius must be positive") self._radius = value @property def diameter(self): return self._radius * 2 @property def area(self): return 3.14159 * self._radius ** 2 @property def circumference(self): return 2 * 3.14159 * self._radiuscircle = Circle(5)print(circle.diameter) # 10print(circle.area) # 78.53975print(circle.circumference) # 31.4159动态类创建使用 type 创建类# 动态创建类def __init__(self, name): self.name = namedef greet(self): return f"Hello, {self.name}!"# 使用 type 创建类DynamicClass = type( 'DynamicClass', (object,), { '__init__': __init__, 'greet': greet, 'class_var': 'dynamic' })obj = DynamicClass("Alice")print(obj.greet()) # Hello, Alice!print(obj.class_var) # dynamic动态创建子类def create_subclass(base_class, subclass_name, extra_methods=None): """动态创建子类""" namespace = extra_methods or {} return type(subclass_name, (base_class,), namespace)class Base: def base_method(self): return "Base method"# 动态创建子类extra_methods = { 'extra_method': lambda self: "Extra method"}SubClass = create_subclass(Base, 'SubClass', extra_methods)obj = SubClass()print(obj.base_method()) # Base methodprint(obj.extra_method()) # Extra method类装饰器基本类装饰器def add_class_method(cls): """添加类方法的装饰器""" @classmethod def class_method(cls): return f"Class method of {cls.__name__}" cls.class_method = class_method return cls@add_class_methodclass MyClass: passprint(MyClass.class_method()) # Class method of MyClass类装饰器的应用def singleton(cls): """单例类装饰器""" instances = {} def get_instance(*args, **kwargs): if cls not in instances: instances[cls] = cls(*args, **kwargs) return instances[cls] return get_instance@singletonclass Database: def __init__(self): self.connection = "Connected"db1 = Database()db2 = Database()print(db1 is db2) # True参数化类装饰器def add_attributes(**attrs): """添加类属性的装饰器""" def decorator(cls): for name, value in attrs.items(): setattr(cls, name, value) return cls return decorator@add_attributes(version="1.0", author="Alice")class MyClass: passprint(MyClass.version) # 1.0print(MyClass.author) # Alice实际应用场景1. ORM 框架class Field: """字段描述符""" def __init__(self, field_type, primary_key=False): self.field_type = field_type self.primary_key = primary_key self.name = None def __set_name__(self, owner, name): self.name = name def __get__(self, instance, owner): if instance is None: return self return instance.__dict__.get(self.name) def __set__(self, instance, value): if not isinstance(value, self.field_type): raise TypeError(f"Expected {self.field_type}, got {type(value)}") instance.__dict__[self.name] = valueclass ModelMeta(type): """模型元类""" def __new__(cls, name, bases, namespace): # 收集字段 fields = {} for key, value in namespace.items(): if isinstance(value, Field): fields[key] = value namespace['_fields'] = fields return super().__new__(cls, name, bases, namespace)class Model(metaclass=ModelMeta): def __init__(self, **kwargs): for name, value in kwargs.items(): setattr(self, name, value)class User(Model): id = Field(int, primary_key=True) name = Field(str) age = Field(int)user = User(id=1, name="Alice", age=25)print(user.name) # Aliceprint(user.age) # 252. API 响应验证class ValidatedResponse: """验证响应类""" def __init__(self, schema): self.schema = schema def __call__(self, cls): def __init__(self, data): self.validate(data) for key, value in data.items(): setattr(self, key, value) def validate(self, data): for field, field_type in self.schema.items(): if field not in data: raise ValueError(f"Missing field: {field}") if not isinstance(data[field], field_type): raise TypeError(f"Invalid type for {field}") cls.__init__ = __init__ cls.validate = validate return cls@ValidatedResponse({'name': str, 'age': int, 'email': str})class UserResponse: passuser_data = {'name': 'Alice', 'age': 25, 'email': 'alice@example.com'}user = UserResponse(user_data)print(user.name) # Alice3. 动态表单生成class FormField: """表单字段""" def __init__(self, field_type, required=False, default=None): self.field_type = field_type self.required = required self.default = default self.name = None def __set_name__(self, owner, name): self.name = name def validate(self, value): if self.required and value is None: raise ValueError(f"{self.name} is required") if value is not None and not isinstance(value, self.field_type): raise TypeError(f"Invalid type for {self.name}") return Trueclass FormMeta(type): """表单元类""" def __new__(cls, name, bases, namespace): fields = {} for key, value in namespace.items(): if isinstance(value, FormField): fields[key] = value namespace['_fields'] = fields return super().__new__(cls, name, bases, namespace)class Form(metaclass=FormMeta): def __init__(self, **kwargs): for name, field in self._fields.items(): value = kwargs.get(name, field.default) field.validate(value) setattr(self, name, value) def to_dict(self): return {name: getattr(self, name) for name in self._fields}class UserForm(Form): name = FormField(str, required=True) age = FormField(int, default=18) email = FormField(str, required=True)form = UserForm(name="Alice", email="alice@example.com")print(form.to_dict()) # {'name': 'Alice', 'age': 18, 'email': 'alice@example.com'}最佳实践1. 谨慎使用元类# 不好的做法 - 过度使用元类class ComplexMeta(type): def __new__(cls, name, bases, namespace): # 复杂的元类逻辑 pass# 好的做法 - 使用类装饰器def add_functionality(cls): # 添加功能 return cls@add_functionalityclass SimpleClass: pass2. 优先使用描述符而非 getattr# 好的做法 - 使用描述符class ValidatedField: def __get__(self, instance, owner): return instance.__dict__.get(self.name) def __set__(self, instance, value): instance.__dict__[self.name] = valueclass MyClass: field = ValidatedField()# 不好的做法 - 使用 __getattr__class BadClass: def __getattr__(self, name): return self.__dict__.get(name)3. 提供清晰的文档class MyMeta(type): """自定义元类，用于添加类级别的功能 这个元类会自动为所有类添加 created_at 属性 """ def __new__(cls, name, bases, namespace): namespace['created_at'] = datetime.now() return super().__new__(cls, name, bases, namespace)4. 考虑性能影响# 缓存属性访问class CachedProperty: def __init__(self, func): self.func = func self.name = func.__name__ def __get__(self, instance, owner): if instance is None: return self if not hasattr(instance, f'_{self.name}'): setattr(instance, f'_{self.name}', self.func(instance)) return getattr(instance, f'_{self.name}')class MyClass: @CachedProperty def expensive_computation(self): # 耗时计算 return sum(range(1000000))总结Python 元编程的核心概念：元类：创建类的类，控制类的创建过程动态属性：使用 __getattr__、__setattr__ 等方法动态管理属性动态方法：运行时添加和修改方法描述符：控制属性的访问和修改属性装饰器：使用 @property 创建计算属性动态类创建：使用 type 函数动态创建类类装饰器：修改或增强类的行为元编程的应用场景：框架开发（ORM、表单验证）代码生成和自动化动态 API 创建序列化和反序列化面向切面编程元编程的注意事项：谨慎使用，避免过度设计提供清晰的文档和示例考虑性能影响优先使用简单的解决方案掌握元编程技巧，能够编写出更灵活、更强大的 Python 代码。

Python 元类详解元类的基本概念元类是 Python 中用于创建类的"类"。在 Python 中，一切皆对象，类本身也是对象，而元类就是创建这些类对象的类。类与元类的关系# 在 Python 中，type 是默认的元类class MyClass: pass# MyClass 是 type 的实例print(type(MyClass)) # <class 'type'># type 是它自己的元类print(type(type)) # <class 'type'>type 函数type 的三种用法# 1. type(obj) - 获取对象的类型obj = "hello"print(type(obj)) # <class 'str'># 2. type(name, bases, dict) - 动态创建类# name: 类名# bases: 基类元组# dict: 类属性字典# 传统方式定义类class TraditionalClass: attr = "value" def method(self): return "method called"# 使用 type 动态创建类DynamicClass = type( "DynamicClass", # 类名 (), # 基类 { # 类属性 "attr": "value", "method": lambda self: "method called" })print(DynamicClass.attr) # valueprint(DynamicClass().method()) # method called动态创建类的实际应用# 动态创建具有特定属性的类def create_class(class_name, attributes): """动态创建具有指定属性的类""" class_dict = {} for attr_name, attr_value in attributes.items(): if callable(attr_value): class_dict[attr_name] = attr_value else: class_dict[attr_name] = attr_value return type(class_name, (), class_dict)# 定义属性和方法attributes = { "name": "Dynamic", "age": 25, "greet": lambda self: f"Hello, I'm {self.name}"}# 创建类Person = create_class("Person", attributes)# 使用类person = Person()print(person.name) # Dynamicprint(person.greet()) # Hello, I'm Dynamic自定义元类基本元类定义# 定义元类class MyMeta(type): def __new__(cls, name, bases, attrs): print(f"创建类: {name}") print(f"基类: {bases}") print(f"属性: {list(attrs.keys())}") # 可以修改属性 attrs['created_by'] = 'MyMeta' # 调用父类的 __new__ 方法创建类 return super().__new__(cls, name, bases, attrs)# 使用元类class MyClass(metaclass=MyMeta): def __init__(self): self.value = 42# 输出:# 创建类: MyClass# 基类: ()# 属性: ['__module__', '__qualname__', '__init__']print(MyClass.created_by) # MyMeta元类继承class BaseMeta(type): """基础元类""" def __new__(cls, name, bases, attrs): attrs['base_attribute'] = 'from_base_meta' return super().__new__(cls, name, bases, attrs)class ExtendedMeta(BaseMeta): """扩展元类""" def __new__(cls, name, bases, attrs): attrs['extended_attribute'] = 'from_extended_meta' return super().__new__(cls, name, bases, attrs)# 使用扩展元类class MyClass(metaclass=ExtendedMeta): passprint(MyClass.base_attribute) # from_base_metaprint(MyClass.extended_attribute) # from_extended_meta元类的应用场景1. 自动添加方法class AutoMethodMeta(type): """自动添加方法的元类""" def __new__(cls, name, bases, attrs): # 为每个属性添加 getter 和 setter for key, value in list(attrs.items()): if not key.startswith('_') and not callable(value): # 添加 getter getter_name = f'get_{key}' attrs[getter_name] = lambda self, k=key: getattr(self, k) # 添加 setter setter_name = f'set_{key}' attrs[setter_name] = lambda self, v, k=key: setattr(self, k, v) return super().__new__(cls, name, bases, attrs)class Person(metaclass=AutoMethodMeta): name = "" age = 0person = Person()person.set_name("Alice")person.set_age(25)print(person.get_name()) # Aliceprint(person.get_age()) # 252. 属性验证class ValidatedMeta(type): """属性验证元类""" def __new__(cls, name, bases, attrs): # 查找验证规则 validations = attrs.pop('_validations', {}) # 创建验证后的属性 for attr_name, validator in validations.items(): original_value = attrs.get(attr_name) def make_property(attr_name, validator, original_value): def getter(self): return getattr(self, f'_{attr_name}', original_value) def setter(self, value): if not validator(value): raise ValueError(f"Invalid value for {attr_name}: {value}") setattr(self, f'_{attr_name}', value) return property(getter, setter) attrs[attr_name] = make_property(attr_name, validator, original_value) return super().__new__(cls, name, bases, attrs)class Person(metaclass=ValidatedMeta): _validations = { 'age': lambda x: isinstance(x, int) and 0 <= x <= 150, 'name': lambda x: isinstance(x, str) and len(x) > 0 } age = 0 name = ""person = Person()person.age = 25 # 正常person.name = "Alice" # 正常# person.age = -5 # ValueError: Invalid value for age: -5# person.name = "" # ValueError: Invalid value for name: 3. 单例模式class SingletonMeta(type): """单例元类""" _instances = {} def __call__(cls, *args, **kwargs): if cls not in cls._instances: cls._instances[cls] = super().__call__(*args, **kwargs) return cls._instances[cls]class Database(metaclass=SingletonMeta): def __init__(self): self.connection = "Connected"db1 = Database()db2 = Database()print(db1 is db2) # Trueprint(db1.connection) # Connected4. 注册机制class PluginMeta(type): """插件注册元类""" _registry = {} def __new__(cls, name, bases, attrs): new_class = super().__new__(cls, name, bases, attrs) # 如果类有 plugin_name 属性，则注册 if hasattr(new_class, 'plugin_name'): PluginMeta._registry[new_class.plugin_name] = new_class return new_class @classmethod def get_plugin(cls, name): return cls._registry.get(name) @classmethod def list_plugins(cls): return list(cls._registry.keys())# 定义插件class EmailPlugin(metaclass=PluginMeta): plugin_name = "email" def send(self, message): return f"Email sent: {message}"class SMSPlugin(metaclass=PluginMeta): plugin_name = "sms" def send(self, message): return f"SMS sent: {message}"# 使用插件print(PluginMeta.list_plugins()) # ['email', 'sms']email_plugin = PluginMeta.get_plugin("email")print(email_plugin().send("Hello")) # Email sent: Hello5. 接口检查class InterfaceMeta(type): """接口检查元类""" def __new__(cls, name, bases, attrs): # 检查是否实现了所有必需的方法 if hasattr(cls, '_required_methods'): for method_name in cls._required_methods: if method_name not in attrs: raise NotImplementedError( f"Class {name} must implement method {method_name}" ) return super().__new__(cls, name, bases, attrs)class DataProcessor(metaclass=InterfaceMeta): _required_methods = ['load', 'process', 'save'] def load(self): pass def process(self): pass def save(self): pass# class IncompleteProcessor(metaclass=InterfaceMeta):# _required_methods = ['load', 'process', 'save']# # def load(self):# pass# # # NotImplementedError: Class IncompleteProcessor must implement method process6. 自动文档生成class DocumentedMeta(type): """自动文档生成元类""" def __new__(cls, name, bases, attrs): # 收集文档信息 doc_info = { 'class_name': name, 'methods': {}, 'attributes': [] } # 收集方法文档 for key, value in attrs.items(): if callable(value) and hasattr(value, '__doc__') and value.__doc__: doc_info['methods'][key] = value.__doc__.strip() elif not key.startswith('_') and not callable(value): doc_info['attributes'].append(key) # 添加文档属性 attrs['_doc_info'] = doc_info return super().__new__(cls, name, bases, attrs)class Calculator(metaclass=DocumentedMeta): """计算器类""" def add(self, a, b): """加法运算""" return a + b def subtract(self, a, b): """减法运算""" return a - b version = "1.0"# 查看文档print(Calculator._doc_info)# {# 'class_name': 'Calculator',# 'methods': {# 'add': '加法运算',# 'subtract': '减法运算'# },# 'attributes': ['version']# }元类的高级用法元类与装饰器的结合def class_decorator(cls): """类装饰器""" cls.decorated = True return clsclass MetaWithDecorator(type): """结合装饰器的元类""" def __new__(cls, name, bases, attrs): # 创建类 new_class = super().__new__(cls, name, bases, attrs) # 应用装饰器 new_class = class_decorator(new_class) return new_classclass MyClass(metaclass=MetaWithDecorator): passprint(MyClass.decorated) # True元类的 init 方法class InitMeta(type): """使用 __init__ 的元类""" def __init__(cls, name, bases, attrs): super().__init__(name, bases, attrs) # 在类创建后执行初始化 cls.initialized = True cls.creation_time = __import__('time').time()class MyClass(metaclass=InitMeta): passprint(MyClass.initialized) # Trueprint(MyClass.creation_time) # 创建时间戳元类的 call 方法class CallMeta(type): """使用 __call__ 的元类""" def __call__(cls, *args, **kwargs): print(f"创建 {cls.__name__} 实例") print(f"参数: args={args}, kwargs={kwargs}") # 创建实例 instance = super().__call__(*args, **kwargs) # 可以在实例创建后进行额外操作 instance.created_by_meta = True return instanceclass MyClass(metaclass=CallMeta): def __init__(self, value): self.value = valueobj = MyClass(42)print(obj.value) # 42print(obj.created_by_meta) # True元类的最佳实践1. 何时使用元类# 适合使用元类的情况：# - 需要修改类的创建过程# - 需要为多个类添加相同的功能# - 需要实现设计模式（如单例、注册表）# - 需要进行接口检查或验证# 不适合使用元类的情况：# - 简单的类装饰器就能解决问题# - 只需要修改实例行为# - 代码可读性比功能更重要2. 元类 vs 类装饰器# 类装饰器 - 更简单，更易读def add_method(cls): cls.new_method = lambda self: "new method" return cls@add_methodclass MyClass1: pass# 元类 - 更强大，更灵活class AddMethodMeta(type): def __new__(cls, name, bases, attrs): attrs['new_method'] = lambda self: "new method" return super().__new__(cls, name, bases, attrs)class MyClass2(metaclass=AddMethodMeta): pass# 两者效果相同，但元类可以继承和组合3. 元类的性能考虑import time# 元类在类创建时执行一次，而不是实例创建时class ExpensiveMeta(type): def __new__(cls, name, bases, attrs): # 耗时操作 time.sleep(0.1) return super().__new__(cls, name, bases, attrs)# 类创建时执行一次（0.1秒）start = time.time()class MyClass(metaclass=ExpensiveMeta): passprint(f"类创建时间: {time.time() - start:.2f}秒")# 实例创建时不执行（0秒）start = time.time()obj1 = MyClass()obj2 = MyClass()print(f"实例创建时间: {time.time() - start:.2f}秒")实际应用案例1. ORM 框架中的元类class ModelMeta(type): """ORM 模型元类""" def __new__(cls, name, bases, attrs): # 收集字段信息 fields = {} for key, value in list(attrs.items()): if hasattr(value, 'is_field'): fields[key] = value del attrs[key] attrs['_fields'] = fields return super().__new__(cls, name, bases, attrs)class Field: def __init__(self, field_type): self.field_type = field_type self.is_field = Trueclass User(metaclass=ModelMeta): name = Field(str) age = Field(int) email = Field(str)print(User._fields)# {'name': Field(str), 'age': Field(int), 'email': Field(str)}2. API 客户端中的元类class APIClientMeta(type): """API 客户端元类""" def __new__(cls, name, bases, attrs): # 为每个方法添加 API 调用包装 for key, value in list(attrs.items()): if callable(value) and not key.startswith('_'): def make_wrapper(original_func): def wrapper(self, *args, **kwargs): print(f"API 调用: {original_func.__name__}") result = original_func(self, *args, **kwargs) print(f"API 响应: {result}") return result return wrapper attrs[key] = make_wrapper(value) return super().__new__(cls, name, bases, attrs)class APIClient(metaclass=APIClientMeta): def get_user(self, user_id): return {"id": user_id, "name": "Alice"} def create_user(self, name, email): return {"name": name, "email": email}client = APIClient()client.get_user(1)client.create_user("Bob", "bob@example.com")总结Python 元类的核心概念：基本概念：元类是创建类的类，type 是默认元类type 函数：可以动态创建类自定义元类：继承 type 类，重写 __new__、__init__、__call__ 方法应用场景：自动添加方法属性验证单例模式注册机制接口检查自动文档生成元类的优势：强大的类创建控制可以批量修改类行为实现复杂的设计模式减少重复代码元类的注意事项：增加代码复杂度可能影响可读性调试难度增加性能考虑元类的使用原则：优先考虑更简单的解决方案（装饰器、继承）只在必要时使用元类提供清晰的文档和示例考虑代码的可维护性掌握元类，能够深入理解 Python 的面向对象机制，编写出更强大、更灵活的代码。但要注意，元类是 Python 的高级特性，应该谨慎使用。

Python 中的闭包详解闭包的基本概念闭包是 Python 中一个重要的概念，它是指一个函数对象，即使在其定义作用域之外执行时，仍然能够访问其定义作用域中的变量。闭包的基本结构def outer_function(x): """外部函数""" def inner_function(y): """内部函数""" return x + y return inner_function# 创建闭包closure = outer_function(10)# 调用闭包print(closure(5)) # 15print(closure(20)) # 30闭包的三个条件必须有一个嵌套函数（内部函数）内部函数必须引用外部函数中的变量外部函数必须返回这个内部函数def make_multiplier(factor): """创建乘法闭包""" def multiply(number): return number * factor return multiply# 创建不同的乘法器double = make_multiplier(2)triple = make_multiplier(3)print(double(5)) # 10print(triple(5)) # 15闭包的工作原理变量的作用域def outer(): x = 10 def inner(): # 内部函数可以访问外部函数的变量 print(f"内部函数访问 x: {x}") return x return innerclosure = outer()print(closure()) # 内部函数访问 x: 10, 10变量的生命周期def counter(): """计数器闭包""" count = 0 def increment(): nonlocal count count += 1 return count return increment# 创建计数器my_counter = counter()print(my_counter()) # 1print(my_counter()) # 2print(my_counter()) # 3# 创建另一个计数器another_counter = counter()print(another_counter()) # 1closure 属性def outer(x): def inner(y): return x + y return innerclosure = outer(10)# 查看闭包的变量print(closure.__closure__) # (<cell at 0x...: int object at 0x...>,)print(closure.__closure__[0].cell_contents) # 10闭包的实际应用1. 数据隐藏和封装def make_account(initial_balance): """创建银行账户""" balance = initial_balance def deposit(amount): nonlocal balance balance += amount return balance def withdraw(amount): nonlocal balance if amount <= balance: balance -= amount return balance else: raise ValueError("余额不足") def get_balance(): return balance # 返回多个函数 return { 'deposit': deposit, 'withdraw': withdraw, 'get_balance': get_balance }# 创建账户account = make_account(100)# 使用账户print(account['deposit'](50)) # 150print(account['withdraw'](30)) # 120print(account['get_balance']()) # 120# balance 变量被隐藏，无法直接访问# print(balance) # NameError: name 'balance' is not defined2. 函数工厂def make_power_function(power): """创建幂函数""" def power_function(base): return base ** power return power_function# 创建不同的幂函数square = make_power_function(2)cube = make_power_function(3)fourth_power = make_power_function(4)print(square(3)) # 9print(cube(3)) # 27print(fourth_power(3)) # 813. 延迟计算def lazy_sum(*args): """延迟求和""" def sum(): total = 0 for num in args: total += num return total return sum# 创建延迟求和函数f = lazy_sum(1, 2, 3, 4, 5)# 调用时才计算print(f()) # 154. 缓存和记忆化def memoize(func): """记忆化装饰器""" cache = {} def memoized(*args): if args not in cache: cache[args] = func(*args) return cache[args] return memoized@memoizedef fibonacci(n): """斐波那契数列""" if n < 2: return n return fibonacci(n-1) + fibonacci(n-2)print(fibonacci(10)) # 55print(fibonacci(20)) # 67655. 回调函数def make_callback(callback): """创建回调函数""" def execute(*args, **kwargs): print("执行回调前...") result = callback(*args, **kwargs) print("执行回调后...") return result return executedef my_function(x, y): return x + y# 创建带回调的函数callback_function = make_callback(my_function)print(callback_function(3, 5)) # 执行回调前..., 8, 执行回调后...6. 状态保持def make_state_machine(): """创建状态机""" state = 'idle' def transition(action): nonlocal state print(f"当前状态: {state}, 动作: {action}") if state == 'idle': if action == 'start': state = 'running' elif state == 'running': if action == 'pause': state = 'paused' elif action == 'stop': state = 'idle' elif state == 'paused': if action == 'resume': state = 'running' elif action == 'stop': state = 'idle' print(f"新状态: {state}") return state return transition# 创建状态机state_machine = make_state_machine()state_machine('start') # idle -> runningstate_machine('pause') # running -> pausedstate_machine('resume') # paused -> runningstate_machine('stop') # running -> idle闭包与装饰器闭包实现装饰器def my_decorator(func): """简单的装饰器""" def wrapper(): print("装饰器：函数调用前") result = func() print("装饰器：函数调用后") return result return wrapper@my_decoratordef say_hello(): print("Hello!")say_hello()# 输出:# 装饰器：函数调用前# Hello!# 装饰器：函数调用后带参数的装饰器def repeat(times): """重复执行装饰器""" def decorator(func): def wrapper(*args, **kwargs): results = [] for _ in range(times): result = func(*args, **kwargs) results.append(result) return results return wrapper return decorator@repeat(3)def greet(name): return f"Hello, {name}!"print(greet("Alice"))# 输出: ['Hello, Alice!', 'Hello, Alice!', 'Hello, Alice!']闭包的注意事项1. 循环变量的陷阱# 错误的做法def create_multipliers(): return [lambda x: x * i for i in range(5)]multipliers = create_multipliers()print([m(2) for m in multipliers]) # [8, 8, 8, 8, 8] - 错误！# 正确的做法 - 使用默认参数def create_multipliers_correct(): return [lambda x, i=i: x * i for i in range(5)]multipliers_correct = create_multipliers_correct()print([m(2) for m in multipliers_correct]) # [0, 2, 4, 6, 8] - 正确2. 修改外部变量def outer(): count = 0 def increment(): nonlocal count # 必须使用 nonlocal 关键字 count += 1 return count return incrementcounter = outer()print(counter()) # 1print(counter()) # 23. 内存泄漏风险def large_closure(): """创建大闭包""" large_data = list(range(1000000)) def process(): return sum(large_data[:100]) return process# 闭包会保持对 large_data 的引用# 即使只使用其中的一小部分closure = large_closure()# 如果不再需要闭包，应该删除引用del closure闭包 vs 类闭包实现def make_counter(): """使用闭包实现计数器""" count = 0 def increment(): nonlocal count count += 1 return count def get_count(): return count return { 'increment': increment, 'get_count': get_count }counter = make_counter()print(counter['increment']()) # 1print(counter['increment']()) # 2print(counter['get_count']()) # 2类实现class Counter: """使用类实现计数器""" def __init__(self): self.count = 0 def increment(self): self.count += 1 return self.count def get_count(self): return self.countcounter = Counter()print(counter.increment()) # 1print(counter.increment()) # 2print(counter.get_count()) # 2何时使用闭包 vs 类# 使用闭包的场景：# 1. 简单的状态保持def make_accumulator(): total = 0 def add(value): nonlocal total total += value return total return add# 2. 函数工厂def make_power(power): def power_function(base): return base ** power return power_function# 使用类的场景：# 1. 复杂的状态管理class BankAccount: def __init__(self, initial_balance): self.balance = initial_balance self.transactions = [] def deposit(self, amount): self.balance += amount self.transactions.append(('deposit', amount)) def withdraw(self, amount): if amount <= self.balance: self.balance -= amount self.transactions.append(('withdraw', amount)) def get_balance(self): return self.balance def get_transactions(self): return self.transactions# 2. 需要多个方法和属性class Calculator: def __init__(self): self.history = [] def add(self, a, b): result = a + b self.history.append(f"{a} + {b} = {result}") return result def subtract(self, a, b): result = a - b self.history.append(f"{a} - {b} = {result}") return result def get_history(self): return self.history闭包的高级应用1. 部分函数应用def partial(func, *args, **kwargs): """部分函数应用""" def wrapper(*more_args, **more_kwargs): all_args = args + more_args all_kwargs = {**kwargs, **more_kwargs} return func(*all_args, **all_kwargs) return wrapperdef power(base, exponent): return base ** exponentsquare = partial(power, exponent=2)cube = partial(power, exponent=3)print(square(5)) # 25print(cube(5)) # 1252. 函数组合def compose(*functions): """函数组合""" def wrapper(arg): result = arg for func in reversed(functions): result = func(result) return result return wrapperdef add_one(x): return x + 1def multiply_two(x): return x * 2def square(x): return x ** 2# 组合函数combined = compose(square, multiply_two, add_one)print(combined(3)) # ((3 + 1) * 2) ** 2 = 643. 验证器def make_validator(validator_func, error_message): """创建验证器""" def validate(value): if not validator_func(value): raise ValueError(error_message) return value return validate# 创建验证器is_positive = make_validator( lambda x: x > 0, "值必须为正数")is_email = make_validator( lambda x: '@' in x and '.' in x, "无效的邮箱地址")# 使用验证器print(is_positive(10)) # 10# is_positive(-5) # ValueError: 值必须为正数print(is_email("user@example.com")) # user@example.com# is_email("invalid") # ValueError: 无效的邮箱地址4. 限流器import timedef rate_limiter(max_calls, time_window): """创建限流器""" calls = [] def limiter(func): def wrapper(*args, **kwargs): current_time = time.time() # 移除超出时间窗口的调用记录 calls[:] = [call_time for call_time in calls if current_time - call_time < time_window] # 检查是否超过限制 if len(calls) >= max_calls: raise Exception(f"超过限流限制：{max_calls} 次/{time_window} 秒") # 记录调用 calls.append(current_time) # 执行函数 return func(*args, **kwargs) return wrapper return limiter@rate_limiter(max_calls=3, time_window=1)def api_call(): print("API 调用成功") return "success"# 测试限流api_call() # 成功api_call() # 成功api_call() # 成功# api_call() # Exception: 超过限流限制：3 次/1 秒总结Python 闭包的核心概念：基本定义：闭包是一个函数对象，能够访问其定义作用域中的变量三个条件：嵌套函数、引用外部变量、返回内部函数工作原理：通过 __closure__ 属性保持对外部变量的引用闭包的实际应用：数据隐藏和封装函数工厂延迟计算缓存和记忆化回调函数状态保持闭包的注意事项：循环变量的陷阱使用 nonlocal 修改外部变量注意内存泄漏风险闭包 vs 类：闭包：适合简单的状态保持和函数工厂类：适合复杂的状态管理和多个方法闭包的高级应用：部分函数应用函数组合验证器限流器闭包是 Python 中一个强大而优雅的特性，它允许函数保持状态，实现数据隐藏，并创建更加灵活和可重用的代码。掌握闭包对于编写高质量的 Python 代码非常重要。

Python 上下文管理器详解上下文管理器的基本概念上下文管理器是 Python 中用于管理资源的对象，它定义了进入和退出上下文时应该执行的操作。最常见的使用方式是 with 语句。为什么需要上下文管理器# 不好的做法 - 手动管理资源file = open('example.txt', 'r')try: content = file.read() process(content)finally: file.close()# 好的做法 - 使用上下文管理器with open('example.txt', 'r') as file: content = file.read() process(content)# 文件自动关闭上下文管理器协议上下文管理器需要实现两个方法：__enter__(self): 进入上下文时调用__exit__(self, exc_type, exc_val, exc_tb): 退出上下文时调用基本实现class MyContextManager: def __init__(self, resource): self.resource = resource def __enter__(self): print("进入上下文") return self.resource def __exit__(self, exc_type, exc_val, exc_tb): print("退出上下文") # 返回 True 表示抑制异常，False 表示传播异常 return False# 使用上下文管理器with MyContextManager("资源") as resource: print(f"使用资源: {resource}")# 输出:# 进入上下文# 使用资源: 资源# 退出上下文异常处理class SafeContextManager: def __enter__(self): print("进入安全上下文") return self def __exit__(self, exc_type, exc_val, exc_tb): print("退出安全上下文") if exc_type is not None: print(f"捕获到异常: {exc_type.__name__}: {exc_val}") # 返回 True 抑制异常 return True return False# 测试异常处理with SafeContextManager(): print("执行操作") raise ValueError("测试异常")print("程序继续执行")# 输出:# 进入安全上下文# 执行操作# 退出安全上下文# 捕获到异常: ValueError: 测试异常# 程序继续执行contextlib 模块@contextmanager 装饰器@contextmanager 装饰器简化了上下文管理器的创建。from contextlib import contextmanager@contextmanagerdef simple_context(): print("进入上下文") try: yield "资源" finally: print("退出上下文")# 使用with simple_context() as resource: print(f"使用资源: {resource}")带异常处理的上下文管理器from contextlib import contextmanager@contextmanagerdef error_handling_context(): print("进入错误处理上下文") try: yield except ValueError as e: print(f"处理 ValueError: {e}") raise # 重新抛出异常 finally: print("清理资源")# 使用try: with error_handling_context(): print("执行操作") raise ValueError("测试异常")except ValueError: print("捕获到重新抛出的异常")closing 函数closing 函数为没有上下文管理器协议的对象创建上下文管理器。from contextlib import closingclass Resource: def __init__(self, name): self.name = name def close(self): print(f"关闭资源: {self.name}")# 使用 closingwith closing(Resource("数据库连接")) as resource: print(f"使用资源: {resource.name}")# 资源自动关闭suppress 函数suppress 函数用于忽略指定的异常。from contextlib import suppress# 忽略 FileNotFoundErrorwith suppress(FileNotFoundError): with open('nonexistent.txt', 'r') as f: content = f.read()print("程序继续执行")# 忽略多个异常with suppress(FileNotFoundError, PermissionError): with open('protected.txt', 'r') as f: content = f.read()redirectstdout 和 redirectstderrfrom contextlib import redirect_stdout, redirect_stderrimport io# 重定向标准输出output = io.StringIO()with redirect_stdout(output): print("这条消息被重定向")print(f"捕获的输出: {output.getvalue()}")# 重定向标准错误error_output = io.StringIO()with redirect_stderr(error_output): print("错误消息", file=sys.stderr)print(f"捕获的错误: {error_output.getvalue()}")实际应用场景1. 文件操作# 自动关闭文件with open('input.txt', 'r') as input_file: with open('output.txt', 'w') as output_file: for line in input_file: output_file.write(line.upper())# 两个文件都会自动关闭2. 数据库连接import sqlite3from contextlib import contextmanager@contextmanagerdef database_connection(db_path): conn = sqlite3.connect(db_path) try: yield conn finally: conn.close()# 使用with database_connection('example.db') as conn: cursor = conn.cursor() cursor.execute("SELECT * FROM users") results = cursor.fetchall()# 连接自动关闭3. 锁管理import threadingfrom contextlib import contextmanagerclass LockManager: def __init__(self): self.lock = threading.Lock() @contextmanager def acquire(self): self.lock.acquire() try: yield finally: self.lock.release()# 使用lock_manager = LockManager()with lock_manager.acquire(): # 临界区代码 print("执行临界区操作")# 锁自动释放4. 临时目录import tempfileimport os# 创建临时目录with tempfile.TemporaryDirectory() as temp_dir: print(f"临时目录: {temp_dir}") temp_file = os.path.join(temp_dir, 'temp.txt') with open(temp_file, 'w') as f: f.write("临时数据") # 临时目录及其内容在退出时自动删除# 临时目录已被删除5. 计时器import timefrom contextlib import contextmanager@contextmanagerdef timer(name): start_time = time.time() yield end_time = time.time() print(f"{name} 耗时: {end_time - start_time:.4f} 秒")# 使用with timer("数据处理"): data = [i ** 2 for i in range(1000000)] sum(data)# 输出: 数据处理 耗时: 0.1234 秒6. 临时改变配置from contextlib import contextmanagerclass Config: def __init__(self): self.debug = False self.verbose = Falseconfig = Config()@contextmanagerdef temporary_config(config_obj, **kwargs): """临时修改配置""" original_values = {} # 保存原始值 for key, value in kwargs.items(): original_values[key] = getattr(config_obj, key) setattr(config_obj, key, value) try: yield finally: # 恢复原始值 for key, value in original_values.items(): setattr(config_obj, key, value)# 使用print(f"调试模式: {config.debug}") # Falsewith temporary_config(config, debug=True, verbose=True): print(f"调试模式: {config.debug}") # True print(f"详细模式: {config.verbose}") # Trueprint(f"调试模式: {config.debug}") # False7. 事务管理from contextlib import contextmanagerclass Database: def __init__(self): self.in_transaction = False self.data = {} def begin_transaction(self): self.in_transaction = True self.transaction_data = self.data.copy() def commit(self): self.in_transaction = False print("事务提交") def rollback(self): self.in_transaction = False self.data = self.transaction_data print("事务回滚")@contextmanagerdef transaction(db): db.begin_transaction() try: yield db db.commit() except Exception as e: db.rollback() raise# 使用db = Database()try: with transaction(db): db.data['key1'] = 'value1' db.data['key2'] = 'value2' # raise Exception("测试异常") # 会触发回滚except Exception as e: print(f"事务失败: {e}")print(db.data)嵌套上下文管理器使用多个 with 语句# 嵌套使用with open('file1.txt', 'r') as f1: with open('file2.txt', 'r') as f2: content1 = f1.read() content2 = f2.read() # 处理两个文件使用 contextlib.ExitStackExitStack 允许动态管理多个上下文管理器。from contextlib import ExitStackfiles = ['file1.txt', 'file2.txt', 'file3.txt']with ExitStack() as stack: file_handles = [stack.enter_context(open(f, 'r')) for f in files] contents = [f.read() for f in file_handles] # 所有文件在退出时自动关闭条件性上下文管理器from contextlib import ExitStack, nullcontextdef get_context(use_context): if use_context: return MyContextManager("资源") else: return nullcontext()# 条件性使用上下文管理器with get_context(True) as resource: if resource is not None: print(f"使用资源: {resource}") else: print("不使用上下文管理器")异步上下文管理器异步上下文管理器协议异步上下文管理器实现：__aenter__(self): 异步进入上下文__aexit__(self, exc_type, exc_val, exc_tb): 异步退出上下文class AsyncContextManager: def __init__(self, resource): self.resource = resource async def __aenter__(self): print("异步进入上下文") await self.connect() return self.resource async def __aexit__(self, exc_type, exc_val, exc_tb): print("异步退出上下文") await self.disconnect() async def connect(self): print("连接资源...") async def disconnect(self): print("断开连接...")# 使用异步上下文管理器async def use_async_context(): async with AsyncContextManager("异步资源") as resource: print(f"使用资源: {resource}")# 运行import asyncioasyncio.run(use_async_context())asynccontextmanager 装饰器from contextlib import asynccontextmanager@asynccontextmanagerasync def async_resource(): print("获取异步资源") try: yield "异步资源" finally: print("释放异步资源")async def use_async_resource(): async with async_resource() as resource: print(f"使用资源: {resource}")asyncio.run(use_async_resource())最佳实践1. 确保资源清理# 好的做法 - 使用 finally 确保清理class ResourceManager: def __enter__(self): self.resource = acquire_resource() return self.resource def __exit__(self, exc_type, exc_val, exc_tb): release_resource(self.resource) return False2. 正确处理异常# 好的做法 - 区分异常类型class SafeContextManager: def __exit__(self, exc_type, exc_val, exc_tb): if exc_type is None: print("正常退出") elif issubclass(exc_type, (ValueError, TypeError)): print(f"处理预期异常: {exc_val}") return True # 抑制预期异常 else: print(f"未处理异常: {exc_val}") return False # 传播未处理异常3. 提供有用的错误信息class DatabaseContextManager: def __enter__(self): try: self.connection = connect_to_database() return self.connection except ConnectionError as e: raise RuntimeError(f"无法连接到数据库: {e}") def __exit__(self, exc_type, exc_val, exc_tb): if self.connection: try: self.connection.close() except Exception as e: print(f"关闭连接时出错: {e}")4. 支持上下文管理器链class ChainedContextManager: def __init__(self, *managers): self.managers = managers def __enter__(self): self.entered_managers = [] try: for manager in self.managers: self.entered_managers.append(manager.__enter__()) return self.entered_managers[-1] except: # 如果失败，清理已进入的上下文 self.__exit__(None, None, None) raise def __exit__(self, exc_type, exc_val, exc_tb): # 反向退出上下文 for manager in reversed(self.entered_managers): manager.__exit__(exc_type, exc_val, exc_tb) return False总结Python 上下文管理器的核心概念：基本协议：实现 __enter__ 和 __exit__ 方法with 语句：自动管理资源的进入和退出异常处理：在 __exit__ 中处理异常contextlib 模块：简化上下文管理器的创建实际应用：文件操作、数据库连接、锁管理等嵌套管理：使用多个 with 语句或 ExitStack异步支持：异步上下文管理器用于异步代码上下文管理器的优势：自动资源管理，避免资源泄漏代码更简洁、更易读异常安全，确保资源清理支持嵌套和链式使用掌握上下文管理器，能够编写出更安全、更优雅的 Python 代码。

Python GIL（全局解释器锁）详解什么是 GILGIL（Global Interpreter Lock，全局解释器锁）是 Python 解释器（主要是 CPython）中的一个互斥锁，它确保在任何时候只有一个线程在执行 Python 字节码。这意味着即使在多核 CPU 上，Python 的多线程程序也无法真正实现并行执行。GIL 存在的原因1. 内存管理安全Python 使用引用计数（Reference Counting）来管理内存，每个对象都有一个引用计数器。当引用计数降为 0 时，对象会被自动回收。如果没有 GIL，多个线程同时修改引用计数会导致竞态条件（Race Condition）。2. C 扩展兼容性许多 Python 的 C 扩展库（如 NumPy、Pandas）都不是线程安全的，GIL 保护了这些扩展库的安全性。3. 实现简单性GIL 是一个相对简单的解决方案，避免了复杂的细粒度锁机制。GIL 的工作机制import threadingimport timedef count_down(n): while n > 0: n -= 1# 单线程执行start = time.time()count_down(100000000)print(f"单线程耗时: {time.time() - start:.4f} 秒")# 多线程执行start = time.time()t1 = threading.Thread(target=count_down, args=(50000000,))t2 = threading.Thread(target=count_down, args=(50000000,))t1.start()t2.start()t1.join()t2.join()print(f"多线程耗时: {time.time() - start:.4f} 秒")在 CPU 密集型任务中，多线程可能比单线程更慢，因为存在 GIL 和线程切换的开销。GIL 的影响场景1. CPU 密集型任务（受 GIL 影响大）import threadingimport timedef cpu_bound_task(n): result = 0 for i in range(n): result += i ** 2 return result# 单线程start = time.time()result1 = cpu_bound_task(1000000)result2 = cpu_bound_task(1000000)print(f"单线程结果: {result1 + result2}, 耗时: {time.time() - start:.4f} 秒")# 多线程start = time.time()t1 = threading.Thread(target=lambda: cpu_bound_task(1000000))t2 = threading.Thread(target=lambda: cpu_bound_task(1000000))t1.start()t2.start()t1.join()t2.join()print(f"多线程耗时: {time.time() - start:.4f} 秒")2. I/O 密集型任务（受 GIL 影响小）import threadingimport timeimport requestsdef download_url(url): response = requests.get(url) return len(response.content)urls = [ "https://www.example.com", "https://www.google.com", "https://www.github.com",]# 单线程start = time.time()for url in urls: download_url(url)print(f"单线程耗时: {time.time() - start:.4f} 秒")# 多线程start = time.time()threads = [threading.Thread(target=download_url, args=(url,)) for url in urls]for t in threads: t.start()for t in threads: t.join()print(f"多线程耗时: {time.time() - start:.4f} 秒")在 I/O 密集型任务中，多线程可以显著提高性能，因为线程在等待 I/O 时会释放 GIL。绕过 GIL 的方法1. 使用多进程（multiprocessing）import multiprocessingimport timedef cpu_bound_task(n): result = 0 for i in range(n): result += i ** 2 return resultif __name__ == '__main__': # 单进程 start = time.time() result1 = cpu_bound_task(1000000) result2 = cpu_bound_task(1000000) print(f"单进程耗时: {time.time() - start:.4f} 秒") # 多进程 start = time.time() pool = multiprocessing.Pool(processes=2) results = pool.map(cpu_bound_task, [1000000, 1000000]) pool.close() pool.join() print(f"多进程耗时: {time.time() - start:.4f} 秒")多进程每个进程有独立的 Python 解释器和 GIL，可以真正实现并行计算。2. 使用异步编程（asyncio）import asyncioimport aiohttpimport timeasync def fetch_url(session, url): async with session.get(url) as response: return await response.text()async def main(urls): async with aiohttp.ClientSession() as session: tasks = [fetch_url(session, url) for url in urls] return await asyncio.gather(*tasks)urls = [ "https://www.example.com", "https://www.google.com", "https://www.github.com",]start = time.time()asyncio.run(main(urls))print(f"异步耗时: {time.time() - start:.4f} 秒")3. 使用 C 扩展或 Cython# 使用 Cython 编写的模块# mymodule.pyxdef fast_function(int n): cdef int i cdef int result = 0 for i in range(n): result += i * i return resultCython 代码可以释放 GIL，实现真正的并行计算。4. 使用 NumPy 等优化库import numpy as npimport time# NumPy 内部操作会释放 GILarr1 = np.random.rand(1000000)arr2 = np.random.rand(1000000)start = time.time()result = np.dot(arr1, arr2)print(f"NumPy 耗时: {time.time() - start:.4f} 秒")GIL 的释放时机Python 解释器在以下情况会释放 GIL：I/O 操作：文件读写、网络请求等时间片到期：默认每 1000 个字节码指令检查一次显式释放：某些 C 扩展可以手动释放 GIL长时间操作：某些长时间运行的操作会释放 GILimport threadingimport timedef test_gil_release(): print(f"线程 {threading.current_thread().name} 开始") time.sleep(1) # I/O 操作，释放 GIL print(f"线程 {threading.current_thread().name} 结束")t1 = threading.Thread(target=test_gil_release, name="Thread-1")t2 = threading.Thread(target=test_gil_release, name="Thread-2")t1.start()t2.start()t1.join()t2.join()不同 Python 实现的 GILCPython：有 GILJython：无 GIL（基于 JVM）IronPython：无 GIL（基于 .NET）PyPy：有 GIL，但性能更好Stackless Python：有 GIL，但支持微线程性能优化建议1. 选择合适的并发模型# CPU 密集型：使用多进程from multiprocessing import Pooldef process_data(data): return sum(x * x for x in data)with Pool(4) as pool: results = pool.map(process_data, data_chunks)2. I/O 密集型：使用多线程或异步# 多线程import threadingdef io_task(url): # I/O 操作 passthreads = [threading.Thread(target=io_task, args=(url,)) for url in urls]for t in threads: t.start()for t in threads: t.join()# 或使用异步import asyncioasync def async_io_task(url): # 异步 I/O 操作 passasync def main(): await asyncio.gather(*[async_io_task(url) for url in urls])asyncio.run(main())3. 混合使用from multiprocessing import Poolimport threadingdef worker(data_chunk): # 每个进程内部可以使用线程处理 I/O results = [] threads = [] for item in data_chunk: t = threading.Thread(target=process_item, args=(item, results)) threads.append(t) t.start() for t in threads: t.join() return resultswith Pool(4) as pool: results = pool.map(worker, data_chunks)总结GIL 的优点简化内存管理，避免复杂的锁机制保护 C 扩展的线程安全单线程性能优秀GIL 的缺点限制了多线程在 CPU 密集型任务中的性能无法充分利用多核 CPU在某些场景下性能不如其他语言最佳实践I/O 密集型：使用多线程或异步编程CPU 密集型：使用多进程或考虑其他语言混合型：结合多进程和多线程性能关键：使用 Cython、NumPy 等优化工具理解 GIL 的工作原理和影响，有助于选择合适的并发策略，编写高效的 Python 程序。

TCP 三次握手详解TCP 三次握手是建立可靠连接的关键过程，确保双方都准备好进行数据传输。握手过程第一次握手（SYN）：客户端发送 SYN=1、seq=x 的报文段，进入 SYN_SENT 状态，等待服务器确认第二次握手（SYN+ACK）：服务器收到 SYN 后，发送 SYN=1、ACK=1、seq=y、ack=x+1 的报文段，进入 SYN_RCVD 状态第三次握手（ACK）：客户端收到 SYN+ACK 后，发送 ACK=1、seq=x+1、ack=y+1 的报文段，双方进入 ESTABLISHED 状态为什么需要三次握手防止已失效的连接请求突然又传到服务器：如果是两次握手，客户端发送的失效连接请求到达服务器，服务器误认为是新连接，造成资源浪费同步双方的初始序列号：三次握手确保双方都知道对方的初始序列号，为可靠传输奠定基础确认双方的接收和发送能力：通过三次交互，双方都能确认对方的收发功能正常相关问题三次握手过程中，如果第三次握手丢失会发生什么？SYN Flood 攻击是如何利用三次握手的？如何防御？为什么不是两次握手或四次握手？