Python常见面试题的详解15

1. 死锁（Deadlock）

死锁指的是在多线程或者多进程的运行环境中，两个或多个线程（进程）彼此等待对方释放所占用的资源，进而陷入无限期等待的僵局，最终导致程序无法继续推进。

必要条件 互斥条件：资源在某一时间段内只能被一个进程（线程）独占使用。请求和保持条件：进程（线程）已经持有了至少一个资源，同时又发起了对其他已被别的进程（线程）占用资源的请求，此时该进程（线程）会被阻塞，但它不会释放自己已持有的资源。不剥夺条件：进程（线程）已经获取的资源，在使用完毕之前不会被强制剥夺，只能由进程（线程）自身主动释放。循环等待条件：在发生死锁时，必然存在一个进程 - 资源的环形链，即进程集合 {P0, P1, P2, …, Pn} 中，P0 等待 P1 占用的资源，P1 等待 P2 占用的资源，依此类推，Pn 等待 P0 占用的资源。 要点

解决：破坏任一条件，如统一资源申请顺序、超时释放。

实际开发中可通过 threading.Lock.acquire(timeout=5) 设置超时避免死锁。

python

import threading lockA = threading.Lock() lockB = threading.Lock() def thread1(): with lockA: print("Thread1 acquired A") with lockB: # 等待 B 释放（可能死锁） print("Thread1 acquired B") def thread2(): with lockB: print("Thread2 acquired B") with lockA: # 等待 A 释放（可能死锁） print("Thread2 acquired A") t1 = threading.Thread(target=thread1) t2 = threading.Thread(target=thread2) t1.start(); t2.start() t1.join(); t2.join() # 可能卡住 示例

在下面代码中，如果线程 1 先获取了 lock1，线程 2 先获取了 lock2，接着线程 1 尝试获取 lock2，线程 2 尝试获取 lock1，就会发生死锁。

python

import threading # 创建两个锁 lock1 = threading.Lock() lock2 = threading.Lock() def thread1_function(): lock1.acquire() print("Thread 1 acquired lock 1") # 模拟一些工作 try: # 尝试获取 lock2 lock2.acquire() print("Thread 1 acquired lock 2") lock2.release() finally: lock1.release() def thread2_function(): lock2.acquire() print("Thread 2 acquired lock 2") # 模拟一些工作 try: # 尝试获取 lock1 lock1.acquire() print("Thread 2 acquired lock 1") lock1.release() finally: lock2.release() # 创建并启动线程 t1 = threading.Thread(target=thread1_function) t2 = threading.Thread(target=thread2_function) t1.start() t2.start() t1.join() t2.join()

 

2. 多线程竞争访问控制

在多线程环境下，为了避免对已访问的数据进行重复访问，可以借助锁机制和标记位来实现对数据访问的控制。

要点

目标：确保某数据仅被一个线程访问一次。

实现：互斥锁、原子操作或标志位。

python

import threading data_accessed = False lock = threading.Lock() def access_data(): global data_accessed if lock.acquire(blocking=False): # 非阻塞尝试获取锁 if not data_accessed: print(f"{threading.current_thread().name} 访问数据") data_accessed = True lock.release() threads = [threading.Thread(target=access_data) for _ in range(5)] for t in threads: t.start() for t in threads: t.join() 示例

在下面代码中，使用 threading.Event 更简洁：

python

event = threading.Event() if not event.is_set() and event.set(): print("访问数据") 示例

在下面代码中，使用 visited 列表来标记每个数据是否已被访问，使用 lock 来保证线程安全。每个线程在访问数据前会先检查标记位，若数据未被访问则进行访问并更新标记位。

python

import threading # 共享数据 shared_data = [1, 2, 3, 4, 5] # 标记列表，用于记录每个数据是否已被访问 visited = [False] * len(shared_data) # 锁，用于线程同步 lock = threading.Lock() def access_data(): global shared_data, visited while True: with lock: index = None for i in range(len(shared_data)): if not visited[i]: index = i visited[i] = True break if index is None: break print(f"Thread {threading.current_thread().name} accessed {shared_data[index]}") threads = [] for i in range(3): t = threading.Thread(target=access_data) threads.append(t) t.start() for t in threads: t.join()

 

3. 线程安全与互斥锁 要点

线程安全：当多个线程访问某个类时，无论运行时环境采用何种调度方式，也不管这些线程如何交替执行，并且在主调代码中无需额外的同步或协同操作，这个类都能表现出正确的行为。简单来讲，就是在多线程环境下，对共享资源的访问不会引发数据不一致或其他异常问题。

