在数据量特别大，并且数据处理速度远高于数据读取速度时，可以考虑多线程读取数据+多线程处理数据的模式，如果单线程读取速度也已经可以满足数据处理的需要，也可以采用主线程读取数据+多线程处理数据的模式
为了说明起来更简单，这里使用python语言进行说明
多线程读取数据
多线程读取数据可以分为两种，读取不同文件和读取相同文件。
（不过具体情况具体分析，根据硬盘的读性能，有可能并发读文件的效率并没有单线程读文件的效率高）
读取不同文件：
为每个线程安排一个文件名称
# 多线程需要引入
import threading
...#这里假设读取csv文件
def read_loadfile(filename):nowline = 0with open(filename,'r') as fi:reader = csv.reader(fi)for i in reader:nowline +=1print i......def readdata(filelist):readthr=[]# 循环文件列表，这里每个文件安排一个线程for i in range(len(filelist)):#开启一个线程readthr.append(threading.Thread(target=read_loadfile, args=(quelist[i])))readthr[i].start()# 根据需要，确定所有进程是否运行完毕while True:finishread=Truefor i in readthr:if i.isAlive() == True:finishread=Falseif finishread == True:break
...
if __name__ == '__main__':...#存储需要读的文件列表filelist=[]readdata(filelist)...
读取相同文件：
多线程读取相同文件，为每个线程标注好读取的行数范围即可。
这里的多线程读取相同文件只为了多线程处理数据服务，
并不能通过多线程读取来使读取大文件速度加快
# 多线程需要引入
import threading
...#这里假设读取csv文件
def read_loadfile(filename,le,ri):nowline = 0with open(filename,'r') as fi:reader = csv.reader(fi)for i in reader:nowline +=1#当前线程读取设置好的行数if nowline > ri :breakif nowline >= le :print i......def readdata(filename):# 得到文件行数，根据行数来进行读取catloadfile = "cat "+loadfile+" | wc -l"filelines = commands.getstatusoutput(catloadfile)[1]#thrnum是读线程的数量step = int(filelines)/thrnumreadthr=[]for i in range(thrnum):le = i*step+1if i == thrnum-1 :ri = filelineselse :ri = (i+1)*step#开启一个线程readthr.append(threading.Thread(target=read_loadfile, args=(filename,le,ri)))readthr[i].start()# 根据需要，确定所有进程是否运行完毕while True:finishread=Truefor i in readthr:if i.isAlive() == True:finishread=Falseif finishread == True:break
...
if __name__ == '__main__':...#存储需要读的文件名称readdata(filename)...
多线程处理数据
队列作为多线程安全的数据结构，是可以支持不加锁，多线程同时操作的。
比如，读取一个文件，需要根据id的范围分为三种处理方式（比如乘以相关系数），那么可以使用并行处理，开启三个线程，每个线程专职处理一种情况。
这里使用一个全局变量来控制每个线程的退出（也可以由参数变量设置）
但是需要注意，如果队列中仍有数据时需要先完成任务再退出，否则容易出现数据疏漏情况。
def mak_work(i,que):while True:#假如该线程需要关闭并且其中的任务已经完成，则退出if thswi[ti] == False and que.empty() == True:break num = que.qsize()#如果当前任务则继续if num == 0:continueelse :#如果有任务，则处理完这些任务for i in range(num):qhe=que.get()...def work():#线程数量workthenum=3#线程开关（数列）global thswithswi = []#队列数组，每个数据处理线程拥有一个队列quelist = []for i in range(shardnum):#每个线程当前状态为开启thswi.append(True)#分配给每个线程一个队列，队列设置一个容量最大值防止爆内存quelist.append(Queue(thrmax))workthr=[]for i in range(workthenum):thr.append(threading.Thread(target=mak_work, args=(i,quelist[i])))thr[i].start()...#可以随时通过读数据的线程向相应的队列中传入数据with open(loadfile,'r') as fi:reader = csv.reader(fi)for i in reader:#hash函数来判断当前需要做哪种操作，使用哪个线程hi = hash(i)quelist[hi].put(i)...#关掉所有线程for i in range(shardnum):thswi[i] = False
多线程读取数据+多线程处理数据
将以上两种结合起来，将对工作线程队列的写入操作（que.put()操作），转由读取数据的线程完成
而开启读取线程和工作线程的操作由主线程完成。