python 多进程和协程配合使用写入数据

一、需求分析

有一批key已经写入到3个txt文件中，每一个txt文件有30万行记录。现在需要读取这些txt文件，判断key是否在数据仓库中。(redis或者mysql)

为空的记录，需要写入到日志文件中！

任务分工

1. 使用多进程技术，每一个进程读取一个txt文件

2. 使用协程技术，批量读取txt文件记录。比如一次性读取 2000条记录

注意：打开文件操作，最好在一个进程中，重复打开文件，会造成系统资源浪费！

二、完整代码

#!/usr/bin/env python3# coding: utf-8'''多线程和协程配合使用示例'''import osimport timefrom gevent import monkey;monkey.patch_all()from gevent.pool import Poolfrom functools import partialfrom multiprocessing import ProcessCOROUTINE_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 = '033[1;{};1m{}033[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 花费时间 %02d:%02d:%02d' % (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个进程。

执行输出

以上就是python 多进程和协程配合使用写入数据的详细内容，更多关于python 多进程和协程的资料请关注好吧啦网其它相关文章！