怎么使用python3线程池ThreadPoolExecutor处理csv文件数据

这篇文章主要介绍“怎么使用python3线程池ThreadPoolExecutor处理csv文件数据”的相关知识，小编通过实际案例向大家展示操作过程，操作方法简单快捷，实用性强，希望这篇“怎么使用python3线程池ThreadPoolExecutor处理csv文件数据”文章能帮助大家解决问题。

背景

由于不同乙方对服务商业务接口字段理解不一致，导致线上上千万数据量数据存在问题，为了修复数据，通过 Python 脚本进行修改

知识点

Python3、线程池、pymysql、CSV 文件操作、requests

拓展

当我们程序在使用到线程、进程或协程的时候，以下三个知识点可以先做个基本认知

CPU 密集型、io 密集型、GIL 全局解释器锁

库

pip3 install requests

pip3 install pyMysql

流程

实现代码

# -*- coding:utf-8 -*-# @FileName:grade_update.py# @Desc    :在一台超级计算机上运行过的牛逼Python代码import timefrom concurrent.futures import ThreadPoolExecutor,FIRST_COMPLETED,waitimport requestsimport pymysqlfrom projectPath import pathgradeId = [4303, 4304, 1000926, 1000927]def writ_mysql():    """    数据库连接    """    return pymysql.connect(host="localhost",                         port=3306,                         user="admin",                         passWord="admin",                         database="test"                         )def oprationdb(grade_id, member_id):  """  操作数据库  """    db = writ_mysql()    try:        cursor = db.cursor()        sql = f"UPDATE `t_m_member_grade` SET `current_grade_id`={grade_id}, `modified` =now() WHERE `member_id`={member_id};"        cursor.execute(sql)        db.commit()        print(f"提交的SQL->{sql}")    except pymysql.Error as e:        db.rollback()        print("DB数据库异常：", e)    db.close()    return Truedef interface(rows, thead):  """  调用第三方接口  """    print(f"处理数据行数--->{thead}----数据--->{rows}")    try:        url = "Http://xxxx/api/xxx-data/Tmall/bindQuery"        body = {            "nickname": str(rows[0]),            "seller_name": "test",            "mobile": "111"        }        heade={"Content-Type": "application/x-www-fORM-urlencoded"}        res = requests.post(url=url, data=body,headers=heade)        result = res.JSON()        if result["data"]["status"] in [1, 2]:            grade = result["data"]["member"]["level"]            grade_id = gradeId[grade]            oprationdb(grade_id=grade_id, member_id=rows[1])            return True        return True    except Exception as e:        print(f"调用异常:{e}")def read_csv():    import csv    # db = writ_mysql()    #线程数    MAX_WORKERS=5    with ThreadPoolExecutor(MAX_WORKERS) as pool:        with open(path + '/file/result2_colu.csv', 'r', newline='', encoding='utf-8') as f:            #set() 函数创建无序不重复元素集            seq_notdone = set()            seq_done = set()            # 使用csv的reader()方法，创建一个reader对象            reader = csv.reader(f)            n = 0            for row in reader:                n += 1                # 遍历reader对象的每一行                try:                    seq_notdone.add(pool.submit(interface, rows=row, thead=n))                    if len(seq_notdone) >= MAX_WORKERS:                        #FIRST_COMPLETED文档说明 -- Return when any future finishes or is cancelled.                        done, seq_notdone = wait(seq_notdone,return_when=FIRST_COMPLETED)                        seq_done.update(done)                except Exception as e:                    print(f"解析结果出错：{e}")    # db.close()    return "完成"if __name__ == '__main__':    read_csv()

解释

引入线程池库

from concurrent.futures import ThreadPoolExecutor,FIRST_COMPLETED,wait

pool.submit(interface, rows=row, thead=n)

提交任务，interface 调用的函数，rows、thead 为 interface() 函数的入参

任务持续提交，线程池通过 MAX_WORKERS 定义的线程数持续消费

说明像这种 I/O 密集型的操作脚本适合使用多线程，如果是 CPU 密集型建议使用进行，根据机器核数进行配置

关于“怎么使用python3线程池ThreadPoolExecutor处理csv文件数据”的内容就介绍到这里了，感谢大家的阅读。如果想了解更多行业相关的知识，可以关注编程网Python频道，小编每天都会为大家更新不同的知识点。