适合纯文科生的 python 100个知识点 实践 一
Python高级语法
24 多线程编程
多线程编程可以提高程序的执行效率,因为在多核 CPU 的情况下,多个线程可以同时执行不同的任务,从而加速整个程序的执行速度。但同时也要注意线程安全问题,避免出现线程冲突、死锁等问题。
实际应用中,多线程编程可以用于网络通信、图像处理、数据分析等需要大量计算的任务。例如,一个图片处理程序可以将图片分成多个部分,每个部分分配给一个线程并行处理,最后合并结果,提高图片处理速度。
24.1 threading
库
这个例子使用了threading库,创建了两个线程来同时执行不同的任务。
import threading
def worker1():
print("工作1开始")
for i in range(1000000):
pass
print("工作1完成")
def worker2():
print("工作2开始")
for i in range(1000000):
pass
print("工作2完成")
if __name__ == "__main__":
print("主线开始")
thread1 = threading.Thread(target=工人1)
thread2 = threading.Thread(target=工人2)
thread1.start()
thread2.start()
thread1.join()
thread2.join()
print("主线完成")
这个例子创建了两个函数worker1和worker2,分别用于模拟两个耗时的任务。在主函数中,使用threading库创建了两个线程,并使用start()方法启动它们。然后,使用join()方法等待线程执行完毕。最后,输出“Main thread finished”表示主线程执行结束。
注意,使用多线程编程需要注意线程之间的同步和数据共享问题,例如使用锁、条件变量等机制来确保线程之间的安全访问。
25 多进程编程
25.1 multiprocessing
库 示例1
python的multiprocessing模块创建多个进程并在这些进程之间共享数据。
import multiprocessing
def worker(num):
"""每个工作进程执行的函数"""
print("工人", 数值, "开始")
return
if __name__ == '__main__':
jobs = []
for i in range(5):
p = multiprocessing.Process(target=工人, args=(i,))
jobs.append(p)
p.start()
该代码创建了5个进程,每个进程都调用worker
函数,打印出一个消息并返回。这里使用了multiprocessing
模块来创建并启动进程,Process
类表示一个进程对象,target
参数指定进程要执行的函数,args
参数是传递给函数的参数元组。
当我们运行上面的代码时,它会输出以下内容:
0号工人开始工作
1号工人开始工作
2号工人开始工作
3号工人开始工作
4号工人开始工作
这表明所有的工作进程都已成功启动,并且正在执行它们的任务。
一个更实际的多进程编程的例子是计算一组大量的数字的平方和。在这个例子中,我们可以使用多个进程来并行计算这些数字的平方和,以提高程序的执行效率。下面是示例代码:
import multiprocessing
def square_sum(numbers):
"""计算数字列表的平方和"""
return sum([n**2 for n in numbers])
if __name__ == '__main__':
numbers = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
pool = multiprocessing.Pool(processes=4)
results = pool.map(square_sum, [numbers[i:i+3] for i in range(0, len(numbers), 3)])
total = sum(results)
print("数字的平方和是",总计)
该代码将数字列表分为3个部分,每个部分交给一个进程来计算。这里使用了Pool
类来创建一个进程池,processes
参数指定进程池中的进程数。map
方法将分割后的数字列表传递给每个进程,并收集每个进程返回的结果。最终,所有结果被添加在一起,输出数字列表的平方和。
当我们运行上面的代码时,它会输出以下内容:
数字的平方和是385
这表明我们成功地使用多个进程并行计算了数字列表的平方和,从而加速了计算过程。
24.1 multiprocessing
库 示例2
import multiprocessing
def worker(num):
"""每个进程会执行的函数"""
print("工人%d正在运行" % num)
for i in range(5):
print("工人 %d: %d" % (num, i))
print("工人 %d 已经完成" % num)
if __name__ == "__main__":
# 创建进程池
pool = multiprocessing.Pool(processes=4)
# 启动多个进程
for i in range(4):
pool.apply_async(worker, args=(i,))
# 等待所有进程完成
pool.close()
pool.join()
print("所有工人都已完成")
在这个例子中,我们首先定义了一个worker
函数,每个进程都会调用该函数。然后我们使用multiprocessing.Pool
创建了一个进程池,池中包含4个进程。接下来,我们使用pool.apply_async
方法来启动4个进程,并将它们分配给进程池中的4个进程。最后,我们使用pool.close()
和pool.join()
方法来等待所有进程完成。
该程序会输出以下结果:
yamlCopy code
Worker 0 is running
Worker 1 is running
Worker 2 is running
Worker 3 is running
Worker 0: 0
Worker 1: 0
Worker 2: 0
Worker 3: 0
Worker 0: 1
Worker 1: 1
Worker 2: 1
Worker 3: 1
Worker 0: 2
Worker 1: 2
Worker 2: 2
Worker 3: 2
Worker 0: 3
Worker 1: 3
Worker 2: 3
Worker 3: 3
Worker 0: 4
Worker 1: 4
Worker 2: 4
Worker 3: 4
Worker 0 is done
Worker 1 is done
Worker 2 is done
Worker 3 is done
All workers are done
在上面的结果中,我们可以看到4个进程分别输出了自己的进程号,以及一些数字。由于使用了进程池,程序的运行速度得到了显著提高。
26协程
26.1 协程的定义
协程(coroutine)是一种用户态的轻量级线程,由用户自行控制调度,不需要操作系统进行上下文切换,因此协程具有高效、灵活等优点。在 Python 中,可以使用 async/await 关键字来定义协程函数。
26.2 协程的代码示例
import asyncio
async def coroutine_func():
print("启动coroutine函数")
await asyncio.sleep(1) # 模拟耗时操作
print("Coroutine函数完成")
async def main():
print("启动主功能")
await coroutine_func()
print("主要功能完成")
asyncio.run(main())
在这个示例中,coroutine_func() 是一个协程函数,它使用 await asyncio.sleep(1) 模拟了一个耗时的操作。而 main() 函数中调用了 coroutine_func(),并使用 await 等待其完成。
26.3 输出结果
启动主函数
启动coroutine函数
Coroutine函数结束
主函数结束
这表明在调用协程函数时,程序不会被阻塞,而是可以继续执行其他的任务。
26.4 异步读写文件
import asyncio
async def read_file(file_path):
print(f "读取文件 {file_path}")
async with open(file_path, "r") as f:
content = await f.read()
print(f "从{file_path}读取{len(content)}字节")
async def write_file(file_path, content):
print(f "正在写入文件{file_path}")
async with open(file_path, "w") as f:
await f.write(content)
print(f "将{len(content)}字节写到{file_path}")
async def main():
await asyncio.gather(
read_file("input.txt"),
write_file("output.txt", "Hello, world!")
)
asyncio.run(main())
在这个示例中,read_file() 和 write_file() 函数都是协程函数,使用 async with open(file_path, "r") as f 和 async with open(file_path, "w") as f 来异步读写文件。main() 函数中使用 asyncio.gather() 来同时调用这两个协程函数。
运行这个示例代码后,会读取 input.txt 文件并将 "Hello, world!" 写入 output.txt 文件。
总之,协程是 Python 中非常强大的异步编程工具,可以大大提高程序的性能和并发能力。
27 类装饰器
27.1 类装饰器的定义
类装饰器是一种用于装饰类的装饰器,其实现方式与函数装饰器有所不同。类装饰器使用类作为装饰器函数,该类必须实现__call__
方法,可以在类实例化时传入被装饰类作为参数,在__call__
方法中对被装饰类进行修改或增强,并返回修改后的被装饰类。
27.2 单例模式装饰器
class Singleton:
def __init__(self, cls):
self.cls = cls
self.instance = None
def __call__(self, *args, **kwargs):
if not self.instance:
self.instance = self.cls(*args, **kwargs)
return self.instance
@Singleton
class MyClass:
def __init__(self, x):
self.x = x
obj1 = MyClass(10)
obj2 = MyClass(20)
print(obj1.x) # 10
print(obj2.x) # 10
在上面的例子中,我们定义了一个类装饰器Singleton
,它将被装饰类变为单例模式。我们使用@Singleton
将MyClass
装饰起来,并且创建了两个MyClass
的实例obj1
和obj2
,发现它们的x
属性值相同,因为它们实际上是同一个对象。
27.3 计时器装饰器
import time
class Timer:
def __init__(self, func):
self.func = func
def __call__(self, *args, **kwargs):
start = time.time()
result = self.func(*args, **kwargs)
end = time.time()
print(f"{self.func.__name__} took {end-start:.4f} seconds")
return result
@Timer
def my_function():
time.sleep(1)
my_function()
在上面的例子中,我们定义了一个类装饰器Timer
,它会计算被装饰函数的执行时间,并在函数执行完毕后打印出来。我们使用@Timer
将my_function
函数装饰起来,并且调用my_function()
,可以看到它打印出了函数执行时间。
27.4 授权装饰器
class Authorize:
def __init__(self, role):
self.role = role
def __call__(self, cls):
def wrapper(*args, **kwargs):
if self.role == "admin":
return cls(*args, **kwargs)
else:
raise ValueError("Unauthorized access")
return wrapper
@Authorize("admin")
class MyClass:
def __init__(self, x):
self.x = x
obj = MyClass(10)
print(obj.x) # 10
@Authorize("user")
class MyOtherClass:
def __init__(self, x):
self.x = x
obj2 = MyOtherClass(20) # raises ValueError("Unauthorized access")
在上面的例子中,我们定义了一个类装饰器Authorize
,它会检查用户角色是否为管理员。如果是管理员,就允许访问被装饰类,否则就抛出异常。我们使用@Authorize("admin")
将MyClass
装饰起来,这样就可以创建一个MyClass
的实例obj
。然而,如果我们使用@Authorize("user")
将`MyOtherClass
class MyDecorator:
def __init__(self, cls):
self.cls = cls
def __call__(self, *args, **kwargs):
obj = self.cls(*args, **kwargs)
obj.attr = "decorated attribute"
return obj
@MyDecorator
class MyClass:
def __init__(self, x):
self.x = x
obj = MyClass(10)
print(obj.x) # 10
print(obj.attr) # "decorated attribute"
在上面的例子中,我们定义了一个类装饰器MyDecorator
,它将一个新属性attr
添加到被装饰类MyClass
的实例上。我们使用@MyDecorator
将MyClass
装饰起来,并且创建了一个MyClass
的实例obj
,打印出obj.x
的值是10,而obj.attr
的值是"decorated attribute"。
27.5 缓存类装饰器
class Cache:
def __init__(self, func):
self.func = func
self.cache = {}
def __call__(self, *args):
if args in self.cache:
return self.cache[args]
result = self.func(*args)
self.cache[args] = result
return result
@Cache
def fibonacci(n):
if n in (0, 1):
return n
return fibonacci(n-1) + fibonacci(n-2)
print(fibonacci(10)) # 55
print(fibonacci.cache) # {(0,): 0, (1,): 1, (2,): 1, (3,): 2, (4,): 3, (5,): 5, (6,): 8, (7,): 13, (8,): 21, (9,): 34, (10,): 55}
在上面的例子中,我们定义了一个类装饰器Cache
,它会缓存被装饰函数的结果,以避免重复计算。我们使用@Cache
将fibonacci
函数装饰起来,并且调用fibonacci(10)
,得到的结果是55,而fibonacci.cache
的值是一个字典,记录了先前调用过的函数结果。
28 asyncio库
28.1 asyncio库的定义
asyncio是Python中的异步编程库,可以用于编写高效且可扩展的异步应用程序。
28.2 asyncio库代码示例
import asyncio
async def main():
print('你好')
await asyncio.sleep(1)
print('世界')
asyncio.run(main())
上面的代码定义了一个名为main
的协程,该协程打印“Hello”,等待1秒钟,然后再打印“World”。asyncio.run()
函数用于运行main()
协程并等待其完成。
下面是一个更复杂的示例,演示如何使用asyncio和aiohttp库从多个URL异步获取网页内容:
pythonCopy code
import asyncio
import aiohttp
async def fetch(session, url):
async with session.get(url) as response:
return await response.text()
async def main():
async with aiohttp.ClientSession() as session:
urls = [
'https://www.baidu.com',
'https://www.tengxun.org',
'https://www.qq.com'
]
tasks = [asyncio.create_task(fetch(session, url)) for url in urls]
html_pages = await asyncio.gather(*tasks)
print(html_pages)
asyncio.run(main())
上面的代码使用了aiohttp
库来异步获取多个URL的网页内容。fetch()
函数用于获取单个URL的网页内容,并返回其文本。main()
函数使用aiohttp.ClientSession()
创建了一个异步HTTP客户端会话,并使用asyncio.create_task()
创建了多个任务,每个任务使用fetch()
函数异步获取一个URL的网页内容。最后,asyncio.gather()
函数等待所有任务完成,并将结果打印出来。
29 元编程
29.1 元编程的定义
元编程是指编写代码来创建、操作或扩展程序自身的代码。ython是一种动态语言,提供了很多元编程技术,例如使用type()函数创建类、使用装饰器动态地修改函数等 等。
29.2 MyClass
class MyClass:
pass
# 动态地添加属性
MyClass.x = 1
MyClass.y = 2
# 动态地添加方法
def my_method(self):
print("Hello, world!")
MyClass.my_method = my_method
# 动态地创建类
MySubclass = type("MySubclass", (MyClass,), {"z": 3})
# 使用装饰器动态地修改函数
def my_decorator(func):
def wrapper(*args, **kwargs):
print("Before function")
result = func(*args, **kwargs)
print("After function")
return result
return wrapper
@my_decorator
def my_function():
print("你好,世界!")
# 测试代码
obj = MyClass()
print(obj.x) # 1
print(obj.y) # 2
obj.my_method() #你好,世界!
print(MySubclass.z) # 3
my_function() # Before function \n Hello, world! \n After function
在这个示例中,我们首先创建了一个简单的类 MyClass,然后使用元编程技术动态地添加属性和方法。接着,我们使用 type() 函数动态地创建了一个子类 MySubclass。最后,我们使用装饰器来动态地修改一个函数。在测试代码中,我们创建了 MyClass 的一个实例并测试了添加的属性和方法,创建了 MySubclass 的一个实例并测试了添加的属性,以及调用了被装饰的函数并测试了装饰器的效果。
29.3 MyMeta
class MyMeta(type):
def __new__(cls, name, bases, attrs):
attrs['added_by_meta'] = True
return super().__new__(cls, name, bases, attrs)
class MyClass(metaclass=MyMeta):
pass
print(hasattr(MyClass, 'added_by_meta')) # True
在上面的示例中,我们定义了一个元类MyMeta
,该元类继承自type
,并重载了__new__
方法。在创建MyClass
类的过程中,Python会调用MyMeta.__new__
方法来完成类的创建过程,其中name
参数表示类名,bases
参数表示基类,attrs
参数表示属性字典。在重载的__new__
方法中,我们向attrs
字典中添加了一个新属性added_by_meta
,并将其值设置为True
,这样在创建MyClass
类后,就可以使用hasattr(MyClass, 'added_by_meta')
方法来检查是否添加了这个属性。
29.4 exec
Python还提供了其他元编程功能,如使用exec()
函数动态执行代码、使用getattr()
和setattr()
函数来动态获取和设置对象的属性等。下面是一个简单的使用exec()
函数动态创建函数的示例:
def create_function(name, arg_names, body):
# 拼接函数定义字符串
func_str = f"def {name}({', '.join(arg_names)}):\n"
func_str += '\n'.join([f'\t{line}' for line in body.split('\n')])
# 使用exec()函数动态执行代码,创建函数对象
globals_dict = {}
exec(func_str, globals_dict)
return globals_dict[name]
# 创建函数对象并调用
add = create_function('add', ['a', 'b'], 'return a + b')
print(add(1, 2)) # 输出:3
在上面的示例中,我们定义了一个create_function()
函数,该函数接受函数名、参数列表和函数体作为参数,使用字符串拼接的方式构造函数定义字符串,并使用exec()
函数动态执行代码,将函数对象保存在globals_dict
字典中。最后,我们可以通过函数名来访问函数对象,并调用该函数。
29.5 SingletonMeta
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 MyClass(metaclass=SingletonMeta):
pass
a = MyClass()
b = MyClass()
print(a is b) # True
在这个例子中,我们定义了一个元类SingletonMeta
,它负责实现单例模式。在这个元类中,我们使用_instances
字典来存储每个类的实例。在类被调用时,我们检查它是否已经存在于_instances
字典中,如果不存在,则创建一个新的实例并将其添加到_instances
字典中,否则直接返回现有的实例。这样就保证了每个类只有一个实例。
29.6 装饰器
另一个常见的元编程技术是使用装饰器来实现“注入”行为。例如,我们可以使用装饰器来实现一个简单的缓存系统:
cache = {}
def cached(fn):
def wrapper(*args, **kwargs):
key = str(args) + str(kwargs)
if key not in cache:
cache[key] = fn(*args, **kwargs)
return cache[key]
return wrapper
@cached
def expensive_function(n):
print("Computing...")
return n * 2
print(expensive_function(5))
print(expensive_function(5))
在这个例子中,我们定义了一个名为cache
的全局字典,它用于存储缓存的结果。我们还定义了一个名为cached
的装饰器,它将装饰的函数包装在另一个函数中。在这个新函数中,我们检查函数参数的字符串表示是否已经存在于cache
字典中,如果存在,则直接返回缓存的结果,否则计算函数的结果,并将其添加到cache
字典中。最后,我们使用@cached
装饰器来修饰expensive_function
函数,这样每次调用expensive_function
函数时,都会自动检查缓存是否存在,避免了重复计算。
30 面向切面编程
30.1 面向切面编程的定义
面向切面编程(Aspect-Oriented Programming,简称AOP)是一种编程范式,它通过在程序执行过程中,动态地将一些与核心业务无关的横切关注点(例如日志记录、安全控制、事务处理等),从业务逻辑中剥离出来,进行统一管理。
30.2 在函数执行前后打印日
在Python中,可以使用第三方库aspectlib
来实现面向切面编程。下面是一个简单的例子,演示如何在函数执行前后打印日志:
from aspectlib import Aspect
with Aspect() as aspect:
@aspect.log_calls()
def greet(name):
return f"你好,{姓名}。!"
greet("小王")
运行以上代码,输出如下:
Calling greet(name='小王')
Called greet(name='小王n') -> '小王!'
以上代码中,使用Aspect()
创建了一个切面对象aspect
,然后使用@aspect.log_calls()
装饰器对greet()
函数进行装饰。log_calls()
装饰器会在函数执行前后分别打印日志。
除了log_calls()
,aspectlib
还提供了其他装饰器,用于实现不同的横切关注点。例如:
log_time()
:在函数执行前后记录时间戳。retry()
:在函数执行失败时进行重试。cache()
:在函数执行结果被缓存,下次调用时直接返回缓存结果。
30.2 缓存函数的执行结果
下面是一个使用cache()
装饰器的例子,演示如何缓存函数的执行结果:
from aspectlib import Aspect
with Aspect() as aspect:
@aspect.cache()
def expensive_function(param):
print(f"Executing expensive_function({param})...")
return param ** 2
print(expensive_function(5)) # 第一次调用,执行 expensive_function(5)
print(expensive_function(5)) # 第二次调用,直接返回缓存结果
print(expensive_function(6)) # 新的参数,重新执行 expensive_function(6)
运行以上代码,输出如下:
Executing expensive_function(5)...
25
25
Executing expensive_function(6)...
36
本文是结合ChatGPT的答案总结的知识点,有不足之处请大佬们给出反馈。
转载自:https://juejin.cn/post/7217459018237870140