python使用多进程和协程实现写入数据?很多新手对此不是很清楚,为了帮助大家解决这个难题,下面小编将为大家详细讲解,有这方面需求的人可以来学习下,希望你能有所收获。
<强>一、需求分析
有一批关键已经写入到3个txt文件中,每一个txt文件有30多万行记录。
现在需要读取这些txt文件,判断键是否在数据仓库中。(复述,或者mysql)
为空的记录,需要写入到日志文件中!
任务分工
1。使用多进程技术,每一个进程读取一个txt文件
2。使用协程技术,批量读取txt文件记录。比如一次性读取2000条记录
注意:打开文件操作,最好在一个进程中,重复打开文件,会造成系统资源浪费!
<强>二,完整代码
# !/usr/bin/env python3
utf - 8编码:
“““
多线程和协程配合使用示例
“““
进口操作系统
导入的时间
从gevent进口猴;monkey.patch_all ()
从gevent。池进口池
从functools导入部分
从多处理导入过程
COROUTINE_NUMBER=2000 #协程池数量
池=池(COROUTINE_NUMBER) #使用协程池
#模拟数据仓库,测试数据
data_dict={“1”:“x1",“3”:“x3",“5”:“x5",“7”:“x7",“9”:“x9"}
类TestProgram(对象):#测试程序
def __init__(自我):
自我。BASE_DIR=os.path.dirname (os.path.abspath (__file__)) #项目根目录
def write_log(自我、数量、内容、色彩=& # 39;白色# 39;,跳过=False):
“““
写入日志文件
:param内容:写入内容
:param颜色:颜色
:param跳过:是否跳过打印时间
返回:
“““
#颜色代码
colour_dict={
& # 39;红色# 39;:31日#红色
& # 39;绿色# 39;:32 #绿色
& # 39;黄色# 39;:33 #黄色
& # 39;蓝# 39;:34岁#蓝色
& # 39;purple_red& # 39;: 35 #紫红色
& # 39;bluish_blue& # 39;: 36 #浅蓝色
& # 39;白色# 39;:37岁#白色
}
选择=colour_dict.get(颜色)#选择颜色
路径=os.path.join(自我。BASE_DIR,“output_ % s.log"%号)#日志文件
张开(路径、模式=& # 39;a + & # 39;,编码=& # 39;utf - 8 # 39;) f:
如果跳过是假的:#不跳过打印时间时
内容=time.strftime (& # 39; Y % - % - % d % H: % m: % & # 39;) + & # 39;& # 39;+内容
信息=?33 \ [1,{};1 033 {}\ [0 m"。格式(选择、内容)
打印(信息)
f。写(内容+“\ n")
def has_null(自我、关键、数量):
“““
输出关键
:param关键:键值
:param号码:文件标记
返回:bool
“““
关键=key.strip ()
如果不是data_dict.get(重要):
self.write_log(数字,“错误,{}记录为空“.format(关键),“red")
返回假
打印(关键)
还真
def read_file(自我、数量):
“““
读取文件
:param号码:文件标记
返回:
“““
file_name=os.path.join(自我。BASE_DIR,“data",“% s.txt"%数量)
#打印(file_name)
自我。write_log(数字,“开始读取文件{}“.format (file_name),“green")
张开(file_name编码=& # 39;utf - 8 # 39;) f:
#使用协程池,执行任务。语法:pool.map(函数迭代器)
#部分使用偏函数传递参数
#注意:has_null第一个参数,必须是迭代器遍历的值
pool.map(部分(自我。has_null数=数),f)
自我。write_log(数字,“结束文件读取{}完成“.format (file_name),“green")
还真
def运行(自我、数量):
“““
读取指定的文件,判断每一个关键是否为空
:param号码:
返回:
“““
startime=time.time() #开始时间
#清空日志
路径=os.path.join(自我。BASE_DIR,“output_ % s.log"%号)#日志文件
张开(路径、模式=& # 39;w # 39;) f:
通过
self.read_file(数量)
endtime=time.time ()
take_time=endtime - startime
如果take_time & lt;1:#判断不足1秒时
take_time=1 #设置为1秒
#计算花费时间
m s=divmod (take_time 60)
h m=divmod(米、60)
自我。write_log(数字,“% s。三花费时间% 2 d: % 2 d: % 02 d"%(数量、h m s),“green")
def主要(自我):
“““
使用多线程执行程序
返回:
“““
#文件标记列表
file_list=(“7001”,“7002”,“7003“)
p_lst=[] #线程列表
因为我在file_list:
# self.run(我)
p=过程(目标=self.run args=(我))#子进程调用函数
p.start() #启动子进程
p_lst.append (p) #将所有进程写入列表中
在p_lst p: p.join() #检测p是否结束,如果没有结束就阻塞直到结束,否则不阻塞
.main TestProgram()() #启动主程序,它会开启3个进程。
