一、需求分析

有一批key已經(jīng)寫入到3個(gè)txt文件中，每一個(gè)txt文件有30萬(wàn)行記錄。現(xiàn)在需要讀取這些txt文件，判斷key是否在數(shù)據(jù)倉(cāng)庫(kù)中。(redis或者mysql)

為空的記錄，需要寫入到日志文件中！

任務(wù)分工

1. 使用多進(jìn)程技術(shù)，每一個(gè)進(jìn)程讀取一個(gè)txt文件

2. 使用協(xié)程技術(shù)，批量讀取txt文件記錄。比如一次性讀取 2000條記錄

注意：打開文件操作，最好在一個(gè)進(jìn)程中，重復(fù)打開文件，會(huì)造成系統(tǒng)資源浪費(fèi)！

二、完整代碼

#!/usr/bin/env python3# coding: utf-8'''多線程和協(xié)程配合使用示例'''import osimport timefrom gevent import monkey;monkey.patch_all()from gevent.pool import Poolfrom functools import partialfrom multiprocessing import ProcessCOROUTINE_NUMBER = 2000 # 協(xié)程池?cái)?shù)量pool = Pool(COROUTINE_NUMBER) # 使用協(xié)程池# 模擬數(shù)據(jù)倉(cāng)庫(kù)，測(cè)試數(shù)據(jù)data_dict = {'1':'x1','3':'x3','5':'x5','7':'x7','9':'x9'}class TestProgram(object): # 測(cè)試程序 def __init__(self): self.BASE_DIR = os.path.dirname(os.path.abspath(__file__)) # 項(xiàng)目根目錄 def write_log(self,number, content, colour=’white’, skip=False): ''' 寫入日志文件 :param content: 寫入內(nèi)容 :param colour: 顏色 :param skip: 是否跳過(guò)打印時(shí)間 :return: ''' # 顏色代碼 colour_dict = { ’red’: 31, # 紅色 ’green’: 32, # 綠色 ’yellow’: 33, # 黃色 ’blue’: 34, # 藍(lán)色 ’purple_red’: 35, # 紫紅色 ’bluish_blue’: 36, # 淺藍(lán)色 ’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: # 不跳過(guò)打印時(shí)間時(shí) 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: 文件標(biāo)記 :return: bool ''' key = key.strip() if not data_dict.get(key): self.write_log(number,'錯(cuò)誤，{} 記錄為空'.format(key),'red') return False print(key) return True def read_file(self, number): ''' 讀取文件 :param number: 文件標(biāo)記 :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: # 使用協(xié)程池，執(zhí)行任務(wù)。語(yǔ)法: pool.map(func,iterator) # partial使用偏函數(shù)傳遞參數(shù) # 注意：has_null第一個(gè)參數(shù),必須是迭代器遍歷的值 pool.map(partial(self.has_null, number=number), f) self.write_log(number, '結(jié)束文件讀取 {} 完成'.format(file_name),'green') return True def run(self, number): ''' 讀取指定的文件，判斷每一個(gè)key是否為空 :param number: :return: ''' startime = time.time() # 開始時(shí)間 # 清空日志 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秒時(shí) take_time = 1 # 設(shè)置為1秒 # 計(jì)算花費(fèi)時(shí)間 m, s = divmod(take_time, 60) h, m = divmod(m, 60) self.write_log(number, '%s.txt 花費(fèi)時(shí)間 %02d:%02d:%02d' % (number,h, m, s),'green') def main(self): ''' 使用多線程執(zhí)行程序 :return: ''' # 文件標(biāo)記列表 file_list = ['7001', '7002', '7003'] p_lst = [] # 線程列表 for i in file_list: # self.run(i) p = Process(target=self.run, args=(i,)) # 子進(jìn)程調(diào)用函數(shù) p.start() # 啟動(dòng)子進(jìn)程 p_lst.append(p) # 將所有進(jìn)程寫入列表中 for p in p_lst: p.join() # 檢測(cè)p是否結(jié)束,如果沒(méi)有結(jié)束就阻塞直到結(jié)束,否則不阻塞TestProgram().main() # 啟動(dòng)主程序，它會(huì)開啟3個(gè)進(jìn)程。

執(zhí)行輸出

以上就是python 多進(jìn)程和協(xié)程配合使用寫入數(shù)據(jù)的詳細(xì)內(nèi)容，更多關(guān)于python 多進(jìn)程和協(xié)程的資料請(qǐng)關(guān)注好吧啦網(wǎng)其它相關(guān)文章！