一、语言特性与底层机制

1、 解释Python的GIL(全局解释器锁)机制及其对多线程程序的影响。如何规避GIL的性能瓶颈?

首先,由于python语言的产生是比较早的,那时候计算多以单核CPU为主,所以当时未考虑这个问题。 
但是随着计算机的发展,多核CPU的出现,为了利用的多线程,设计了GIL,同时只能有一个线程执行

1.当一个线程在执行时,其他线程无法运行python代码,虽然操作系统会把CPU的时间片分配给不同的线程。
2.但是GIL对IO密集型任务的影响比较小,因为IO操作等等相对来说比较耗时,因为锁可能已经释放了

GIL对多线程程序的影响:
1. CPU密集的程序: 由于无法利用多核CPU的资源,尽管使用了多线程,但是每次只有一个线程在执行python代码。
2. IO密集型任务: GIL对IO密集型任务的影响比较小,因为IO操作等等相对来说比较耗时,因为锁可能已经释放了。


如何解决:
    1.使用多进程的方式而非多线程:
        1.1 multiprocessing模块:启动多进程,每个进程有自己的python解释器和GIL,可以充分利用多核的性能,适合CPU密集任务。
        1.2 多进程可以避开GIL的限制,但是每个进程有自己独立的内存空间,进程间通讯更新,Queue或者Pipe
    2.使用C扩展或者第三方库。
    3.异步编程: 对于IO密级任务,使用asyncio等异步框架。异步编程通过时间循环和协程在单线程中高效处理大量的IO操作,避免多线程GIL问题

2、描述Python的垃圾回收机制(引用计数+分代回收),并解释如何解决循环引用问题。

2.1 引用计数

每个python对象都有一个引用计数器,记录有多少个变量引用该对象。当引用计数为0时,表示该对象不再被使用,可以被释放,从而释放内存。
-  增加引用计数: 当一个新对象创建或者数据结构引用该对象时,引用计数+1
-  减少引用计数:  当一个引用变量被删除、重新赋值或者超出作用域,引用计数-1
- 内存回收: 当对象的引用计数为0时,python会自动回收该对象的内存,避免内存泄漏。

2.2 分代回收

引用计数虽然很有效,但是在有些情况下不能解决所有垃圾回收问题。 尤其是在处理长时间存在且生命周期较长的对象。
分代回收机制将对象分为不同“代”来优化垃圾回收的过程。

python的分代回收机制包括三代:
1.第0代: 创建新的对象,由于大多数对象的生命周期较短,因此大多数对象在0代就被回收的概率较高。
2.第1代: 经过第一次垃圾回收之后还存在的对象,因此存活时间教长
3.第2代: 经过多次垃圾回收任然还存活的对象。 他们被认为是长期存在的一些,因此回收率较低。

2.3 如何解决循环引用的问题:

  • 什么是循环引用?

    循环引用问题是指两个对象或者多个对象相互引用对方,导致对象的引用计数无法降低到0,造成内存泄漏。
    
    python的垃圾回收专门对循环引用做了优化,循环引用检测由GC模块的**垃圾回收器**负责,它会定期
    检测对象之间的引用关系,检测是否存在循环引用,并将其清理。
    
    - GC模块: 
    - 自动清理:在分代回收的过程中,Python的垃圾回收器会识别并清理循环引用。它会定期检查第0代和第1代对象,并执行检测以发现任何无法通过引用计数机制清理的对象。
    
  • 如何解决?

    弱引用:使用weakref模块创建弱引用,弱引用不会增加对象的引用计数,避免了对象间的强引用循环。对于需要共享和缓存的对象,可以使用weakref来避免循环引用。
    分离数据和行为:避免对象之间的直接引用关系,尽量将数据存储和行为分开,减少不必要的引用。
    优化代码的组织结构: 设计问题
    

3、以下代码的输出是什么?解释原因:

def func(x, lst=[]):
    lst.append(x)
    return lst
print(func(1))  # 输出? [1]
print(func(2))  # 输出? [1,2]


在python中,默认参数是函数定义式计算的,而不是每次调用时重新计算,所以后续调用会一直共享lst

