Python 进阶技巧与概念
参考视频: Bilibili - Python进阶
1. 闭包 (Closure)
闭包是一种特殊的函数,它由一个嵌套的函数结构组成,其中内层函数可以记住并访问其外层函数的变量,即使外层函数已经执行完毕。
定义: 在一个外层函数中定义一个内层函数,并将这个内层函数作为返回值。
优点:
可以在不使用全局变量的情况下,让函数"记住"某个状态。
变量的作用域被限制在函数内部,不易被外部代码意外修改。
缺点:
由于内部函数持续引用外部函数的变量,这部分内存不会被垃圾回收机制释放,可能导致内存占用。
nonlocal
关键字
如果需要在闭包的内部函数中修改外部函数的变量,必须使用 nonlocal
关键字声明。
xxxxxxxxxx
def create_printer(logo: str):
"""
一个创建打印机的工厂函数,它返回一个闭包。
"""
def printer(msg: str, modify_logo: bool = False):
# nonlocal 关键字允许我们修改外层函数的变量 'logo'
nonlocal logo
if modify_logo:
new_logo = "new_logo"
print(f"--- Logo changed from '{logo}' to '{new_logo}' ---")
logo = new_logo
print(f"[{logo}]: {msg}")
return printer
# 创建一个带有 "touhou" logo 的打印机
touhou_printer = create_printer("touhou")
# 使用这个打印机
touhou_printer("yoyomu")
touhou_printer("reimu")
# 修改 logo 并再次使用
touhou_printer("marisa", modify_logo=True)
touhou_printer("sakuya")
输出:
xxxxxxxxxx
[touhou]: yoyomu
[touhou]: reimu
--- Logo changed from 'touhou' to 'new_logo' ---
[new_logo]: marisa
[new_logo]: sakuya
2. 装饰器 (Decorator)
装饰器本质上也是一种闭包,它是一种强大的设计模式,允许我们在不修改目标函数源代码的前提下,为其增加额外的功能(如日志、计时、权限校验等)。
xxxxxxxxxx
import time
import random
def timer_decorator(func):
"""一个简单的装饰器,用于打印函数执行前后的信息。"""
def wrapper():
print("--- Function is about to start ---")
func()
print("--- Function has finished ---")
return wrapper
# 使用 @ 语法糖来应用装饰器
def sleep_randomly():
"""一个随机睡眠的函数。"""
delay = random.randint(1, 3)
print(f"Sleeping for {delay} seconds...")
time.sleep(delay)
# 调用被装饰后的函数
sleep_randomly()
@timer_decorator
这行代码完全等价于 sleep_randomly = timer_decorator(sleep_randomly)
。
3. 设计模式 (Design Pattern)
设计模式是经过验证的、在特定场景下解决常见问题的编程“套路”或最佳实践。
单例模式 (Singleton)
确保一个类在整个应用程序中只有一个实例,并提供一个全局访问点来获取该实例。
优点: 节省内存和创建对象的开销,确保全局状态的一致性。
实现方式: 通常通过重写类的
__new__
方法来实现。
xxxxxxxxxx
class MusicPlayer:
_instance = None # 用于存储唯一的实例
def __new__(cls, *args, **kwargs):
if cls._instance is None:
print("Creating a new MusicPlayer instance.")
cls._instance = super().__new__(cls)
else:
print("Returning the existing MusicPlayer instance.")
return cls._instance
def play(self, song):
print(f"Playing: {song}")
# 多次“创建”实例
player1 = MusicPlayer()
player2 = MusicPlayer()
print(f"player1 is player2: {player1 is player2}")
player1.play("Bohemian Rhapsody")
player2.play("Stairway to Heaven")
输出:
xxxxxxxxxx
Creating a new MusicPlayer instance.
Returning the existing MusicPlayer instance.
player1 is player2: True
Playing: Bohemian Rhapsody
Playing: Stairway to Heaven
工厂模式 (Factory)
提供一个统一的接口来创建一系列相关的对象,而无需指定它们具体的类。客户端通过工厂方法来请求对象,而不是直接使用类的构造函数。
优点:
将对象的创建逻辑与使用逻辑解耦,易于维护。
当需要更换或添加新的产品类时,只需修改工厂内部逻辑,客户端代码无需变动。
xxxxxxxxxx
# --- 产品类 ---
class Person:
def greet(self):
raise NotImplementedError
class Worker(Person):
def greet(self):
return "I am a worker."
class Student(Person):
def greet(self):
return "I am a student."
# --- 工厂类 ---
class PersonFactory:
def get_person(self, person_type: str) -> Person:
if person_type == 'worker':
return Worker()
elif person_type == 'student':
return Student()
else:
raise ValueError(f"Unknown person type: {person_type}")
# --- 客户端代码 ---
factory = PersonFactory()
p1 = factory.get_person('worker')
p2 = factory.get_person('student')
print(p1.greet())
print(p2.greet())
输出:
xxxxxxxxxx
I am a worker.
I am a student.
4. Python 多线程
进程 (Process): 操作系统进行资源分配和调度的基本单位,可以看作一个正在运行的程序。进程之间内存相互隔离。
线程 (Thread): 进程的实际工作单位,一个进程可以包含多个线程。同一进程内的线程共享该进程的内存空间。
并行 (Parallelism): 同一时刻执行多个任务(需要多核CPU)。
并发 (Concurrency): 在一段时间内交替执行多个任务,看起来像在同时执行(单核CPU即可实现)。
Python 的 threading
模块可以帮助我们实现多线程编程。
threading
模块基础
xxxxxxxxxx
import threading
import time
def sing(song_name: str, duration: int):
"""唱歌任务"""
print(f"Thread '{threading.current_thread().name}' starting to sing '{song_name}'...")
time.sleep(duration)
print(f"Thread '{threading.current_thread().name}' finished singing.")
def dance(style: str, duration: int):
"""跳舞任务"""
print(f"Thread '{threading.current_thread().name}' starting to dance {style}...")
time.sleep(duration)
print(f"Thread '{threading.current_thread().name}' finished dancing.")
# 创建线程对象
# target: 指定线程要执行的函数
# args: 以元组形式传递位置参数
# kwargs: 以字典形式传递关键字参数
# name: 自定义线程名称
sing_thread = threading.Thread(target=sing, args=("Let It Go", 3), name="SingerThread")
dance_thread = threading.Thread(target=dance, kwargs={"style": "Ballet", "duration": 2}, name="DancerThread")
# 启动线程
sing_thread.start()
dance_thread.start()
print(f"Main thread continues to run...")
# 等待子线程执行完毕
sing_thread.join()
dance_thread.join()
print("All threads have finished.")
输出 (顺序可能不同):
xxxxxxxxxx
Thread 'SingerThread' starting to sing 'Let It Go'...
Thread 'DancerThread' starting to dance Ballet...
Main thread continues to run...
Thread 'DancerThread' finished dancing.
Thread 'SingerThread' finished singing.
All threads have finished.
5. Socket 通信
Socket(套接字)是实现进程间网络通信的工具。它分为服务端 (Server) 和客户端 (Client)。
Socket 服务端基础流程
创建 Socket 对象:
socket.socket()
绑定 IP 和端口:
socket_server.bind((host, port))
开始监听:
socket_server.listen(backlog)
接受连接 (阻塞):
conn, addr = socket_server.accept()
接收/发送数据:
conn.recv(bufsize)
/conn.send(data)
关闭连接:
conn.close()
,socket_server.close()
xxxxxxxxxx
# server.py
import socket
# 1. 创建 Socket 对象
server = socket.socket()
# 2. 绑定 IP 和端口
server.bind(('localhost', 8888))
# 3. 开始监听
server.listen(1)
print("Server is listening on port 8888...")
# 4. 接受客户端连接(这是一个阻塞操作)
conn, address = server.accept()
print(f"Accepted connection from {address}")
while True:
# 5. 接收客户端数据
data_bytes = conn.recv(1024)
if not data_bytes: # 如果接收到空数据,表示客户端已断开
break
data_str = data_bytes.decode("utf-8")
print(f"Received from client: {data_str}")
# 6. 回复消息
reply = f"Message '{data_str}' received."
conn.send(reply.encode("utf-8"))
# 7. 关闭连接
conn.close()
server.close()
print("Connection closed.")
0 评论:
发表评论