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 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. 多线程竞争访问控制

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

• 要点

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

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. 线程安全与互斥锁

• 要点

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

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

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. 同步、异步、阻塞、非阻塞

• 要点

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

python

# 同步阻塞（requests 库）
import requests
response = requests.get("https://example.com")  # 阻塞直到响应返回

# 异步非阻塞（asyncio + aiohttp）
import aiohttp, asyncio

async 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, 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. 线程与进程的并发/并行

• 要点

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 = 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）

• 要点

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 的限制。

• 优化策略

•  示例

在下面代码中，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 原理

• 要点

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

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

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