线程池，unittest参数化，协程

python的多线程只能利用cpu的一个核心，一个核心同时只能运行一个任务那么为什么你使用多线程的时候，它的确是比单线程快答:如果是一个计算为主的程序（专业一点称为CPU密集型程序），这一点确实是比较吃亏的，每个线程运行一遍，就相当于单线程再跑，甚至比单线程还要慢——CPU切换线程的上下文也是要有开销的。但是，如果是一个磁盘或网络为主的程序（IO密集型）就不同了。一个线程处在IO等待的时候，另一个线程还可以在CPU里面跑，有时候CPU闲着没事干，所有的线程都在等着IO，这时候他们就是同时的了,而单线程的话此时还是在一个一个等待的。我们都知道IO的速度比起CPU来是慢到令人发指的，python的多线程就在这时候发挥作用了。比方说多线程网络传输，多线程往不同的目录写文件，等等。

1、线程池就是为了防止无限启动很多线程，造成服务器压力大。2、协程        就只有一个线程。        异步IO            例子：去排队买票，你在这给他说一下你要买票，等到你的时候 他会自动把票给你，节省你排队的时间。        nginx1、从excel里面获取参数化数据

import threadingdef say(lis):   for i in lis:      print(i)

res = list(range(100))

for i in range(5):   t = threading.Thread(target=say,args=(res[i*20:(i+1)*20],))   t.start()

线程池

# 放线程的一个池子import threadpooldef say(num):   print(num)

res = list(range(101))pool = threadpool.ThreadPool(10)#创建一个线程池reqs = threadpool.makeRequests(say,res)#生成线程要执行的所有线程# for req in reqs:#  pool.putRequest(req) #实际才去执行的[ pool.putRequest(req) for req in reqs]

pool.wait() #等待 其他线程执行结束

线程池封装

import threadpool,asyncioclass MyPool(object):   def __init__(self,func,size=20,data=None):      self.func = func      self.size = size      self.data = data   def pool(self):      pool = threadpool.ThreadPool(self.size)#创建一个线程池,指定大小      reqs = threadpool.makeRequests(self.func,self.data)      #生成请求，分配数据      [pool.putRequest(req) for req in reqs]      #执行函数      pool.wait()      #等待线程执行完成def down(num):   print(num)my = MyPool(func=down,data=[1,2,3,4,5,6,7])my.pool()

unittest参数化

import unittestimport nose_parameterizedfrom ftl import DataToParamfrom MyLog import Loggerdef calc(a,b):   a = int(a)   b = int(b)   res = round(a/b,2)   print(res)   return res

class MyTest(unittest.TestCase):   @nose_parameterized.parameterized.expand(DataToParam.text('case_data.txt'))   def test_func(self,a,b,e):      res = calc(a,b)      self.assertEqual(res,int(e))

if __name__ == '__main__':   unittest.main()

从文件里面获取参数化列表

import os,xlrdclass DataToParam(object):   @classmethod   def text(cls,filename,seq=','):      cls.file_exist(filename)      with open(filename,encoding='utf-8') as f:         res = []         for line in f:            res.append(line.strip().split(seq))         return res

   @classmethod   def excel(cls,filename):      cls.file_exist(filename)      book = xlrd.open_workbook(filename) #打开excel      sheet = book.sheet_by_index(0) #获取sheet页      res = []      for row in range(sheet.nrows):  #sheet.nrows excel的行数         line_list = sheet.row_values(row) #取excel里面的每一行数据，返回的是一个list         res.append(line_list)      return res

   @classmethod   def file_exist(cls,filename):      if os.path.isfile(filename):#判断文件是否存在         return True      raise Exception('参数化文件不存在！')# print(DataToParam.text('case_data.txt'))res = DataToParam.excel(r'C:\Users\bjniuhanyang\Desktop\data.xlsx')print(res)# import json# print(json.dumps(res,indent=2))