互斥锁：用于线程同步的机制，它提供了排他性的访问控制。当一个线程获取了互斥锁后，其他线程就无法再获取该锁，必须等待持有锁的线程释放锁后才能继续竞争获取。

无锁线程安全：使用 queue.Queue 或 collections.deque 等线程安全数据结构。

python

from threading import Lock counter = 0 lock = Lock() def increment(): global counter for _ in range(1000): 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) # 正确输出 10000 示例

在下面代码中，使用 threading.Lock() 创建了一个互斥锁 lock。在 increment 函数中，通过 lock.acquire() 获取锁，确保同一时间只有一个线程可以对 counter 进行修改，修改完成后使用 lock.release() 释放锁。

python

import threading # 共享资源 counter = 0 # 创建互斥锁 lock = threading.Lock() def increment(): global counter for _ in range(100000): # 获取锁 lock.acquire() try: counter += 1 finally: # 释放锁 lock.release() threads = [] for i in range(2): t = threading.Thread(target=increment) threads.append(t) t.start() for t in threads: t.join() print(counter)

 

4. 同步、异步、阻塞、非阻塞 要点 同步：在同步操作中，调用者必须等待被调用的操作完成后才能继续执行后续代码，程序的执行是按顺序依次进行的。异步：异步操作中，调用者发起操作后无需等待操作完成，可继续执行后续代码。当异步操作完成后，会通过回调函数、事件等方式通知调用者。阻塞：执行某个操作时，当前线程会被挂起，直到该操作完成才能继续执行后续代码。例如，调用阻塞的 I/O 操作时，线程会一直等待，直至操作完成。非阻塞：执行操作时，无论操作是否完成，线程都不会被挂起，会立即返回。线程可继续执行其他任务，然后通过轮询等方式检查操作是否完成。

python

# 同步阻塞（requests 库） import requests response = requests.get(" example ") # 阻塞直到响应返回 # 异步非阻塞（asyncio + aiohttp） import aiohttp, asyncio async def fetch(): async with aiohttp.ClientSession() as session: async with session.get(" example ") as response: return await response.text() async def main(): task = asyncio.create_task(fetch()) print("其他操作...") # 不阻塞 content = await task # 等待结果 asyncio.run(main()) 示例

在下面代码中，sync_blocking 函数是同步阻塞的，调用时主线程会等待 2 秒。async_non_blocking 函数是异步非阻塞的，它会启动一个新线程来执行任务，主线程不会等待该任务完成，而是继续执行后续代码。

python

import time import threading # 同步阻塞函数 def sync_blocking(): print("Sync blocking operation started") time.sleep(2) print("Sync blocking operation finished") # 异步非阻塞函数 def async_non_blocking(callback): def task(): print("Async non - blocking operation started") time.sleep(2) print("Async non - blocking operation finished") callback() t = threading.Thread(target=task) t.start() # 回调函数 def callback_function(): print("Callback function called") # 同步阻塞调用 sync_blocking() # 异步非阻塞调用 async_non_blocking(callback_function) print("Main thread continues to execute")

 

5. 僵尸进程与孤儿进程 要点 僵尸进程：当子进程结束运行（调用 exit 或 _exit）后，其进程描述符不会立即被删除，而是会保留在系统中，直到父进程调用 wait 或 waitpid 函数来获取子进程的退出状态。在此期间，子进程处于僵尸状态，称为僵尸进程。大量僵尸进程会占用系统资源。孤儿进程：若父进程提前结束运行，而其子进程仍在运行，这些子进程就会成为孤儿进程。孤儿进程会被 init 进程（进程 ID 为 1）收养，由 init 进程负责处理它们的退出状态。调用 wait 或 waitpid：父进程在子进程结束后，调用 wait 或 waitpid 函数来获取子进程的退出状态，从而清除子进程的僵尸状态。信号处理：父进程可以通过信号处理函数捕获 SIGCHLD 信号，在信号处理函数中调用 wait 或 waitpid 函数。

python

import os, time pid = os.fork() if pid == 0: # 子进程 print("Child process exiting") else: # 父进程 time.sleep(30) # 父进程不调用 os.wait()，子进程成为僵尸 print("Parent process exiting") 示例

在下面代码中，父进程通过 signal.signal(signal.SIGCHLD, sigchld_handler) 注册了一个信号处理函数 sigchld_handler，当子进程结束时，会触发 SIGCHLD 信号，在信号处理函数中调用 os.waitpid 来处理子进程的退出状态，避免僵尸进程的产生。

python

