神剑山庄资源网 Design By www.hcban.com

一、需求分析

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

为空的记录,需要写入到日志文件中!

任务分工

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

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

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

二、完整代码

#!/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 = "\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 多进程和协程配合使用写入数据的详细内容,更多关于python 多进程和协程的资料请关注其它相关文章!

标签:
python,多进程,python,协程,python,写入数据

神剑山庄资源网 Design By www.hcban.com
神剑山庄资源网 免责声明:本站文章均来自网站采集或用户投稿,网站不提供任何软件下载或自行开发的软件! 如有用户或公司发现本站内容信息存在侵权行为,请邮件告知! 858582#qq.com
神剑山庄资源网 Design By www.hcban.com

评论“python 多进程和协程配合使用写入数据”

暂无python 多进程和协程配合使用写入数据的评论...

《魔兽世界》大逃杀!60人新游玩模式《强袭风暴》3月21日上线

暴雪近日发布了《魔兽世界》10.2.6 更新内容,新游玩模式《强袭风暴》即将于3月21 日在亚服上线,届时玩家将前往阿拉希高地展开一场 60 人大逃杀对战。

艾泽拉斯的冒险者已经征服了艾泽拉斯的大地及遥远的彼岸。他们在对抗世界上最致命的敌人时展现出过人的手腕,并且成功阻止终结宇宙等级的威胁。当他们在为即将于《魔兽世界》资料片《地心之战》中来袭的萨拉塔斯势力做战斗准备时,他们还需要在熟悉的阿拉希高地面对一个全新的敌人──那就是彼此。在《巨龙崛起》10.2.6 更新的《强袭风暴》中,玩家将会进入一个全新的海盗主题大逃杀式限时活动,其中包含极高的风险和史诗级的奖励。

《强袭风暴》不是普通的战场,作为一个独立于主游戏之外的活动,玩家可以用大逃杀的风格来体验《魔兽世界》,不分职业、不分装备(除了你在赛局中捡到的),光是技巧和战略的强弱之分就能决定出谁才是能坚持到最后的赢家。本次活动将会开放单人和双人模式,玩家在加入海盗主题的预赛大厅区域前,可以从强袭风暴角色画面新增好友。游玩游戏将可以累计名望轨迹,《巨龙崛起》和《魔兽世界:巫妖王之怒 经典版》的玩家都可以获得奖励。