Python多线程处理Excel数据导入实战：从单线程到线程池重构

背景与问题描述

美餐每日生成一份使用Excel汇总的就餐数据报表，我们将这些数据导入数据库后，行政办公服务系统会利用其内部就餐数据进行比对和查重处理。

最初的实现方案采用单线程模式，其代码结构与import_records模块在移除多线程逻辑后的状态大致相同。

主要处理流程为：读取数据，随后发送至行政服务接口进行后续处理。

出于安全考虑，线上环境的数据导入操作被安排在夜间执行。运行过程中发现，每条数据的导入耗时超过1秒，若从晚上十点开始处理数千条数据，预计耗时将非常漫长，令人担忧。

多线程优化思路

在等待执行的过程中，下楼透气时因室内空气浑浊感到昏沉，寒冷反而让人清醒。此时突然想到，为何不采用多线程技术来加速处理？

代码重构与实现

第一版多线程实现将业务逻辑与线程控制代码混杂在一起，导致代码可读性极差。随后两天，我抽时间对代码进行了多次重构，最终分离出了独立的线程池、迭代器以及import_records模块。

重构后代码结构清晰许多，但迭代器被暴露出来，需要由import_records模块调用以判断当前任务是否应由当前线程处理，这种设计思路类似于协程机制。

虽然暴露迭代器既有优势也有劣势，但目前已基本满足日常使用需求，可以暂时搁置进一步优化。闲暇时读读书、看看电影，也是一种享受。

import threading

def task_pool(thread_num, task_fn):

  if thread_num <= 0 :
      raise ValueError

  threads = []

  def gen_thread_checker(thread_id, step):

      base = 1
      i = 0

      def thread_checker():
          nonlocal i

          i += 1
          # print((thread_id,i,step, i < base or (i - base) % step != thread_id))

          if i < base or (i - base) % step != thread_id:
              return False

          return True

      return thread_checker


  for x in range(0, thread_num):
    threads.append(threading.Thread(target=task_fn, args=(x,thread_num, gen_thread_checker(x, thread_num))))

  # 启动所有线程
  for t in threads:
    t.start()
  # 主线程中等待所有子线程退出
  for t in threads:
    t.join()

import argparse
import re

import requests
from openpyxl import load_workbook
from requests import RequestException

import myThread

parser = argparse.ArgumentParser(description='美餐到店交易数据导入')
parser.add_argument('--filename', '-f', help='美餐到店交易数据 .xlsx 文件路径', required=True)
parser.add_argument('--thread_num', '-t', help='线程数量', default= 100, required=False)
parser.add_argument('--debug', '-d', help='调试模式', default= 0, required=False)
args = parser.parse_args()

filename = args.filename
thread_num = int(args.thread_num)
debug = args.debug

if debug:
    print((filename,thread_num,debug))


def add_meican_meal_record(data):
   pass

def import_records(thread_id, thread_number, thread_checker):
    wb = load_workbook(filename=filename)
    ws = wb.active

    for row in ws:
        #------------------------------------------
        if row[0].value is None:
            break

        if not thread_checker():
            continue
        #------------------------------------------

        if row[0].value == '日期' or row[0].value == '总计' or not re.findall('^\d{4}-\d{1,2}-\d{1,2}$', row[0].value):
            continue
        else:

            date = str.replace(row[0].value,'-', '')

            order_id = row[3].value
            restaurant_name = row[5].value
            meal_plan_name = row[6].value
            meal_staffid = row[10].value
            identify = row[11].value
    
            add_meican_meal_record({
                'orderId':order_id,
                'date': date,
                'meal_plan_name':meal_plan_name,
                'meal_staffid':meal_staffid,
                'identify':identify,
                'restaurant_name':restaurant_name
            })

myThread.task_pool(thread_num,import_records)

本文关于Python多线程处理任务实例的介绍到此结束，更多相关Python多线程处理内容请搜索本站以前的文章或继续浏览下面的相关文章，希望大家以后多多支持本站！