likes
comments
collection
share

适合纯文科生的 python 100个知识点 实践 一

作者站长头像
站长
· 阅读数 30

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,它将被装饰类变为单例模式。我们使用@SingletonMyClass装饰起来,并且创建了两个MyClass的实例obj1obj2,发现它们的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,它会计算被装饰函数的执行时间,并在函数执行完毕后打印出来。我们使用@Timermy_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的实例上。我们使用@MyDecoratorMyClass装饰起来,并且创建了一个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,它会缓存被装饰函数的结果,以避免重复计算。我们使用@Cachefibonacci函数装饰起来,并且调用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
评论
请登录