import os import signal import time def sigchld_handler(signum, frame): while True: try: pid, status = os.waitpid(-1, os.WNOHANG) if pid == 0: break except OSError: break signal.signal(signal.SIGCHLD, sigchld_handler) pid = os.fork() if pid == 0: print(f"Child process {os.getpid()} is running") time.sleep(2) print(f"Child process {os.getpid()} is exiting") else: print(f"Parent process {os.getpid()} is running") time.sleep(5) print(f"Parent process {os.getpid()} is exiting")

6. 进程与线程使用场景 要点

     1. 进程的使用场景

CPU 密集型任务：对于需要大量计算的任务，如科学计算、图像处理等，使用多进程可以充分利用多核 CPU 的优势，提高程序的执行效率。因为进程之间相互独立，不受 GIL（全局解释器锁）的限制。隔离性要求高的任务：当任务之间需要高度隔离，例如一个任务可能会崩溃或出现异常，为了不影响其他任务的执行，可以使用多进程。

     2. 线程的使用场景

IO 密集型任务：对于需要大量 IO 操作的任务，如网络请求、文件读写等，使用多线程可以在 IO 操作时释放 GIL，让其他线程继续执行，从而提高程序的并发性能。任务之间共享数据频繁：线程之间共享同一进程的内存空间，因此在需要频繁共享数据的场景下，使用多线程比多进程更方便。

python

# 多线程（适合 IO） import threading def io_task(): time.sleep(1) # 模拟 IO threads = [threading.Thread(target=io_task) for _ in range(100)] for t in threads: t.start() # 多进程（适合 CPU） from multiprocessing import Process def cpu_task(): sum(range(10**7)) # 模拟计算 processes = [Process(target=cpu_task) for _ in range(4)] for p in processes: p.start() 示例

在下面代码中，cpu_intensive_task 是 CPU 密集型任务，使用多进程来处理；io_intensive_task 是 IO 密集型任务，使用多线程来处理。

python

import multiprocessing import threading import time # CPU 密集型任务 def cpu_intensive_task(): result = 0 for i in range(1000000): result += i return result # IO 密集型任务 def io_intensive_task(): time.sleep(1) return # 多进程处理 CPU 密集型任务 if __name__ == '__main__': processes = [] for _ in range(4): p = multiprocessing.Process(target=cpu_intensive_task) processes.append(p) p.start() for p in processes: p.join() # 多线程处理 IO 密集型任务 threads = [] for _ in range(4): t = threading.Thread(target=io_intensive_task) threads.append(t) t.start() for t in threads: t.join() 7. 线程与进程的并发/并行 要点

线程：由于 GIL 的存在，多线程在 CPU 密集型任务中只能实现并发，不能实现并行。并发是指在同一时间段内，多个任务交替执行；并行是指在同一时刻，多个任务同时执行。但在 IO 密集型任务中，多线程可以通过在 IO 操作时释放 GIL，让其他线程继续执行，从而实现并发。

进程：多进程可以实现并行。因为每个进程都有自己独立的内存空间和 CPU 资源，多个进程可以在多核 CPU 上同时执行，提高程序的执行效率。

python

# 多线程（受 GIL 限制） def count(n): while n > 0: n -= 1 # 多线程执行时间 ≈ 单线程 t1 = threading.Thread(target=count, args=(10**8,)) t2 = threading.Thread(target=count, args=(10**8,)) t1.start(); t2.start() # 总时间 ≈ 单线程的两倍 # 多进程（真正并行） p1 = Process(target=count, args=(10**8,)) p2 = Process(target=count, args=(10**8,)) p1.start(); p2.start() # 总时间 ≈ 单线程的一半  示例

在下面代码中，通过对比多线程和多进程执行 CPU 密集型任务的时间，可以看出多进程在 CPU 密集型任务中能更好地利用多核 CPU 实现并行，而多线程由于 GIL 的限制，主要是并发执行。

python

import threading import multiprocessing import time # CPU 密集型任务 def cpu_task(): result = 0 for i in range(1000000): result += i # 多线程执行 CPU 密集型任务 start_time_thread = time.time() threads = [] for _ in range(4): t = threading.Thread(target=cpu_task) threads.append(t) t.start() for t in threads: t.join() end_time_thread = time.time() print(f"Time taken by threads: {end_time_thread - start_time_thread} seconds") # 多进程执行 CPU 密集型任务 if __name__ == '__main__': start_time_process = time.time() processes = [] for _ in range(4): p = multiprocessing.Process(target=cpu_task) processes.append(p) p.start() for p in processes: p.join() end_time_process = time.time() print(f"Time taken by processes: {end_time_process - start_time_process} seconds") 8. 并行（Parallel）与并发（Concurrency） 要点

