Python常见面试题的详解15

1. 死锁（Deadlock）

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

• 必要条件

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

• 要点

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

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

python

import threadinglockA = 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:# 尝试获取 lock2lock2.acquire()print("Thread 1 acquired lock 2")lock2.release()finally:lock1.release()def thread2_function():lock2.acquire()print("Thread 2 acquired lock 2")# 模拟一些工作try:# 尝试获取 lock1lock1.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. 多线程竞争访问控制

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

• 要点

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

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

python

import threadingdata_accessed = False
lock = threading.Lock()def access_data():global data_accessedif lock.acquire(blocking=False):  # 非阻塞尝试获取锁if not data_accessed:print(f"{threading.current_thread().name} 访问数据")data_accessed = Truelock.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, visitedwhile True:with lock:index = Nonefor i in range(len(shared_data)):if not visited[i]:index = ivisited[i] = Truebreakif index is None:breakprint(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. 线程安全与互斥锁

• 要点

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

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

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

python

from threading import Lockcounter = 0
lock = Lock()def increment():global counterfor _ in range(1000):with lock:  # 加锁保证原子性counter += 1threads = [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 counterfor _ in range(100000):# 获取锁lock.acquire()try:counter += 1finally:# 释放锁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. 同步、异步、阻塞、非阻塞

• 要点

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

python

# 同步阻塞（requests 库）
import requests
response = requests.get("https://example.com")  # 阻塞直到响应返回# 异步非阻塞（asyncio + aiohttp）
import aiohttp, asyncioasync def fetch():async with aiohttp.ClientSession() as session:async with session.get("https://example.com") 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. 僵尸进程与孤儿进程

• 要点

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

python

import os, timepid = 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 timedef sigchld_handler(signum, frame):while True:try:pid, status = os.waitpid(-1, os.WNOHANG)if pid == 0:breakexcept OSError:breaksignal.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 threadingdef io_task():time.sleep(1)  # 模拟 IOthreads = [threading.Thread(target=io_task) for _ in range(100)]
for t in threads: t.start()# 多进程（适合 CPU）
from multiprocessing import Processdef 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 = 0for i in range(1000000):result += ireturn 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. 线程与进程的并发/并行

• 要点

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

2. 进程：多进程可以实现并行。因为每个进程都有自己独立的内存空间和 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 = 0for 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）

• 要点

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

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

• 对比

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

2. 并行：多车道同时行驶（多核 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 密集型

• 要点

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

• 优化策略

类型	优化方案	Python 工具
IO 密集型	异步非阻塞、多线程	asyncio, aiohttp
CPU 密集型	多进程、分布式计算	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 = 0for i in range(10000000):result += iprint("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 原理

• 要点

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

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

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

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

•  示例

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

python

import asyncioasync 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

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

https://download.csdn.net/download/ylfhpy/90406977