Python使用多进程并行加速业务操作 完整代码

import logging
import math
import multiprocessing
import time
import pandas as pd
from contextlib import contextmanager
import threading
# 设置日志配置
logging.basicConfig(format='%(asctime)s %(message)s', datefmt='%m/%d/%Y %I:%M:%S %p', level=logging.INFO)
# 定义超时异常
class TimeoutException(Exception): pass
# 超时控制
@contextmanager
def time_limit(seconds):
    timer = threading.Timer(seconds, lambda: _raise_timeout_exception())
    def _raise_timeout_exception():
        raise TimeoutException("Timed out!")
    try:
        timer.start()
        yield
    finally:
        timer.cancel()
def process_data(i, data, results, lock):
    logging.info('------group: ' + str(i) + '------')
    logging.info('------len: ' + str(len(data)) + '------')
    for _, row in data.iterrows():
        if _ % (math.ceil(len(data) / 10.0)) == 0:
            logging.info('------group' + str(i) + ': ' + str(_) + '/' + str(len(data)) + '------')
        try:
            # 超时限制 TODO 秒数
            with time_limit(20):
                # 模拟任务 TODO 任务
                time.sleep(1)
                # 使用锁来保证对结果列表的进程安全访问
                lock.acquire()
                try:
                    # 将结果添加到共享的结果列表中 TODO 收集结果
                    results.append(row['id'])
                finally:
                    lock.release()
        except Exception as e:
            logging.info('------err: ' + str(e) + '------')
if __name__ == '__main__':
    # 手动设置并行进程数目 TODO 进程数目
    group_num = 8
    # 从电脑配置中设置并行进程数目
    # group_num = multiprocessing.cpu_count()
    # 读取数据 TODO 数据源
    data = pd.read_excel('data.xlsx')
    # 使用pandas平均划分数据
    grouped_data = data.groupby(data.index % group_num)
    # 定义共享的结果列表
    manager = multiprocessing.Manager()
    results = manager.list()
    # 创建锁
    lock = multiprocessing.Lock()
    start_time = time.time()
    # 定义多进程
    processes = []
    for i in range(group_num):
        p = multiprocessing.Process(target=process_data,
                                    args=(i, grouped_data.get_group(i).reset_index(), results, lock))
        processes.append(p)
    # 启动
    for _p in processes:
        _p.start()
    for _p in processes:
        _p.join()
    end_time = time.time()
    execution_time = end_time - start_time
    # 打印数据
    print(f"代码执行时间：{execution_time}秒")
    print(results)