并发：并发是指在同一时间段内，多个任务交替执行。并发并不要求多个任务同时执行，而是通过任务的切换来实现多个任务的同时推进。例如，在单 CPU 系统中，通过时间片轮转的方式，多个线程可以交替执行，看起来就像是同时执行一样。

并行：并行指的是在同一时刻，多个任务同时执行。这需要多个处理器或多核处理器的支持，每个处理器可以同时处理一个任务，从而提高程序的执行效率。例如，在多核 CPU 的计算机上，多个进程可以同时在不同的 CPU 核心上执行。

对比

并发：单车道交替通行（线程切换）。

并行：多车道同时行驶（多核 CPU）。

 示例

在下面代码中，使用多线程实现并发，使用多进程模拟并行。

python

import threading import time # 任务函数 def task(): print("Task started") time.sleep(1) print("Task finished") # 并发示例 threads = [] for _ in range(3): t = threading.Thread(target=task) threads.append(t) t.start() for t in threads: t.join() # 并行示例（借助多进程模拟） import multiprocessing if __name__ == '__main__': processes = [] for _ in range(3): p = multiprocessing.Process(target=task) processes.append(p) p.start() for p in processes: p.join()

9. IO 密集型 vs CPU 密集型 要点 IO 密集型： 定义：IO 密集型任务在程序执行过程中，大部分时间都花费在 IO 操作上，如网络请求、文件读写等。CPU 在这些任务中大部分时间处于空闲状态。特点：任务的执行时间主要取决于 IO 设备的性能，而非 CPU 的性能。优化方式：可以使用多线程或异步 IO 来提高程序的并发性能，因为在 IO 操作时可以释放 CPU 资源，让其他任务继续执行。 CPU 密集型： 定义：CPU 密集型任务在程序执行过程中，大部分时间都花费在 CPU 计算上，如科学计算、图像处理等。特点：任务的执行时间主要取决于 CPU 的性能，而非 IO 设备的性能。优化方式：可以使用多进程来充分利用多核 CPU 的优势，因为进程之间相互独立，不受 GIL 的限制。 优化策略 类型优化方案Python 工具IO 密集型异步非阻塞、多线程asyncio, aiohttpCPU 密集型多进程、分布式计算multiprocessing  示例

在下面代码中，io_task 是 IO 密集型任务，使用多线程

 python

import time import threading import multiprocessing # IO 密集型任务 def io_task(): print("IO task started") time.sleep(2) print("IO task finished") # CPU 密集型任务 def cpu_task(): result = 0 for i in range(10000000): result += i print("CPU task finished") # 多线程处理 IO 密集型任务 threads = [] for _ in range(3): t = threading.Thread(target=io_task) threads.append(t) t.start() for t in threads: t.join() # 多进程处理 CPU 密集型任务 if __name__ == '__main__': processes = [] for _ in range(3): p = multiprocessing.Process(target=cpu_task) processes.append(p) p.start() for p in processes: p.join()

10. asyncio 原理 要点

事件循环：事件循环是asyncio的核心，它负责调度和执行协程。事件循环会不断地从任务队列中取出待执行的协程，执行协程直到遇到await关键字，然后将协程挂起，继续执行其他协程。当await等待的异步操作完成后，事件循环会将挂起的协程恢复执行。

协程：asyncio使用协程作为异步操作的基本单位。协程是一种轻量级的线程，它可以在代码中暂停和恢复执行。在 Python 中，使用async def定义协程函数，使用await关键字来暂停协程的执行，直到等待的异步操作完成。

Future和Task：Future是一个表示异步操作结果的对象，它可以用来跟踪异步操作的状态。Task是Future的子类，它封装了一个协程，用于在事件循环中执行。可以通过asyncio.create_task()函数来创建一个Task对象。

异步 IO 操作：asyncio提供了一系列的异步 IO 操作，如异步网络请求、异步文件读写等。这些异步 IO 操作会在底层使用非阻塞 IO 和事件通知机制，当 IO 操作完成后，会通过回调函数通知事件循环。

 示例

在下面代码中，定义了一个协程函数task，在main函数中创建了多个Task对象，并使用asyncio.gather()函数来等待所有任务完成。最后，使用asyncio.run()函数来启动事件循环。

python

import asyncio async def task(n): print(f"Task {n} start") await asyncio.sleep(1) # 非阻塞等待 print(f"Task {n} end") async def main(): await asyncio.gather(task(1), task(2)) # 并发执行 asyncio.run(main()) # 输出： # Task 1 start → Task 2 start → （1秒后）→ Task 1 end → Task 2 end

友情提示：本文已经整理成文档，可以到如下链接免积分下载阅读

download.csdn.net/download/ylfhpy/90406977