4、__slots__的作用是什么?使用它有什么优缺点?

4.1 __slots__的作用

- 节省内存
- 提高性能
- 限制属性动态添加

4.2 __slots__的优缺点

  • 优点:
    • 节省内存:不再为乜咯实例维护一个__dict__对象,而使用一个紧凑的结构存储属性数据
    • 提高性能: 属性查找和访问比动态字典存储更快,时间复杂度更低
    • 防止意外动态属性: 避免不小心添加不需要的属性,能保证类设计的一致性。
  • 缺点:
    • 无法动态添加新属性
    • 与继承的兼容性差: 如果子类不使用__slots__,将恢复__dict__
    • 增加了代码的复杂性: 比如试图添加新的属性

二、元编程与设计模式

2.1 实现一个单例模式(Singleton),要求线程安全且支持懒加载。


import threading

class Singleton:
    __instance = None
    _lock = threading.Lock()

    def __new__(cls, *args, **kwargs):
        if cls.__instance is None:
            with cls._lock:
                if not cls.__instance:
                    cls.__instance = super(Singleton, cls).__new__(cls)
        return cls.__instance

    def __init__(self, value):
        if not hasattr(self, '_initialized'):
            self._initialized = True
            self.value = value

    def get_value(self):
        return self.value


# 测试单例模式
def test_singleton(value):
    singleton = Singleton(value)
    print(f"Singleton instance with value: {singleton.get_value()}")


# 模拟多线程环境
thread1 = threading.Thread(target=test_singleton, args=("first",))
thread2 = threading.Thread(target=test_singleton, args=("second",))

thread1.start()
thread2.start()

thread1.join()
thread2.join()

2.2 编写一个装饰器@retry(max_attempts=3, delay=1),使被装饰的函数在抛出异常时自动重试,最多尝试max_attempts次,每次间隔delay秒。

import functools
import time
import random


def retry(max_attempts=3, delay=1):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            attempt = 0
            while attempt < max_attempts:
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    attempt += 1
                    if attempt == max_attempts:
                        raise e
                    print(f"Retrying ({attempt}/{max_attempts}) after error: {e}")
                    time.sleep(delay)
        return wrapper
    return decorator


@retry(max_attempts=3, delay=2)
def unstable_function():
    if random.random() < 0.7:  # 70% 概率抛出异常
        raise ValueError("Random failure")
    return "Success"


# 运行测试
try:
    print(unstable_function())
except Exception as e:
    print(f"Function failed after retries: {e}")

2.3 解释元类(Metaclass)的作用,并用元类实现一个类属性自动大写的功能(如类中定义的foo = 1会被转换为FOO = 1)。

2.3.1 什么是元类?

元类(MetaClass)是python的一个高级特性,他是用来创建类的类。换句话说,元类定义的类的行为。
默认下,python的类是type创建的,但是可以通过MataClass来控制创建类的创建过程。

2.3.1 元类的工作原理

  • 类是由元类创建的: 类本身是元类创建,类的实例是类创建的。
  • 通过元类控制类的行为:元类可以修改类的属性、方法,甚至改变类的继承结构。

2.3.2 关键的元类方法

new:这个方法在类的实例化过程中首先被调用,用来创建类对象。
init:在 new 方法之后被调用,用来初始化类的属性。


# 定义一个元类
class UppercaseMeta(type):
    def __new__(cls, name, bases, dct):
        # 创建一个新的字典来存储修改后的类属性
        new_dct = {}
        for key, value in dct.items():
            if not key.startswith('__'):  # 排除内置属性和方法
                new_dct[key.upper()] = value
            else:
                new_dct[key] = value  # 保留内置属性
        return super().__new__(cls, name, bases, new_dct)

# 使用元类创建一个类
class MyClass(metaclass=UppercaseMeta):
    foo = 1
    bar = 2

# 测试
print(MyClass.FOO)  # 输出: 1
print(MyClass.BAR)  # 输出: 2

三、并发与异步编程

3.1 对比Python中multiprocessing、threading和asyncio的适用场景及性能差异。

