python多线程执行测试用例

1 下列有一个简单的任务,每个任务都会睡眠10秒,如果是用单线程运行,至少要运行100秒,请使用多线程启动10个线程运行,实现10秒左右运行完所有任务

任务代码:

import unittest,threading
def test_task(name):
    print(f"{threading.current_thread().name}:",name)
    time.sleep(10)

class TestTask(unittest.TestCase):
    def test01(self):
        test_task(1)
    def test02(self):
        test_task(2)
    def test03(self):
        test_task(3)
    def test04(self):
        test_task(4)
    def test05(self):
        test_task(5)
    def test06(self):
        test_task(6)
    def test07(self):
        test_task(7)
    def test08(self):
        test_task(8)
    def test09(self):
        test_task(9)
    def test10(self):
        test_task(10)

import unittest,threading

import time


def test_task(name):
    print("thread:{} start".format(name))
    time.sleep(10)
    print("thread:{} finish".format(name))

class TestTask(unittest.TestCase):
    def __init__(self):
        self.fun=[self.test01,self.test02,self.test03,self.test04,self.test05,
                  self.test06,self.test07,self.test08,self.test09,self.test10]

    def test01(self):
        test_task(1)

    def test02(self):
        test_task(2)

    def test03(self):
        test_task(3)

    def test04(self):
        test_task(4)

    def test05(self):
        test_task(5)

    def test06(self):
        test_task(6)

    def test07(self):
        test_task(7)

    def test08(self):
        test_task(8)

    def test09(self):
        test_task(9)

    def test10(self):
        test_task(10)

if __name__ == "__main__":
    testcases = TestTask()
    for i in range(10):
        case_thread = threading.Thread(target=testcases.fun[i])
        case_thread.daemon = True
        case_thread.start()
    #wait all finish
    time.sleep(2000)

你题目的解答代码如下:

import unittest,threading
import time
lock = threading.Lock()

def test_task(name):
    lock.acquire() # 输出时线程加锁,防止两个print()同时输出,输出的内容交错混合在一起
    print(f"{threading.current_thread().name}:",name)
    lock.release()
    time.sleep(10)

class TestTask(unittest.TestCase):
    def test01(self):
        test_task(1)
    def test02(self):
        test_task(2)
    def test03(self):
        test_task(3)
    def test04(self):
        test_task(4)
    def test05(self):
        test_task(5)
    def test06(self):
        test_task(6)
    def test07(self):
        test_task(7)
    def test08(self):
        test_task(8)
    def test09(self):
        test_task(9)
    def test10(self):
        test_task(10)


p = TestTask()
threading.Thread(target=p.test01).start()
threading.Thread(target=p.test02).start()
threading.Thread(target=p.test03).start()
threading.Thread(target=p.test04).start()
threading.Thread(target=p.test05).start()
threading.Thread(target=p.test06).start()
threading.Thread(target=p.test07).start()
threading.Thread(target=p.test08).start()
threading.Thread(target=p.test09).start()
threading.Thread(target=p.test10).start()

如有帮助,请点击我的回答下方的【采纳该答案】按钮帮忙采纳下,谢谢!

img

这个问题需要用到协程。因为python上存在 Global Interpreter Lock,即全局解释器锁,简单来说,把 Python 解释器(默认是 CPython 版本)作为共享资源,GIL 就是一个保护解释器资源的锁,它确保在 Python 的虚拟机中同一时刻只有一个线程在执行因为 GIL 这把大锁,每次都只有一个线程在运行。由于GIL 的存在,Python 多线程是串行的(同时只有一个线程在工作),协程的概念:协程,是一种比线程更加轻量级的存在。正如一个进程可以拥有多个线程一样,一个线程也可以拥有多个协程。协程是运行在同一个线程中的一组事件,由人工进行调度(系统可以调度的最小单位是线程,而协程比线程更小),事件切换所花销的时间几乎可以忽略不计,简单说,协程是在单进程单线程中就可以实现并发的大利器。