Python中joblib库并行求解多个方程组

并行求解程序一

# -*- coding: utf-8 -*-import  numpy as np
from numba import jit
from joblib import Parallel,delayed
from timeit import default_timer as timer
import timedef solve_system_memory_friendly(c_k, F_k, M, A):"""内存友好的求解函数"""# 直接使用M和A，避免创建临时的大矩阵# 对于特殊结构的矩阵，可以实现更高效的求解# 这里使用直接求解，对于大矩阵应该使用迭代法coefficient_matrix = c_k * M + AU_k = np.linalg.solve(coefficient_matrix, F_k)return U_kdef parallel_solve_memory_friendly(c_list, F_list, M, A, n_jobs=11):"""内存友好的并行求解"""# 使用列表推导式准备参数solutions = Parallel(n_jobs=n_jobs, verbose=0)(delayed(solve_system_memory_friendly)(c_k, F_k, M, A) for c_k, F_k in zip(c_list, F_list))total_sum = np.sum(solutions, axis=0)return total_sum, solutions
def verify_solution(c_list, F_list, M, A, solutions, total_sum):"""验证求解的正确性"""print("验证求解结果:")for i, (c_k, F_k, U_k) in enumerate(zip(c_list, F_list, solutions)):residual = np.linalg.norm((c_k * M + A) @ U_k - F_k)print(f"方程组 {i+1} 残差: {residual:.2e}")print(f"累加和范数: {np.linalg.norm(total_sum):.2e}")# 在主要代码中使用验证
if __name__ == "__main__":# 设置随机种子以便复现np.random.seed(42)n = 100  # 小一点的维度用于测试M = np.random.rand(n, n) + 10 * np.eye(n)A = np.random.rand(n, n) + 10 * np.eye(n)m=1c_list = range(m)#[1.0, 2.0, 3.0, 4.0, 5.0]F_list = [np.random.rand(n) for _ in range(m)]# 求解start = time.time()total_sum, solutions = parallel_solve_memory_friendly(c_list, F_list, M, A)print(f'Time consuming:{time.time() - start}')print(total_sum)print(solutions)# 验证#verify_solution(c_list, F_list, M, A, solutions, total_sum)

并行求解程序二

# -*- coding: utf-8 -*-import  numpy as np
from numba import jit
from joblib import Parallel,delayed
from timeit import default_timer as timer
import timedef solve_system_memory_friendly(c_k, F, M, A):"""内存友好的求解函数"""# 直接使用M和A，避免创建临时的大矩阵# 对于特殊结构的矩阵，可以实现更高效的求解# 这里使用直接求解，对于大矩阵应该使用迭代法coefficient_matrix = c_k * M + AU_k = np.linalg.solve(coefficient_matrix, F)return U_kdef parallel_solve_memory_friendly(c_list, F, M, A, n_jobs=11):"""内存友好的并行求解"""# 使用列表推导式准备参数solutions = Parallel(n_jobs=n_jobs, verbose=0)(delayed(solve_system_memory_friendly)(c_k, F, M, A) for c_k in zip(c_list))total_sum = np.sum(solutions, axis=0)return total_sum, solutions
# 在主要代码中使用验证
if __name__ == "__main__":# 设置随机种子以便复现np.random.seed(42)n = 100  # 小一点的维度用于测试M = np.random.rand(n, n) + 10 * np.eye(n)A = np.random.rand(n, n) + 10 * np.eye(n)m=10c_list = range(m)#[1.0, 2.0, 3.0, 4.0, 5.0]F = np.random.rand(n)# 求解start = time.time()total_sum, solutions = parallel_solve_memory_friendly(c_list, F, M, A)print(f'Time consuming:{time.time() - start}')print(total_sum)print(solutions)# 验证#verify_solution(c_list, F_list, M, A, solutions, total_sum)