3.1.1 multiprocessing多进程

  • 使用场景:
    • CPU密集型: 当任务需要大量的CPU计算时,multiprocessing适合于分布任务到多个进程,并且可以充分利用多核。
    • 避免GIL锁: 由于GIL的存在,无法真正实现多核并行计算,multiprocessing创建多个进程,每个进程有独立的内存空间和解释器,充分利用多核。
  • 性能特点:
    • 高性能: 通过多进程并行,充分利用多核CPU,适合负载的计算任务。
    • 内存开销大:每个进程有自己的独立的内存空间,进程之间通讯需要IPC,可能会增加内存开销和延时。
    • 启动开销:启动进程比启动线程慢,因此在需要拼单创建进程的场景中可能会有性能损耗。

import multiprocessing

def cpu_bound_task(n):
    return sum(i * i for i in range(n))

if __name__ == "__main__":
    with multiprocessing.Pool(4) as pool:
        results = pool.map(cpu_bound_task, [10**6] * 4)
    print(results)

3.1.2 threading

  • 适用场景:
    • IO密集型任务:当任务主要是等待IO操作(网络请求、磁盘操作、数据库查询)时,threading可以通过线程切换来提高效率。
    • GUI应用程序:在GUI应用中,通常需要多线程来避免界面卡顿,力图后台执行任务,同时保持前台响应。
  • 性能特点:
    • 受限于GIL: 受限于GIL的限制,同一时刻只有一个线程执行python程序。
    • 轻量级:线程的创建和切换要轻量,内存开销相对较小。
    • 适用于IO密集型:由于线程在等待IO操作时能够释放GIL,因此多个线程同时操作IO,提高并发性能。
import threading

def io_bound_task():
    import time
    time.sleep(2)
    print("I/O task completed")


threads = []

for _ in range(10):
    t = threading.Thread(target=io_bound_task)
    threads.append(t)
    t.start()
for thread in threads:
    thread.join()

3.1.3 asyncio (异步编程)

  • 适合场景:
    • IO密集型任务:与threading类似, asyncio适合于异步IO操作,如网络请求
    • 高并发应用:适用于需要高并发处理大量I/O,例如爬虫等等
    • 单线程异步任务:适用于单线程的异步任务,避免上下文的切换等
  • 性能特点:
    • 轻量级:比于多线程和多进程,asyncio 更加轻量。通过事件循环机制,单线程可以并发执行多个任务,而不需要多线程或多进程的开销。
    • 非阻塞:asyncio 任务在执行 I/O 操作时不会阻塞主线程,因此可以更高效地处理大量并发任务。
    • 适用于IO密级:asyncio 在处理 I/O 密集型任务时,能够通过非阻塞 I/O 提高效率,尤其是在需要大量等待 I/O 响应的情况下。

3.2 以下代码是否存在问题?如何修正?


import threading
counter = 0
def increment():
    global counter
    for _ in range(100000):
        counter += 1
threads = [threading.Thread(target=increment) for _ in range(10)]
for t in threads: t.start()
for t in threads: t.join()
print(counter)  # 输出是否总是1000000?
  • 修改之后:
import threading

counter = 0
lock = threading.Lock()


def increment():
    global counter
    for _ in range(100000):
        with lock:
            counter += 1


threads = [threading.Thread(target=increment) for _ in range(10)]
for t in threads: t.start()
for t in threads: t.join()
print(counter)  # 输出是否总是1000000?

3.3 用asyncio编写一个并发HTTP请求函数,要求同时请求3个URL,并在所有请求完成后返回结果列表。

四、性能优化与高级用法

4.1 如何用生成器(Generator)实现一个内存高效的惰性数据处理管道?举例说明。


def read_large_file(file_path):
    with open(file_path, 'r', encoding='utf-8') as f:
        for line in f:
            yield line.strip()

# 使用管道,避免一次性加载大文件
lines = read_large_file('data/test.txt')

4.2 使用functools.lru_cache优化递归斐波那契数列函数,并解释其原理。

from functools import lru_cache

@lru_cache(maxsize=None)
def fib(n):
    if n < 2:
        return n
    return fib(n - 1) + fib(n - 2)

