python多线程基础

• 多线程准备脚本test01

import unittest
from time import ctime, sleep
from random import randint

class Test01(unittest.TestCase):

    def setUp(self):
        pass

    def tearDown(self):
        pass

    def test_01(self):
        print("test_01 start:",ctime())
        sleep(randint(1,8))
        print("test_01 end:",ctime())

• 多线程准备脚本test02

import unittest
from time import ctime,sleep
from random import randint

class Test02(unittest.TestCase):

    def setUp(self):
        pass

    def tearDown(self):
        pass

    def test_01(self):
        print("test_02 start:",ctime())
        sleep(randint(1,8))
        print("test_02 end:",ctime())

• 多线程准备脚本test03

import unittest
from time import ctime,sleep
from random import randint

class Test03(unittest.TestCase):

    def setUp(self):
        pass

    def tearDown(self):
        pass

    def test_01(self):
        print("test_03 start:",ctime())
        sleep(randint(1,8))
        print("test_03 end:",ctime())

• 多线程准备脚本test04

import unittest
from time import ctime,sleep
from random import randint

class Test04(unittest.TestCase):

    def setUp(self):
        pass

    def tearDown(self):
        pass

    def test_01(self):
        print("test_04 start:",ctime())
        sleep(randint(1,8))
        print("test_04 end:",ctime())

1.多线程demo

import unittest
import os
import threading
import time

# 获取到所有的符合规则的脚本
suite = unittest.TestLoader().discover(os.getcwd(), "test0*.py")
print(suite)
print(suite.countTestCases())
#解析所有的用例，并存到列表中
lst = []
for s in suite:
    for ss in s:
        for sss in ss:
            lst.append(sss)
print(len(lst))
#  
# #第一种方式：顺序执行
runer = unittest.TextTestRunner()
for l in lst:
    runer.run(l)
    
# 第二种方式：多线程方式执行
# 起3个线程，分别执行这5个脚本，但是分不够 
thread_num = 3
 
def groups(lst,thread_num):
    # 定义一个列表用于存放整个分组情况
    all_list = []
    # 根据线程数量初始化脚本列表
    for i in range(0,thread_num):
        all_list.append([])
    # 对列表进行浅复制
    tmp_list = lst.copy()
    # 循环进行脚本的分配
    while len(tmp_list) != 0:
        # 内层循环进行取值
        for i in range(0,thread_num):
            if len(tmp_list) > 0:
                temp = tmp_list[0]
                all_list[i].append(temp)
                tmp_list.remove(temp)
            else:
                break
#     # 返回分配好的脚本列表
    return all_list
# 
all_list = groups(lst, thread_num)
print(all_list)

# 定义可以运行suite列表的方法
def run_suite(suite_list):
    # 实例化一个运行器
    runner = unittest.TextTestRunner()
    # 循环执行列表中的脚本
    for suite in suite_list:
        runner.run(suite)
 
# 创建线程组池
threads_list = []
# 循环将分组脚本以多线程的方式实例化，并加入线程池中
for i in range(0,thread_num):
    thread = threading.Thread(target=run_suite,args=(all_list[i],))
    threads_list.append(thread)
 
# 循环运行整个线程池
print("测试开始时间：{}".format(time.strftime("%Y%m%d%H%M%S")))
for t in threads_list:
    t.start()
for t in threads_list:
    t.join()
print("测试结束时间：{}".format(time.strftime("%Y%m%d%H%M%S")))