使用多处理并行运行测试

Question

我想在 2 台或更多设备上并行执行测试。

我有一个包含测试的列表，我想将它们分发到 test_proc 函数中的所有设备上。例如，test1.py到第一个设备，test2.py到第二个设备等等。它们是并行执行的，如果test1.py执行得更快，那么test3.py转到带有测试的列表下的第一个设备等。我创建了一个包含测试的队列，现在我有了：如果test1.py运行速度快于 test2.py ，则test3.py不是在第一台设备上运行，而是在等待test2.py完成。

Answer 1

你应该实现多处理。按照文档中的说明进行队列，这将解决此问题。它在进程之间创建一个共享队列，然后每个进程在完成之前的任务时都会从中提取。

from multiprocessing import Process, Queue
def worker(input):
    for func, args in iter(input.get, 'STOP'):
        result = func(*args)
        # do something
def test1():
    pass
def test2():
    pass
def test3():
    pass
functions = [test1, test2, test3]
tests = Queue()
def test_proc(functions, tests):
    all_process = []
    for x in functions:   # functions would be list of tests you want to run
        tests.put(x)
    for i in number_of_processes:   # start up a number of processes 
        process = Process(target=worker, args=(tests))
        all_process.append(process)
        process.start()
    for p in all_process:
        tests.put('STOP')   # stop each process via the queue
        p.join()
if name == "__main__":
    test_proc(functions, tests)