面试题：Python多线程与异步I/O结合优化复杂业务场景

理论说明

计算密集型任务与多线程：
- 计算密集型任务在Python中由于全局解释器锁（GIL）的存在，多线程并不能充分利用多核CPU的优势。然而，在单核心上，多线程可以通过切换执行来避免因等待I/O等操作造成的CPU空闲。对于计算密集型任务，可以将其分配到多个线程中，利用线程的切换机制，在一个线程计算时，其他线程有机会执行。
I/O操作与异步I/O（asyncio）：
- 异步I/O（如asyncio库）通过事件循环机制，允许程序在等待I/O操作完成时，不阻塞其他任务的执行。在I/O操作开始时，程序将控制权交回事件循环，事件循环可以调度其他异步任务执行。当I/O操作完成时，事件循环再将控制权交回相应的任务，继续执行后续代码。这使得在大量I/O操作的场景下，系统可以高效地利用CPU资源，提高整体性能。
结合使用优势：
- 将多线程与异步I/O结合，可以充分发挥两者的优势。对于计算密集型任务，通过多线程在单核心上进行并发执行；对于I/O操作，利用异步I/O的非阻塞特性，减少I/O等待时间，提高系统的整体并发处理能力。
线程间资源共享：
- 使用锁（threading.Lock）来保护共享资源。当一个线程需要访问共享资源时，它必须先获取锁，访问完成后释放锁，这样可以避免多个线程同时访问共享资源导致的数据不一致问题。
- 还可以使用队列（queue.Queue）来在线程间安全地传递数据，队列本身是线程安全的，避免了直接共享数据带来的同步问题。
异步任务调度：
- asyncio库通过事件循环来调度异步任务。asyncio.create_task()函数可以创建一个异步任务并将其添加到事件循环中。事件循环会按照任务的优先级、执行状态等因素进行调度，优先执行可以执行的任务，暂停等待I/O的任务，从而实现高效的异步任务调度。

实际代码实现

import asyncio
import threading
import queue
import time


# 模拟计算密集型任务
def compute_task(num):
    result = 0
    for i in range(10000000):
        result += i * num
    return result


# 模拟I/O密集型任务
async def io_task(num):
    await asyncio.sleep(1)  # 模拟I/O等待
    return num * 2


# 线程执行函数
def thread_worker(task_queue, result_queue):
    while True:
        task = task_queue.get()
        if task is None:
            break
        result = compute_task(task)
        result_queue.put(result)
        task_queue.task_done()


async def main():
    task_queue = queue.Queue()
    result_queue = queue.Queue()

    # 创建并启动线程
    num_threads = 2
    threads = []
    for _ in range(num_threads):
        t = threading.Thread(target=thread_worker, args=(task_queue, result_queue))
        t.start()
        threads.append(t)

    # 提交计算密集型任务到线程
    for i in range(5):
        task_queue.put(i)

    # 提交异步I/O任务
    io_tasks = [io_task(i) for i in range(5)]
    io_results = await asyncio.gather(*io_tasks)

    # 获取计算密集型任务结果
    compute_results = []
    for _ in range(5):
        compute_results.append(result_queue.get())
        result_queue.task_done()

    # 停止线程
    for _ in range(num_threads):
        task_queue.put(None)
    for t in threads:
        t.join()

    print("计算密集型任务结果:", compute_results)
    print("I/O密集型任务结果:", io_results)


if __name__ == "__main__":
    asyncio.run(main())

在上述代码中：

compute_task函数模拟计算密集型任务，在多线程中执行。
io_task函数模拟I/O密集型任务，使用asyncio库以异步方式执行。
thread_worker函数是线程执行的函数，从任务队列中获取计算任务并执行，将结果放入结果队列。
在main函数中，创建线程并提交计算任务到线程，同时提交异步I/O任务。最后获取并打印两种任务的结果。通过这种方式实现了多线程与异步I/O的结合使用，同时处理了线程间资源共享（通过任务队列和结果队列）和异步任务调度（通过asyncio的事件循环）的问题。

星途面试题库

面试题：Python多线程与异步I/O结合优化复杂业务场景

知识考点

面试题答案

理论说明

实际代码实现