Python并行处理技巧:利用多线程加速For循环

Jasper
Jasper
发布时间:2026-10-13 | 5人浏览过
原创

本文探讨如何在Python中通过多线程技术加速耗时的For循环操作。针对大规模数据(如千万级记录)处理场景,通过示例展示如何将串行处理改为并行处理,利用multiprocessing.dummy模块中的ThreadPool实现并发执行,显著减少运行时间,并提供具体代码实现与性能对比结果。

大规模数据处理的串行瓶颈

以某城市出租车数据为例,单日记录量高达33210000条。若需将每辆车的数据单独提取并保存至独立文件,该如何高效实现?

基本思路如下:

遍历全部33210000条记录,将每条数据写入对应的专属文件中。

然而,面对3000多万条数据,串行逐个循环处理耗时极长。实测中,处理60万数据耗时2小时,推算处理全部3000万数据需约100小时(即4-5天)。此外,还需确保设备连续运行五天且无崩溃风险。

并行For循环的设计思路

因此,引入并行处理For循环的技巧成为必要方案:

由于3000万数据直接存入单个CSV文件会导致文件过大无法打开,故使用split工具将原CSV分割为53个文件,每个文件约60万条记录。

原方案逻辑为:读取文件夹,获取所有60万条记录的CSV文件列表,再逐一处理。这本质上仍是33210000次循环。并行For循环的核心在于同时处理多个60万条记录的CSV文件,从而成倍降低总耗时。

并行For循环的设计灵感源自以下方法:

传统串行处理方式类似如下代码结构:


  words = ['apple', 'bananan', 'cake', 'dumpling']
  for word in words:
    print word

并行For循环的实现方式如下:


from multiprocessing.dummy import Pool as ThreadPool
items = list()
pool = ThreadPool()
pool.map(process, items)
pool.close()
pool.join()

其中,process为实际执行处理的函数。

Python多线程并行处理实现

具体实例代码如下:


# -*- coding: utf-8 -*-
import time
from multiprocessing.dummy import Pool as ThreadPool
def process(item):
  print('正在并行for循环')
  print(item)
  time.sleep(5)
items = ['apple', 'bananan', 'cake', 'dumpling']
pool = ThreadPool()
pool.map(process, items)
pool.close()
pool.join()

补充知识:在Python3中,可使用多线程替代传统For循环以提升程序运行速度。

性能优化对比与结果分析

优化前后的代码对比如下:


from git_tools.git_tool import get_collect_projects, QQNews_Git
from threading import Thread, Lock
import datetime

base_url = "http://git.xx.com"
project_members_commits_lang_info = {}
lock = Lock()
threads = []

'''
Author:zenkilan
'''

def count_time(func):
  def took_up_time(*args, **kwargs):
    start_time = datetime.datetime.now()
    ret = func(*args, **kwargs)
    end_time = datetime.datetime.now()
    took_up_time = (end_time - start_time).total_seconds()
    print(f"{func.__name__} execution took up time:{took_up_time}")
    return ret

  return took_up_time

def get_project_member_lang_code_lines(git, member, begin_date, end_date):
  global project_members_commits_lang_info
  global lock
  member_name = member["username"]
  r = git.get_user_info(member_name)
  if not r["id"]:
    return
  user_commits_lang_info = git.get_commits_user_lang_diff_between(r["id"], begin_date, end_date)
  if len(user_commits_lang_info) == 0:
    return
  lock.acquire()
  project_members_commits_lang_info.setdefault(git.project, dict())
  project_members_commits_lang_info[git.project][member_name] = user_commits_lang_info
  lock.release()


def get_project_lang_code_lines(project, begin_date, end_date):
  global threads
  git = QQNews_Git(project[1], base_url, project[0])
  project_members = git.get_project_members()
  if len(project_members) == 0:
    return
  for member in project_members:
    thread = Thread(target=get_project_member_lang_code_lines, args=(git, member, begin_date, end_date))
    threads.append(thread)
    thread.start()

@count_time
def get_projects_lang_code_lines(begin_date, end_date):
  """
  获取项目代码行语言相关统计——新方法(提升效率)
  应用多线程替代for循环
  并发访问共享外部资源
  :return:
  """
  global project_members_commits_lang_info
  global threads
  for project in get_collect_projects():
    thread = Thread(target=get_project_lang_code_lines, args=(project, begin_date, end_date))
    threads.append(thread)
    thread.start()

@count_time
def get_projects_lang_code_lines_old(begin_date, end_date):
  """
  获取项目代码行语言相关统计——老方法(耗时严重)
  使用最基本的思路进行编程
  双层for循环嵌套并且每层都包含耗时操作
  :return:
  """
  project_members_commits_lang_info = {}
  for project in get_collect_projects():
    git = QQNews_Git(project[1], base_url, project[0])
    project_members = git.get_project_members()
    user_commits_lang_info_dict = {}
    if len(project_members) == 0:
      continue
    for member in project_members:
      member_name = member["username"]
      r = git.get_user_info(member_name, debug=False)
      if not r["id"]:
        continue
      try:
        user_commits_lang_info = git.get_commits_user_lang_diff_between(r["id"], begin_date, end_date)
        if len(user_commits_lang_info) == 0:
          continue
        user_commits_lang_info_dict[member_name] = user_commits_lang_info
        project_members_commits_lang_info[git.project] = user_commits_lang_info_dict
      except:
        pass
  return project_members_commits_lang_info

def test_results_equal(resultA, resultB):
  """
  测试方法
  :param resultA:
  :param resultB:
  :return:
  """
  print(resultA)
  print(resultB)
  assert len(str(resultA)) == len(str(resultB))


if __name__ == '__main__':
  from git_tools.config import begin_date, end_date

  get_projects_lang_code_lines(begin_date, end_date)
  for t in threads:
    t.join()
  old_result = get_projects_lang_code_lines_old(begin_date, end_date)
  test_results_equal(old_result, project_members_commits_lang_info)

老方法中,外层For循环和内层For循环均包含耗时操作:

1)git.get_project_members()

2)git.get_user_info(member_name, debug=False)

优化分两步进行,先内层后外层或先外层后内层均可。采用多线程替换For循环,并发共享外部资源,并通过加锁机制避免写入冲突。

测试结果验证通过,函数运行时间装饰器显示结果如下(单位:秒):

get_projects_lang_code_lines execution took up time:1.85294

get_projects_lang_code_lines_old execution took up time:108.604177

速度提升约58倍。

以上即为提高Python中For循环效率的完整分享,供读者参考,欢迎支持本站。

本站声明

本文内容仅供参考,不代表本站观点。转载或引用请注明出处。如发现内容问题,请通过联系我们页面反馈。

Python 相关产品和应用

全部下载

专题列表