4.3 如何用Cython或Numba加速Python代码?请简述步骤。

五、数据库与ORM

5.1 解释SQLAlchemy的Session和Connection的区别,以及scoped_session的作用。

5.2 如何用Python实现数据库连接池?写出伪代码逻辑。

5.3 设计一个支持批量插入10万条数据到MySQL的方案,避免内存溢出。

六、算法与数据结构

6.1 实现一个LRU缓存(Least Recently Used),要求所有操作时间复杂度为O(1)。


"""
LRU算法: 实现一个LRU缓存(Least Recently Used),要求所有操作时间复杂度为O(1)。
"""
from collections import OrderedDict


class LRU:
    def __init__(self, capacity):
        self.capacity = capacity
        self.cache = OrderedDict()

    def get(self, key):
        """
        获取数据
        :param key:
        :return:
        """
        if key not in self.cache:
            return -1
        self.cache.move_to_end(key)
        return self.cache[key]

    def set(self, key, value):
        """
        设置数据
        :param key:
        :param value:
        :return:
        """
        if key in self.cache:
            self.cache.move_to_end(key)
        self.cache[key] = value
        if len(self.cache) > self.capacity:
            self.cache.popitem(last=False)

6.2 编写一个快速排序(QuickSort)的生成器版本,支持惰性输出排序结果。

"""
编写一个快速排序(QuickSort)的生成器版本,支持惰性输出排序结果。
"""


def quickSort(arr, left, right):
    if left >= right:
        return
    p = partition(arr, left, right) # 对数组进行划分,返回基准元素的最终位置
    quickSort(arr, left, p - 1)  # 对左边部分进行
    quickSort(arr, p + 1, right)  # 对右边部分进行排序


def partition(arr, left, right):
    value = arr[right]  # 最右边的元素作为基准元素
    i = left
    for j in range(left, right):
        if arr[j] <= value:
            arr[j], arr[i] = arr[i], arr[j] #将小于基准的元素移到左边
            i += 1
    arr[i], arr[right] = arr[right], arr[i] # 将基准放在合适的位置
    return i


if __name__ == '__main__':
    arr = [8, 1, 2, 3, 6, 0, 100]
    quickSort(arr, 0, len(arr) - 1)
    print(arr)

# 方式二: 惰性输出
def quickSortGenerator(arr, left, right):
    if left < right:
        p = partition(arr, left, right)
        # yield 左半部分的排序结果
        yield from quickSortGenerator(arr, left, p - 1)
        # yield 当前元素(基准元素)已排序的结果
        yield arr
        # yield 右半部分的排序结果
        yield from quickSortGenerator(arr, p + 1, right)

def partition(arr, left, right):
    value = arr[right]  # 最右边的元素作为基准
    i = left
    for j in range(left, right):
        if arr[j] <= value:
            arr[j], arr[i] = arr[i], arr[j]
            i += 1
    arr[i], arr[right] = arr[right], arr[i]
    return i

if __name__ == '__main__':
    arr = [8, 1, 2, 3, 6, 0, 100]
    generator = quickSortGenerator(arr, 0, len(arr) - 1)
    
    for sorted_arr in generator:
        print(sorted_arr)

6.3 反转链表(LeetCode 206),要求递归和迭代两种解法。


"""

反转链表
"""

def reverse_list(head):
    """
    递归反转链表
    :param head:
    :return:
    """
    if not head or not head.next:
        return head
    p = reverse_list(head.next)
    head.next.next = head
    head.next = None
    return p


def reverse_list(head):
    """
    迭代反转链表
    :param head:
    :return:
    """
    prev, curr = None, head
    while curr:
        next_node = curr.next
        curr.next = prev
        prev = curr
        curr = next_node
    return prev

七、系统设计

7.1 设计一个支持动态扩缩容的任务队列系统(类似Celery),需考虑任务去重、失败重试和优先级。

7.2 如何用Python构建一个高性能REST API服务?列举关键库(如FastAPI)及优化点(如异步、缓存)。

7.3 设计一个实时日志分析系统,要求每秒处理10万条日志并统计错误率。

__END__