面试题：Python concurrent.futures模块高级应用之自定义资源分配与异常处理

自定义资源分配策略

资源池设计：可以使用字典或其他数据结构来维护资源池，其中键为资源类型，值为可用资源的列表或队列。例如，对于数据库连接，可以有一个 db_connection_pool 列表，对于文件句柄，可以有一个 file_handle_pool 列表。
资源分配函数：编写一个函数，根据任务需求从资源池中分配资源。该函数需要考虑资源的可用性，如果没有可用资源，可以选择等待（使用 Queue 的阻塞方法）或抛出异常。

异常捕获和处理

任务包装：使用 try - except 块包装每个任务的执行逻辑，捕获可能抛出的异常。
异常类型处理：根据不同的异常类型采取不同的处理策略。例如，对于数据库连接相关的异常，可以尝试重新获取连接并重新执行任务；对于文件操作异常，可以记录错误并跳过该文件。
全局异常处理：在使用 concurrent.futures 执行任务时，可以通过 Future 对象的 add_done_callback 方法或在获取结果时捕获异常，确保所有任务的异常都能得到妥善处理。

代码框架示例

import concurrent.futures
import time


# 模拟资源池
db_connection_pool = []
file_handle_pool = []


# 初始化资源池
def init_resource_pools():
    for _ in range(5):
        # 模拟创建数据库连接
        db_connection_pool.append("db_connection")
        # 模拟创建文件句柄
        file_handle_pool.append("file_handle")


# 从资源池获取资源
def get_resources(task_type):
    if task_type == "db_task":
        if not db_connection_pool:
            raise Exception("No available database connections")
        return db_connection_pool.pop()
    elif task_type == "file_task":
        if not file_handle_pool:
            raise Exception("No available file handles")
        return file_handle_pool.pop()
    else:
        raise ValueError("Unsupported task type")


# 释放资源回资源池
def release_resources(task_type, resource):
    if task_type == "db_task":
        db_connection_pool.append(resource)
    elif task_type == "file_task":
        file_handle_pool.append(resource)


# 模拟任务
def task(task_type):
    resource = get_resources(task_type)
    try:
        time.sleep(1)  # 模拟任务执行时间
        print(f"Task {task_type} executed successfully with {resource}")
        return f"Task {task_type} result"
    except Exception as e:
        print(f"Task {task_type} failed: {e}")
        raise
    finally:
        release_resources(task_type, resource)


def main():
    init_resource_pools()
    tasks = ["db_task", "file_task", "db_task"]
    with concurrent.futures.ThreadPoolExecutor() as executor:
        future_to_task = {executor.submit(task, task_type): task_type for task_type in tasks}
        for future in concurrent.futures.as_completed(future_to_task):
            task_type = future_to_task[future]
            try:
                result = future.result()
                print(f"Task {task_type} result: {result}")
            except Exception as e:
                print(f"Exception in task {task_type}: {e}")


if __name__ == "__main__":
    main()

上述代码实现了一个简单的资源分配和任务执行框架。通过自定义资源池和资源分配函数，以及在任务执行过程中捕获和处理异常，确保任务执行的稳定性和正确性。在实际应用中，需要根据具体的资源类型和异常类型进行更细致的处理。

星途面试题库

面试题：Python concurrent.futures模块高级应用之自定义资源分配与异常处理

知识考点

面试题答案

自定义资源分配策略

异常捕获和处理

代码框架示例