武夷山市建设局网站,高明建网站服务,网站建设加推广需要多少钱,网站建设工具的种类Python高级教程
引言#xff1a;走进Python的高级世界
Python作为一门简洁优雅的编程语言#xff0c;其魅力不仅在于易学易用#xff0c;更在于它强大的高级特性。掌握这些高级特性#xff0c;能让你的代码更加优雅、高效和强大。本文将带你深入探索Python的高级功能#…Python高级教程引言走进Python的高级世界Python作为一门简洁优雅的编程语言其魅力不仅在于易学易用更在于它强大的高级特性。掌握这些高级特性能让你的代码更加优雅、高效和强大。本文将带你深入探索Python的高级功能从装饰器到元编程从并发到元类开启Python编程的新境界。第一章函数式编程的艺术1.1 高阶函数函数作为一等公民在Python中函数是一等公民可以像其他数据类型一样传递、赋值和返回。# 函数作为参数defapply_function(func,data):return[func(item)foritemindata]defsquare(x):returnx**2numbers[1,2,3,4,5]resultapply_function(square,numbers)print(result)# [1, 4, 9, 16, 25]# 函数作为返回值defcreate_multiplier(factor):defmultiplier(x):returnx*factorreturnmultiplier doublecreate_multiplier(2)triplecreate_multiplier(3)print(double(5))# 10print(triple(5))# 151.2 Lambda表达式与函数式工具fromfunctoolsimportreduce# Lambda表达式addlambdax,y:xyprint(add(3,4))# 7# map, filter, reducenumbers[1,2,3,4,5,6,7,8,9,10]# map: 对每个元素应用函数squaredlist(map(lambdax:x**2,numbers))print(squared)# filter: 过滤元素even_numberslist(filter(lambdax:x%20,numbers))print(even_numbers)# reduce: 累积计算productreduce(lambdax,y:x*y,numbers)print(product)# 3628800# 列表推导式的威力# 传统方式squares[]forxinrange(10):squares.append(x**2)# 列表推导式squares[x**2forxinrange(10)]# 带条件的列表推导式even_squares[x**2forxinrange(10)ifx%20]# 字典推导式square_dict{x:x**2forxinrange(5)}print(square_dict)# {0: 0, 1: 1, 2: 4, 3: 9, 4: 16}# 集合推导式unique_squares{x**2forxin[-2,-1,0,1,2]}print(unique_squares)# {0, 1, 4}1.3 闭包与装饰器闭包是Python中一个强大的概念它允许函数记住其创建时的环境。# 闭包示例defmake_counter():count0defcounter():nonlocalcount# 使用nonlocal修改外层变量count1returncountreturncounter counter1make_counter()counter2make_counter()print(counter1())# 1print(counter1())# 2print(counter2())# 1print(counter1())# 3# 装饰器基础defmy_decorator(func):defwrapper():print(函数执行前)func()print(函数执行后)returnwrappermy_decoratordefsay_hello():print(Hello!)say_hello()# 输出:# 函数执行前# Hello!# 函数执行后# 带参数的装饰器defrepeat(num_times):defdecorator_repeat(func):defwrapper(*args,**kwargs):for_inrange(num_times):resultfunc(*args,**kwargs)returnresultreturnwrapperreturndecorator_repeatrepeat(num_times3)defgreet(name):print(fHello{name})greet(World)# 输出:# Hello World# Hello World# Hello World# 类装饰器classTimer:def__init__(self,func):self.funcfunc self.times[]def__call__(self,*args,**kwargs):importtime starttime.time()resultself.func(*args,**kwargs)endtime.time()self.times.append(end-start)print(f函数{self.func.__name__}执行时间:{end-start:.4f}秒)returnresultTimerdefslow_function():importtime time.sleep(1)return完成slow_function()第二章面向对象编程的高级特性2.1 魔法方法与运算符重载Python的魔法方法双下划线方法允许我们自定义类的行为。classVector:自定义向量类演示运算符重载def__init__(self,x0,y0):self.xx self.yy# 字符串表示def__repr__(self):returnfVector({self.x},{self.y})def__str__(self):returnf({self.x},{self.y})# 运算符重载def__add__(self,other):returnVector(self.xother.x,self.yother.y)def__sub__(self,other):returnVector(self.x-other.x,self.y-other.y)def__mul__(self,scalar):returnVector(self.x*scalar,self.y*scalar)def__rmul__(self,scalar):returnself.__mul__(scalar)# 比较运算符def__eq__(self,other):returnself.xother.xandself.yother.ydef__lt__(self,other):returnself.magnitude()other.magnitude()# 容器行为def__getitem__(self,index):ifindex0:returnself.xelifindex1:returnself.yelse:raiseIndexError(向量索引只能是0或1)def__len__(self):return2# 实用方法defmagnitude(self):return(self.x**2self.y**2)**0.5defdot(self,other):returnself.x*other.xself.y*other.y# 使用示例v1Vector(3,4)v2Vector(1,2)print(v1)# (3, 4)print(v1v2)# (4, 6)print(v1-v2)# (2, 2)print(v1*2)# (6, 8)print(3*v1)# (9, 12)print(v1v2)# Falseprint(v1[0])# 3print(len(v1))# 2print(v1.magnitude())# 5.02.2 属性描述符与property# property装饰器classCircle:def__init__(self,radius):self._radiusradiuspropertydefradius(self):半径的getterreturnself._radiusradius.setterdefradius(self,value):半径的setterifvalue0:raiseValueError(半径不能为负数)self._radiusvaluepropertydefdiameter(self):直径只读属性return2*self._radiuspropertydefarea(self):面积只读属性return3.14159*self._radius**2circleCircle(5)print(f半径:{circle.radius})# 5print(f直径:{circle.diameter})# 10print(f面积:{circle.area:.2f})# 78.54circle.radius10print(f新面积:{circle.area:.2f})# 314.16# 自定义描述符classPositiveNumber:确保值为正数的描述符def__init__(self,name):self.namenamedef__get__(self,obj,objtypeNone):ifobjisNone:returnselfreturnobj.__dict__.get(self.name,0)def__set__(self,obj,value):ifvalue0:raiseValueError(f{self.name}不能为负数)obj.__dict__[self.name]valueclassRectangle:# 使用自定义描述符widthPositiveNumber(width)heightPositiveNumber(height)def__init__(self,width,height):self.widthwidth self.heightheightpropertydefarea(self):returnself.width*self.height rectRectangle(10,5)print(f面积:{rect.area})# 50try:rect.width-5exceptValueErrorase:print(f错误:{e})2.3 多重继承与MRO# 多重继承示例classAnimal:def__init__(self,name):self.namenamedefspeak(self):return...classMammal(Animal):def__init__(self,name):super().__init__(name)self.has_furTruedefgive_birth(self):returnf{self.name}生下了幼崽classBird(Animal):def__init__(self,name):super().__init__(name)self.has_feathersTruedeflay_eggs(self):returnf{self.name}下了蛋classPlatypus(Mammal,Bird):鸭嘴兽 - 既是哺乳动物又是鸟类的特征def__init__(self,name):# 使用super()正确初始化super().__init__(name)self.is_venomousTruedefspeak(self):return嘎嘎# 方法解析顺序classmethoddefshow_mro(cls):return[c.__name__forcincls.__mro__]perryPlatypus(佩里)print(perry.name)# 佩里print(perry.speak())# 嘎嘎print(perry.give_birth())# 佩里 生下了幼崽print(perry.lay_eggs())# 佩里 下了蛋print(f有皮毛:{perry.has_fur})# Trueprint(f有羽毛:{perry.has_feathers})# Trueprint(f有毒:{perry.is_venomous})# Trueprint(MRO:,Platypus.show_mro())# [Platypus, Mammal, Bird, Animal, object]2.4 抽象基类与接口fromabcimportABC,abstractmethodfromcollections.abcimportSequence# 抽象基类classDrawable(ABC):可绘制对象的抽象基类abstractmethoddefdraw(self):passabstractmethoddefarea(self):passclassShape(Drawable):形状基类def__init__(self,colorblack):self.colorcolordefdraw(self):returnf用{self.color}颜色绘制{self.__class__.__name__}classCircle(Shape):def__init__(self,radius,colorblack):super().__init__(color)self.radiusradiusdefarea(self):return3.14159*self.radius**2classRectangle(Shape):def__init__(self,width,height,colorblack):super().__init__(color)self.widthwidth self.heightheightdefarea(self):returnself.width*self.height# 使用shapes[Circle(5,red),Rectangle(4,6,blue)]forshapeinshapes:print(shape.draw())print(f面积:{shape.area():.2f})# 注册虚拟子类classTriangle:def__init__(self,base,height,colorgreen):self.basebase self.heightheight self.colorcolordefdraw(self):returnf用{self.color}颜色绘制三角形defarea(self):return0.5*self.base*self.height# 注册为Drawable的虚拟子类Drawable.register(Triangle)# 现在Triangle可以被视为DrawabletriangleTriangle(4,3)print(isinstance(triangle,Drawable))# True# 检查对象是否符合某个接口classStack:def__init__(self):self._items[]defpush(self,item):self._items.append(item)defpop(self):ifnotself._items:raiseIndexError(栈为空)returnself._items.pop()def__len__(self):returnlen(self._items)stackStack()print(fStack是Sequence吗?{isinstance(stack,Sequence)})# Falseprint(fStack有__len__吗?{hasattr(stack,__len__)})# True第三章并发与异步编程3.1 生成器与协程进阶# 生成器作为协程defcoroutine_example():简单的协程示例print(协程开始)whileTrue:valueyieldprint(f收到值:{value})# 使用协程corocoroutine_example()next(coro)# 启动协程coro.send(10)# 收到值: 10coro.send(20)# 收到值: 20coro.close()# 关闭协程# 带返回值的协程defaverager():计算平均值的协程total0count0averageNonewhileTrue:valueyieldaverage totalvalue count1averagetotal/count avgaverager()next(avg)# 启动print(avg.send(10))# 10.0print(avg.send(20))# 15.0print(avg.send(30))# 20.0# yield from 语法defchain_generators():连接多个生成器yieldfromrange(5)yieldfromabcyieldfrom[True,False]print(list(chain_generators()))# [0, 1, 2, 3, 4, a, b, c, True, False]# 使用yield from实现协程委托defsub_coroutine():子协程resultyieldfromrange(3)returnresultdefdelegator():委托协程resultyieldfromsub_coroutine()yieldf子协程完成结果:{result}print(list(delegator()))# [0, 1, 2, 子协程完成结果: None]3.2 asyncio高级用法importasyncioimportaiohttpimporttimefromconcurrent.futuresimportThreadPoolExecutor# 异步上下文管理器classAsyncTimer:异步计时器上下文管理器asyncdef__aenter__(self):self.starttime.time()returnselfasyncdef__aexit__(self,exc_type,exc_val,exc_tb):self.endtime.time()print(f耗时:{self.end-self.start:.2f}秒)defelapsed(self):returntime.time()-self.startasyncdeftask_with_timer(name,delay):asyncwithAsyncTimer()astimer:print(f任务{name}开始)awaitasyncio.sleep(delay)print(f任务{name}完成已用时{timer.elapsed():.2f}秒)# 异步迭代器classAsyncCounter:异步计数器def__init__(self,limit):self.limitlimit self.current0def__aiter__(self):returnselfasyncdef__anext__(self):ifself.currentself.limit:awaitasyncio.sleep(0.5)# 模拟异步操作self.current1returnself.currentelse:raiseStopAsyncIterationasyncdefuse_async_iterator():使用异步迭代器asyncfornumberinAsyncCounter(5):print(f计数:{number})# 异步生成器asyncdefasync_generator(n):异步生成器foriinrange(n):awaitasyncio.sleep(0.3)# 模拟异步操作yieldiasyncdefuse_async_generator():使用异步生成器asyncforiteminasync_generator(5):print(f生成项:{item})# 异步队列asyncdefproducer(queue,name,items):生产者foriinrange(items):itemf{name}-{i}awaitqueue.put(item)print(f生产者{name}生产了{item})awaitasyncio.sleep(0.2)asyncdefconsumer(queue,name):消费者whileTrue:try:# 设置超时避免永远等待itemawaitasyncio.wait_for(queue.get(),timeout3)print(f消费者{name}消费了{item})queue.task_done()exceptasyncio.TimeoutError:print(f消费者{name}超时退出)breakasyncdefqueue_demo():队列演示queueasyncio.Queue(maxsize5)# 创建生产者和消费者任务producers[asyncio.create_task(producer(queue,fP{i},3))foriinrange(2)]consumers[asyncio.create_task(consumer(queue,fC{i}))foriinrange(3)]# 等待所有生产者完成awaitasyncio.gather(*producers)# 等待队列清空awaitqueue.join()# 取消消费者forcinconsumers:c.cancel()# 运行所有示例asyncdefmain():print( 异步上下文管理器示例 )awaittask_with_timer(示例任务,1)print(\n 异步迭代器示例 )awaituse_async_iterator()print(\n 异步生成器示例 )awaituse_async_generator()print(\n 异步队列示例 )awaitqueue_demo()# 在Jupyter或支持asyncio的环境中运行# asyncio.run(main())3.3 多线程与多进程高级用法importconcurrent.futuresimportmultiprocessingimportthreadingimportqueueimporttime# 线程池高级用法defparallel_map_example():使用ThreadPoolExecutor进行并行映射defprocess_item(item):time.sleep(0.5)# 模拟耗时操作returnitem**2datalist(range(10))# 方法1: 使用mapwithconcurrent.futures.ThreadPoolExecutor(max_workers3)asexecutor:resultslist(executor.map(process_item,data))print(f使用map的结果:{results})# 方法2: 使用submit和as_completed可以处理异常withconcurrent.futures.ThreadPoolExecutor(max_workers3)asexecutor:future_to_item{executor.submit(process_item,item):itemforitemindata}forfutureinconcurrent.futures.as_completed(future_to_item):itemfuture_to_item[future]try:resultfuture.result()print(f项目{item}的结果:{result})exceptExceptionase:print(f项目{item}生成异常:{e})# 进程间通信defprocess_communication():多进程通信示例defproducer(queue,data):生产者进程foritemindata:queue.put(item)print(f生产者放入:{item})time.sleep(0.1)queue.put(None)# 结束信号defconsumer(queue):消费者进程whileTrue:itemqueue.get()ifitemisNone:breakprint(f消费者取出:{item})time.sleep(0.2)# 使用Manager创建共享队列withmultiprocessing.Manager()asmanager:shared_queuemanager.Queue(maxsize5)datalist(range(10))# 创建进程prod_processmultiprocessing.Process(targetproducer,args(shared_queue,data))cons_processmultiprocessing.Process(targetconsumer,args(shared_queue,))# 启动进程prod_process.start()cons_process.start()# 等待进程完成prod_process.join()cons_process.join()print(所有进程完成)# 线程同步高级技巧classReadWriteLock:读写锁实现def__init__(self):self._read_readythreading.Condition(threading.Lock())self._readers0defacquire_read(self):获取读锁withself._read_ready:self._readers1defrelease_read(self):释放读锁withself._read_ready:self._readers-1ifself._readers0:self._read_ready.notify_all()defacquire_write(self):获取写锁self._read_ready.acquire()whileself._readers0:self._read_ready.wait()defrelease_write(self):释放写锁self._read_ready.release()# 使用读写锁defrw_lock_example():读写锁使用示例rw_lockReadWriteLock()shared_data{counter:0}defreader(id):读者线程for_inrange(3):rw_lock.acquire_read()try:valueshared_data[counter]print(f读者{id}读取到:{value})finally:rw_lock.release_read()time.sleep(0.1)defwriter(id):写者线程for_inrange(2):rw_lock.acquire_write()try:shared_data[counter]1print(f写者{id}写入:{shared_data[counter]})finally:rw_lock.release_write()time.sleep(0.2)# 创建线程readers[threading.Thread(targetreader,args(i,))foriinrange(3)]writers[threading.Thread(targetwriter,args(i,))foriinrange(2)]# 启动线程fortinreaderswriters:t.start()# 等待线程完成fortinreaderswriters:t.join()print(f最终结果:{shared_data[counter]})# 运行示例if__name____main__:print( 并行映射示例 )parallel_map_example()print(\n 进程通信示例 )process_communication()print(\n 读写锁示例 )rw_lock_example()第四章元编程与反射4.1 元类编程# 基础元类示例classSingletonMeta(type):单例模式元类_instances{}def__call__(cls,*args,**kwargs):ifclsnotincls._instances:cls._instances[cls]super().__call__(*args,**kwargs)returncls._instances[cls]classSingletonClass(metaclassSingletonMeta):使用单例元类的类def__init__(self,value):self.valuevaluedef__repr__(self):returnfSingletonClass(value{self.value})# 测试单例obj1SingletonClass(10)obj2SingletonClass(20)print(fobj1 is obj2:{obj1isobj2})# Trueprint(fobj1:{obj1})# SingletonClass(value10)print(fobj2:{obj2})# SingletonClass(value10)# 验证器元类classValidatedMeta(type):验证属性的元类def__new__(cls,name,bases,namespace):# 收集所有验证器validators{}forkey,valueinnamespace.items():ifkey.startswith(validate_):attr_namekey[9:]# 去掉validate_前缀validators[attr_name]value# 创建__setattr__方法def__setattr__(self,name,value):# 如果有验证器先验证ifnameinvalidators:validators[name](self,value)# 调用父类的__setattr__super(self.__class__,self).__setattr__(name,value)namespace[__setattr__]__setattr__returnsuper().__new__(cls,name,bases,namespace)classValidatedClass(metaclassValidatedMeta):使用验证器元类的类defvalidate_age(self,value):ifnotisinstance(value,int):raiseTypeError(年龄必须是整数)ifvalue0orvalue150:raiseValueError(年龄必须在0-150之间)defvalidate_name(self,value):ifnotisinstance(value,str):raiseTypeError(姓名必须是字符串)iflen(value)2:raiseValueError(姓名至少2个字符)personValidatedClass()person.name张三# 正常person.age25# 正常try:person.age200# 会抛出异常exceptValueErrorase:print(f错误:{e})try:person.nameA# 会抛出异常exceptValueErrorase:print(f错误:{e})# 自动注册子类的元类classPluginMeta(type):自动注册插件的元类plugins[]def__new__(cls,name,bases,namespace):new_classsuper().__new__(cls,name,bases,namespace)# 如果不是基类则注册ifname!PluginBase:cls.plugins.append(new_class)returnnew_classclassPluginBase(metaclassPluginMeta):插件基类defexecute(self):raiseNotImplementedErrorclassPluginA(PluginBase):defexecute(self):return执行插件AclassPluginB(PluginBase):defexecute(self):return执行插件Bprint(f注册的插件:{[p.__name__forpinPluginMeta.plugins]})# 输出: [PluginA, PluginB]# 使用注册的插件forplugin_classinPluginMeta.plugins:pluginplugin_class()print(plugin.execute())4.2 动态创建与修改类# 使用type动态创建类def__init__(self,name,age):self.namename self.ageagedefsay_hello(self):returnf你好我是{self.name}今年{self.age}岁# 动态创建Person类Persontype(Person,# 类名(),# 基类{# 属性字典__init__:__init__,say_hello:say_hello,species:人类})pPerson(李四,30)print(p.say_hello())# 你好我是李四今年30岁print(p.species)# 人类# 动态添加方法defnew_method(self):return这是一个动态添加的方法# 给现有类添加方法Person.greetnew_methodprint(p.greet())# 这是一个动态添加的方法# 使用types.MethodType绑定方法importtypesdefinstance_method(self):returnf实例方法name{self.name}p.special_methodtypes.MethodType(instance_method,p)print(p.special_method())# 实例方法name李四# 动态修改类classOriginalClass:deforiginal_method(self):return原始方法# 修改类的方法defnew_original_method(self):return修改后的方法OriginalClass.original_methodnew_original_method objOriginalClass()print(obj.original_method())# 修改后的方法# 使用__dict__查看类的内部print(\nPerson类的属性和方法:)forkey,valueinPerson.__dict__.items():ifnotkey.startswith(__):print(f{key}:{value})4.3 反射与内省classMyClass:演示反射的类def__init__(self,value):self.valuevaluedefpublic_method(self):returnf公共方法:{self.value}def_private_method(self):return私有方法classmethoddefclass_method(cls):return类方法staticmethoddefstatic_method():return静态方法# 创建实例objMyClass(42)# 1. 检查类型print(f类型检查:{type(obj)})# class __main__.MyClassprint(f实例检查:{isinstance(obj,MyClass)})# Trueprint(f子类检查:{issubclass(MyClass,object)})# True# 2. 获取属性print(f\n获取属性value:{getattr(obj,value,默认值)})# 42print(f获取不存在的属性:{getattr(obj,nonexistent,默认值)})# 默认值# 3. 设置属性setattr(obj,new_attribute,新值)print(f新属性:{obj.new_attribute})# 新值# 4. 检查属性和方法print(f\n是否有value属性:{hasattr(obj,value)})# Trueprint(f是否有public_method:{hasattr(obj,public_method)})# Trueprint(f是否有nonexistent:{hasattr(obj,nonexistent)})# False# 5. 获取所有属性和方法print(\n所有属性和方法:)forattr_nameindir(obj):ifnotattr_name.startswith(__):attrgetattr(obj,attr_name)ifcallable(attr):print(f 方法:{attr_name})else:print(f 属性:{attr_name}{attr})# 6. 调用方法methodgetattr(obj,public_method)print(f\n通过反射调用方法:{method()})# 公共方法: 42# 7. 检查方法类型importinspectprint(\n方法签名检查:)forname,methodininspect.getmembers(obj,predicateinspect.ismethod):print(f 方法:{name}, 类型:{type(method).__name__})# 8. 获取源代码如果可用print(\n获取源代码:)try:sourceinspect.getsource(MyClass)print(f类定义:\n{source[:200]}...)except:print(无法获取源代码)# 9. 动态创建和调用函数defadd(a,b):returnab# 使用exec动态创建函数func_code def multiply(x, y): return x * y namespace{}exec(func_code,namespace)multiply_funcnamespace[multiply]print(f\n动态创建的函数:{multiply_func(3,4)})# 12# 10. 使用__getattribute__和__getattr__classCustomAccess:自定义属性访问def__init__(self):self._data{}def__getattr__(self,name):当属性不存在时调用print(f__getattr__被调用:{name})ifnameinself._data:returnself._data[name]raiseAttributeError(fCustomAccess object has no attribute {name})def__getattribute__(self,name):所有属性访问都会调用此方法print(f__getattribute__被调用:{name})# 注意不能直接访问self._data否则会造成无限递归returnsuper().__getattribute__(name)def__setattr__(self,name,value):设置属性时调用print(f__setattr__被调用:{name}{value})ifname_data:super().__setattr__(name,value)else:self._data[name]value objCustomAccess()obj.x10# __setattr__被调用: x 10print(f访问x:{obj.x})# __getattribute__被调用: x → __getattr__被调用: x → 10print(f访问不存在的y:{getattr(obj,y,默认)})# __getattribute__被调用: y → __getattr__被调用: y → 默认第五章性能优化与高级技巧5.1 内存视图与缓冲区协议importarrayimportnumpyasnp# 使用array创建连续内存arrarray.array(i,range(10))# i表示有符号整数print(f原始数组:{arr})# 创建内存视图mem_viewmemoryview(arr)print(f内存视图长度:{len(mem_view)})print(f内存视图项目大小:{mem_view.itemsize}字节)print(f内存视图格式:{mem_view.format})# 修改内存视图会影响原数组mem_view[5]999print(f修改后数组:{arr})# 第6个元素变为999# 切片内存视图不会复制数据slice_viewmem_view[3:7]print(f切片视图:{list(slice_view)})slice_view[0]888print(f修改切片后数组:{arr})# 第4个元素变为888# 与NumPy数组互操作np_arraynp.array([1,2,3,4,5],dtypenp.float32)print(f\nNumPy数组:{np_array})# NumPy数组也支持缓冲区协议np_viewmemoryview(np_array)print(fNumPy内存视图格式:{np_view.format})# 使用tobytes和frombufferbytes_dataarr.tobytes()print(f\n数组字节表示:{bytes_data})new_arrarray.array(i)new_arr.frombytes(bytes_data)print(f从字节恢复的数组:{new_arr})# 大型数据处理示例defprocess_large_data():处理大型数据的示例# 创建大型数组large_arrayarray.array(f,[i*0.1foriinrange(1000000)])# 使用内存视图处理数据避免复制viewmemoryview(large_array)# 批量处理数据chunk_size10000results[]foriinrange(0,len(view),chunk_size):chunkview[i:ichunk_size]# 处理数据块这里简单计算平均值chunk_arrayarray.array(f,chunk)avgsum(chunk_array)/len(chunk_array)results.append(avg)print(f处理了{len(large_array)}个元素)print(f分块平均值:{results[:5]}...)# 显示前5个结果process_large_data()5.2 描述符与属性缓存importfunctoolsimporttime# 属性缓存描述符classCachedProperty:缓存属性值的描述符def__init__(self,func):self.funcfunc self.namefunc.__name__def__get__(self,obj,objtypeNone):ifobjisNone:returnself# 检查缓存cacheobj.__dict__.get(_cache,{})ifself.nameincache:returncache[self.name]# 计算并缓存结果valueself.func(obj)if_cachenotinobj.__dict__:obj.__dict__[_cache]{}obj.__dict__[_cache][self.name]valuereturnvalueclassExpensiveComputation:演示缓存属性的类def__init__(self,n):self.nnCachedPropertydeffactorial(self):计算阶乘耗时操作print(f计算{self.n}!...)time.sleep(1)# 模拟耗时计算result1foriinrange(1,self.n1):result*ireturnresultCachedPropertydeffibonacci(self):计算斐波那契数列耗时操作print(f计算斐波那契({self.n})...)time.sleep(1)# 模拟耗时计算deffib(n):ifn1:returnn a,b0,1for_inrange(n-1):a,bb,abreturnbreturnfib(self.n)# 测试缓存属性print( 缓存属性测试 )objExpensiveComputation(10)print(\n第一次访问属性会计算:)starttime.time()print(f阶乘:{obj.factorial})print(f耗时:{time.time()-start:.2f}秒)print(\n第二次访问属性从缓存读取:)starttime.time()print(f阶乘:{obj.factorial})print(f耗时:{time.time()-start:.4f}秒)print(\n访问另一个属性:)starttime.time()print(f斐波那契:{obj.fibonacci})print(f耗时:{time.time()-start:.2f}秒)print(\n再次访问斐波那契缓存:)starttime.time()print(f斐波那契:{obj.fibonacci})print(f耗时:{time.time()-start:.4f}秒)# 使用lru_cachefunctools.lru_cache(maxsize128)deffibonacci_cached(n):使用lru_cache缓存的斐波那契函数ifn1:returnnreturnfibonacci_cached(n-1)fibonacci_cached(n-2)print(\n LRU缓存测试 )print(计算斐波那契数列:)foriin[10,20,30,10,20]:# 10和20会重复计算starttime.time()resultfibonacci_cached(i)elapsedtime.time()-startprint(ffib({i}) {result}, 耗时:{elapsed:.4f}秒)print(f\n缓存信息:{fibonacci_cached.cache_info()})# 输出缓存命中率等信息# 清空缓存fibonacci_cached.cache_clear()print(缓存已清空)# 属性观察者模式classObservableProperty:可观察的属性描述符def__init__(self,defaultNone):self.defaultdefault self.observers[]def__set_name__(self,owner,name):self.namenamedef__get__(self,obj,objtypeNone):ifobjisNone:returnselfreturnobj.__dict__.get(self.name,self.default)def__set__(self,obj,value):old_valueobj.__dict__.get(self.name,self.default)obj.__dict__[self.name]value# 通知所有观察者forobserverinself.observers:observer(obj,self.name,old_value,value)defadd_observer(self,callback):添加观察者self.observers.append(callback)classObservableObject:使用可观察属性的类nameObservableProperty(默认名称)valueObservableProperty(0)def__init__(self):# 添加观察者self.name.add_observer(self.on_property_change)self.value.add_observer(self.on_property_change)defon_property_change(self,obj,prop_name,old_value,new_value):print(f属性{prop_name}从{old_value}变为{new_value})print(\n 可观察属性测试 )objObservableObject()obj.name新名称obj.value42obj.value1005.3 协程与生成器优化importasyncioimporttimefromcollectionsimportdeque# 使用async for处理数据流classAsyncDataStream:异步数据流def__init__(self):self._datadeque()self._eventasyncio.Event()asyncdefput(self,item):放入数据self._data.append(item)self._event.set()asyncdefget(self):获取数据whilenotself._data:awaitself._event.wait()itemself._data.popleft()ifnotself._data:self._event.clear()returnitemdef__aiter__(self):returnselfasyncdef__anext__(self):try:returnawaitself.get()exceptException:raiseStopAsyncIterationasyncdefproducer(stream,name,count):生产者协程foriinrange(count):itemf{name}-{i}awaitstream.put(item)awaitasyncio.sleep(0.1)# 模拟生产延迟awaitstream.put(None)# 结束信号asyncdefconsumer(stream,name):消费者协程asyncforiteminstream:ifitemisNone:# 将结束信号放回供其他消费者使用awaitstream.put(None)breakprint(f{name}收到:{item})awaitasyncio.sleep(0.2)# 模拟处理延迟asyncdefstream_example():异步数据流示例streamAsyncDataStream()# 创建任务producers[asyncio.create_task(producer(stream,f生产者{i},3))foriinrange(2)]consumers[asyncio.create_task(consumer(stream,f消费者{i}))foriinrange(3)]# 等待生产者完成awaitasyncio.gather(*producers)# 等待消费者处理完所有数据whilestream._data:awaitasyncio.sleep(0.1)# 取消消费者forcinconsumers:c.cancel()# 异步上下文管理器与资源管理classAsyncConnection:异步数据库连接模拟def__init__(self,name):self.namename self.is_connectedFalseasyncdefconnect(self):连接print(f{self.name}: 连接中...)awaitasyncio.sleep(0.5)self.is_connectedTrueprint(f{self.name}: 已连接)asyncdefdisconnect(self):断开连接print(f{self.name}: 断开连接...)awaitasyncio.sleep(0.2)self.is_connectedFalseprint(f{self.name}: 已断开)asyncdefexecute(self,query):执行查询ifnotself.is_connected:raiseConnectionError(未连接)print(f{self.name}: 执行查询:{query})awaitasyncio.sleep(0.3)returnf{query}的结果asyncdef__aenter__(self):awaitself.connect()returnselfasyncdef__aexit__(self,exc_type,exc_val,exc_tb):awaitself.disconnect()ifexc_type:print(f{self.name}: 发生异常:{exc_type.__name__})asyncdefconnection_pool_example():连接池示例fromcontextlibimportasynccontextmanagerasynccontextmanagerasyncdefget_connection(name):获取连接的上下文管理器connAsyncConnection(name)try:awaitconn.connect()yieldconnfinally:awaitconn.disconnect()# 使用连接asyncwithget_connection(DB1)asconn:resultawaitconn.execute(SELECT * FROM users)print(f结果:{result})# 并行使用多个连接asyncdefuse_connection(name,query):asyncwithget_connection(name)asconn:returnawaitconn.execute(query)queries[SELECT * FROM table1,SELECT * FROM table2,SELECT * FROM table3]tasks[use_connection(fDB{i},query)fori,queryinenumerate(queries,1)]resultsawaitasyncio.gather(*tasks)print(f\n所有查询结果:{results})# 运行示例asyncdefperformance_examples():print( 异步数据流示例 )awaitstream_example()print(\n 异步连接池示例 )awaitconnection_pool_example()# 在支持的环境中运行# asyncio.run(performance_examples())第六章扩展Python6.1 C扩展与ctypes# 虽然Python是解释型语言但可以通过C扩展提升性能# 这里展示如何使用ctypes调用C函数importctypesimportos# 假设我们有一个C库编译为libmylib.soLinux或mylib.dllWindows# 加载C库示例这里使用系统数学库作为示例try:# Linux/Unix系统libmctypes.CDLL(libm.so.6)except:try:# Windows系统libmctypes.CDLL(msvcrt.dll)except:# macOS系统libmctypes.CDLL(libSystem.dylib)# 调用C函数# 计算平方根libm.sqrt.argtypes[ctypes.c_double]libm.sqrt.restypectypes.c_double resultlibm.sqrt(25.0)print(fC sqrt(25) {result})# 5.0# 计算正弦值libm.sin.argtypes[ctypes.c_double]libm.sin.restypectypes.c_double resultlibm.sin(3.14159/2)# sin(π/2)print(fC sin(π/2) {result:.6f})# 约等于1.0# 自定义C函数示例# 创建一个简单的C函数并编译 // mymath.c #include math.h double my_hypot(double a, double b) { return sqrt(a*a b*b); } int sum_array(int* arr, int n) { int sum 0; for (int i 0; i n; i) { sum arr[i]; } return sum; } # 编译命令# Linux/Mac: gcc -shared -fPIC -o libmymath.so mymath.c# Windows: gcc -shared -o mymath.dll mymath.c# 加载自定义库defload_custom_lib():加载自定义C库的示例# 在实际使用中需要先编译C代码生成动态库# 这里只是演示如何使用print(\nC扩展示例需要编译C代码后才能运行:)print( 1. 编写C代码如上面的mymath.c 2. 编译为动态库 Linux/Mac: gcc -shared -fPIC -o libmymath.so mymath.c Windows: gcc -shared -o mymath.dll mymath.c 3. 在Python中加载 lib ctypes.CDLL(./libmymath.so) 4. 调用函数 lib.my_hypot.argtypes [ctypes.c_double, ctypes.c_double] lib.my_hypot.restype ctypes.c_double result lib.my_hypot(3.0, 4.0) # 返回5.0 )load_custom_lib()# 使用ctypes创建C数据结构classPoint(ctypes.Structure):C结构体对应的Python类_fields_[(x,ctypes.c_int),(y,ctypes.c_int)]def__repr__(self):returnfPoint(x{self.x}, y{self.y})# 创建结构体实例pPoint(10,20)print(f\nC结构体实例:{p})print(fx{p.x}, y{p.y})# 修改结构体字段p.x100print(f修改后:{p})# 结构体数组points_array(Point*3)()points_array[0]Point(1,2)points_array[1]Point(3,4)points_array[2]Point(5,6)print(f\n结构体数组:)fori,pointinenumerate(points_array):print(f [{i}] {point})# 与C函数交互假设有处理Point数组的C函数# lib.process_points.argtypes [ctypes.POINTER(Point), ctypes.c_int]# lib.process_points.restype None# lib.process_points(points_array, 3)6.2 使用Cython提升性能# Cython示例需要安装Cython: pip install cython # 以下代码保存为fastmath.pyx # cython: language_level3 def prime_count(int n): 计算小于n的质数个数 cdef int count 0 cdef int i, j cdef int is_prime for i in range(2, n): is_prime 1 for j in range(2, int(i**0.5) 1): if i % j 0: is_prime 0 break if is_prime: count 1 return count def fast_sum(long[:] array): 快速求和 cdef long total 0 cdef int i for i in range(array.shape[0]): total array[i] return total # 编译Cython模块defcompile_cython_module():print(\nCython性能优化示例:)print( 1. 安装Cython: pip install cython 2. 创建.pyx文件如fastmath.pyx 3. 创建setup.py from setuptools import setup from Cython.Build import cythonize setup( ext_modulescythonize(fastmath.pyx), compiler_directives{language_level: 3} ) 4. 编译: python setup.py build_ext --inplace 5. 在Python中使用: import fastmath 6. 调用函数: fastmath.prime_count(1000000) 注意Cython代码需要编译后才能运行无法在交互式环境中直接执行。 )compile_cython_module()# 性能对比示例defperformance_comparison():纯Python与优化代码性能对比importtime# 纯Python实现defprime_count_python(n):count0foriinrange(2,n):is_primeTrueforjinrange(2,int(i**0.5)1):ifi%j0:is_primeFalsebreakifis_prime:count1returncount# 优化的Python实现defprime_count_optimized(n):ifn2:return0# 使用埃拉托斯特尼筛法is_prime[True]*n is_prime[0]is_prime[1]Falseforiinrange(2,int(n**0.5)1):ifis_prime[i]:forjinrange(i*i,n,i):is_prime[j]Falsereturnsum(is_prime)# 测试性能test_n100000print(f\n计算小于{test_n}的质数个数:)# 测试优化版本starttime.time()result1prime_count_optimized(test_n)time1time.time()-startprint(f优化版本:{result1}个质数耗时:{time1:.4f}秒)# 测试纯Python版本starttime.time()result2prime_count_python(test_n)time2time.time()-startprint(f纯Python版本:{result2}个质数耗时:{time2:.4f}秒)print(f速度提升:{time2/time1:.2f}倍)performance_comparison()结语持续学习的路径Python的高级特性远不止本文所涵盖的内容。要成为真正的Python高手你需要深入阅读源码阅读标准库和流行框架的源码参与开源项目在真实项目中实践高级特性掌握设计模式了解如何将高级特性与设计模式结合学习相关工具掌握调试器、性能分析器等工具关注语言发展Python在不断进化新版本会带来新特性记住掌握高级特性不是为了炫技而是为了写出更优雅、更高效、更易维护的代码。在实践中不断探索你将发现Python编程的无穷魅力。