python多线程巨大的csv文件

发布时间：2020-12-20 12:01:36 所属栏目：Python 来源：网络整理

导读：我有一个函数,可以懒惰地从巨大的csv文件中生成行 def get_next_line(): with open(sample_csv,'r') as f: for line in f: yield linedef do_long_operation(row): print('Do some operation that takes a long time') 我需要使用线程,这样我从上面的函数中

我有一个函数,可以懒惰地从巨大的csv文件中生成行

def get_next_line():
    with open(sample_csv,'r') as f:
        for line in f:
            yield line

def do_long_operation(row):
    print('Do some operation that takes a long time')

我需要使用线程,这样我从上面的函数中得到的每条记录都可以调用
do_long_operation

互联网上的大多数地方都有这样的例子,我不确定我是否走在正确的道路上

import threading
thread_list = []
for i in range(8):
   t = threading.Thread(target=do_long_operation,args=(get_next_row from get_next_line) )
   thread_list.append(t)

for thread in thread_list:
    thread.start()

for thread in thread_list:
    thread.join()

我的问题是

a)我怎么开始只说有限数量的线程说8？

b)如何确保每个线程都获得一行
来自get_next_line？

解决方法

您可以使用多处理中的线程池并将任务映射到工作池：

from multiprocessing.pool import ThreadPool as Pool
# from multiprocessing import Pool
from random import randint
from time import sleep


def process_line(l):
    print l,"started"
    sleep(randint(0,3))
    print l,"done"


def get_next_line():
    with open("sample.csv",'r') as f:
        for line in f:
            yield line

f = get_next_line()

t = Pool(processes=8)

for i in f:
    t.map(process_line,(i,))

t.join()
t.close()

这将创建8名工作人员,并逐一向他们提交您的行.一旦进程“免费”,它将被分配一个新任务.

还有一个注释掉的import语句.如果您注释掉ThreadPool并从多处理导入池,您将获得子进程而不是线程,这在您的情况下可能更有效.

汉努

（编辑：李大同）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!