一、需求分析
有一批key已经写入到3个txt文件中，每一个txt文件有30万行记录。 现在需要读取这些txt文件，判断key是否在数据仓库中。(redis或者mysql)
为空的记录，需要写入到日志文件中！
任务分工
1. 使用多进程技术，每一个进程读取一个txt文件
2. 使用协程技术，批量读取txt文件记录。比如一次性读取 2000条记录
注意：打开文件操作，最好在一个进程中，重复打开文件，会造成系统资源浪费！
二、完整代码
代码语言：javascript复制#!/usr/bin/env python3
# coding: utf-8
"""
多线程和协程配合使用示例
"""
import os
import time
from gevent import monkey;monkey.patch_all()
from gevent.pool import Pool
from functools import partial
from multiprocessing import Process
COROUTINE_NUMBER = 2000 # 协程池数量
pool = Pool(COROUTINE_NUMBER) # 使用协程池
# 模拟数据仓库，测试数据
data_dict = {"1":"x1","3":"x3","5":"x5","7":"x7","9":"x9"}
class TestProgram(object): # 测试程序
def __init__(self):
self.BASE_DIR = os.path.dirname(os.path.abspath(__file__)) # 项目根目录
def write_log(self,number, content, colour='white', skip=False):
"""
写入日志文件
:param content: 写入内容
:param colour: 颜色
:param skip: 是否跳过打印时间
:return:
"""
# 颜色代码
colour_dict = {
'red': 31, # 红色
'green': 32, # 绿色
'yellow': 33, # 黄色
'blue': 34, # 蓝色
'purple_red': 35, # 紫红色
'bluish_blue': 36, # 浅蓝色
'white': 37, # 白色
}
choice = colour_dict.get(colour) # 选择颜色
path = os.path.join(self.BASE_DIR, "output_%s.log" % number) # 日志文件
with open(path, mode='a ', encoding='utf-8') as f:
if skip is False: # 不跳过打印时间时
content = time.strftime('%Y-%m-%d %H:%M:%S') ' ' content
info = " 33[1;{};1m{} 33[0m".format(choice, content)
print(info)
f.write(content "n")
def has_null(self, key, number):
"""
输出key
:param key: 键值
:param number: 文件标记
:return: bool
"""
key = key.strip()
if not data_dict.get(key):
self.write_log(number,"错误，{} 记录为空".format(key),"red")
return False
print(key)
return True
def read_file(self, number):
"""
读取文件
:param number: 文件标记
:return:
"""
file_name = os.path.join(self.BASE_DIR, "data", "%s.txt" % number)
# print(file_name)
self.write_log(number, "开始读取文件 {}".format(file_name),"green")
with open(file_name, encoding='utf-8') as f:
# 使用协程池，执行任务。语法: pool.map(func,iterator)
# partial使用偏函数传递参数
# 注意：has_null第一个参数,必须是迭代器遍历的值
pool.map(partial(self.has_null, number=number), f)
self.write_log(number, "结束文件读取 {} 完成".format(file_name),"green")
return True
def run(self, number):
"""
读取指定的文件，判断每一个key是否为空
:param number:
:return:
"""
startime = time.time() # 开始时间
# 清空日志
path = os.path.join(self.BASE_DIR, "output_%s.log" % number) # 日志文件
with open(path, mode='w') as f:
pass
self.read_file(number)
endtime = time.time()
take_time = endtime - startime
if take_time < 1: # 判断不足1秒时
take_time = 1 # 设置为1秒
# 计算花费时间
m, s = divmod(take_time, 60)
h, m = divmod(m, 60)
self.write_log(number, "%s.txt 花费时间 d:d:d" % (number,h, m, s),"green")
def main(self):
"""
使用多线程执行程序
:return:
"""
# 文件标记列表
file_list = ["7001", "7002", "7003"]
p_lst = [] # 线程列表
for i in file_list:
# self.run(i)
p = Process(target=self.run, args=(i,)) # 子进程调用函数
p.start() # 启动子进程
p_lst.append(p) # 将所有进程写入列表中
for p in p_lst: p.join() # 检测p是否结束,如果没有结束就阻塞直到结束,否则不阻塞
TestProgram().main() # 启动主程序，它会开启3个进程。执行输出：
