Python 多线程处理任务实例

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)