Python多处理 - 几秒钟后开始下一个进程

Question

我是 Python 的初学者，如果您能帮助我进行清晰易懂的解释，我将不胜感激。

在我的 Python 脚本中，我有一个让多个线程执行 I/O 绑定任务的函数（它的真正作用是使用 Azure Python SDK 同时发出多个 Azure 请求），并且我还有一个时差列表像 [1 seconds, 3 seconds, 10 seconds, 5 seconds, ..., 7 seconds] 这样我在每次时间差后再次执行该函数。

假设我要执行该函数并在 5 秒后再次执行它。第一次执行可能需要超过 5 秒才能完成，因为它必须等待它发出的请求完成。所以，我想在不同的进程中执行每个函数，以便函数的不同执行不会相互阻塞（即使它们不使用不同的进程而不会相互阻塞，我只是不希望不同执行中的线程混合）。

我的代码是这样的：

import multiprocessing as mp
from time import sleep

def function(num_threads):
    """
    This functions makes num_threads number of threads to make num_threads number of requests 
    """

# Time to wait in seconds between each execution of the function
times = [1, 10, 7, 3, 13, 19]

# List of number of requests to make for each execution of the function
num_threads_list = [1, 2, 3, 4, 5, 6]

processes = []

for i in range(len(times)):
    p = mp.Process(target=function, args=[num_threads_list[i]])
    p.start()
    processes.append(p)

    sleep(times[i])

for process in processes:
    process.join()

我要问的问题：

1. 在我的真实脚本中，“times”列表的长度非常大（即 1000）。考虑到“时间”列表中的时间差异，我猜该函数最多有 5 次执行同时使用进程运行。我想知道每个进程在执行完函数后是否终止，因此实际上最多有5个进程在运行。或者，它是否仍然存在，所以会有 1000 个进程，考虑到我电脑的 CPU 内核数量，这听起来很奇怪？

2. 如果您认为有更好的方法来执行我上面解释的操作，请告诉我。

谢谢！

Answer 1

我从你的问题中提取的主要问题是有大量进程同时运行。

您可以通过维护具有最大长度的进程列表来防止这种情况。像这样。

import multiprocessing as mp
from time import sleep
from random import randint


def function(num_threads):
    """
    This functions makes num_threads number of threads to make num_threads number of requests
    """
    sleep(randint(3, 7))


# Time to wait in seconds between each execution of the function
times = [1, 10, 7, 3, 13, 19]

# List of number of requests to make for each execution of the function
num_threads_list = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]

process_data_list = []
max_processes = 4


# =======================================================================================
def main():
    times_index = 0
    while times_index < len(times):

        # cleanup stopped processes -------------------------------
        cleanup_done = False
        while not cleanup_done:
            cleanup_done = True
            # search stopped processes
            for i, process_data in enumerate(process_data_list):
                if not process_data[1].is_alive():
                    print(f'process {process_data[0]} finished')
                    # remove from processes
                    p = process_data_list.pop(i)
                    del p
                    # start new search
                    cleanup_done = False
                    break

        # try start new process ---------------------------------
        if len(process_data_list) < max_processes:
            process = mp.Process(target=function, args=[num_threads_list[times_index]])
            process.start()
            process_data_list.append([times_index, process])
            print(f'process {times_index} started')
            times_index += 1
        else:
            sleep(0.1)

    # wait for all processes to finish --------------------------------
    while process_data_list:
        for i, process_data in enumerate(process_data_list):
            if not process_data[1].is_alive():
                print(f'process {process_data[0]} finished')
                # remove from processes
                p = process_data_list.pop(i)
                del p
                # start new search
                break

    print('ALL DONE !!!!!!')


# =======================================================================================
if __name__ == '__main__':
    main()

正如您在结果中看到的那样，它一次运行 max_processes。

process 0 started
process 1 started
process 2 started
process 3 started
process 3 finished
process 4 started
process 1 finished
process 5 started
process 0 finished
process 2 finished
process 5 finished
process 4 finished
ALL DONE !!!!!!

Answer 2

您还可以使用计时器来完成以下代码中的工作。 我自愿将 15 秒用于线程 2，以便人们可以看到它在时间过去后有效地结束在最后一个位置。

此代码示例有两个主要功能。 第一个 your_process_here() 就像它的名字所说的那样正在等待你自己的代码 第二个是一个管理器，它组织线程切片以避免系统过载。

参数

max_process：脚本正在执行的进程总数

simultp：最大同时进程数

timegl：时间准则，它定义了自父级启动以来每个线程的等待时间。所以等待时间至少是指南中定义的时间（指父母的开始时间）。

换句话说，由于它的指导时间已经过去，考虑到允许的最大并发线程数，线程会尽快启动。

在这个例子中

最大进程 = 6

simultp = 3

timegl = [1, 15, 1, 0.22, 6, 0.5]（只是为了解释，因为更合乎逻辑的是那里有一个增加系列）

shell 中的结果

同时启动的进程：3

进程 n°2 处于活动状态，将在治疗功能开始前再等待 14.99 秒

进程 n°1 处于活动状态，将在治疗功能开始前再等待 0.98 秒

进程 n°3 处于活动状态，将在治疗功能开始前再等待 0.98 秒

---- 进程 n°1 结束 ----

---- 进程 n°3 结束 ----

同时启动的进程：3

进程 n°5 处于活动状态，将在治疗功能开始前再等待 2.88 秒

进程 n°4 处于活动状态，现在将开始

---- 进程 n°4 结束 ----

---- 进程 n°5 结束 ----

同时启动的进程：2

进程 n°6 处于活动状态，现在将开始

---- 进程 n°6 结束 ----

---- 进程 n°2 结束 ----

代码

import multiprocessing as mp
from threading import Timer
import time


def your_process_here(starttime, pnum, timegl):
    # Delay since the parent thread starts
    delay_since_pstart = time.time() - starttime
    # Time to sleep in order to follow the most possible the time guideline 
    diff = timegl[pnum-1]- delay_since_pstart
    if diff > 0: # if time ellapsed since Parent starts < guideline time
        print('process n°{0} is active and will wait {1} seconds more before treatment function starts'\
              .format(pnum, round(diff, 2)))        
        time.sleep(diff) # wait for X more seconds
    else:
        print('process n°{0} is active and will start now'.format(pnum))
    ########################################################
    ## PUT THE CODE AFTER SLEEP() TO START CODE WITH A DELAY
    ## if pnum == 1:
    ##     function1()  
    ## elif pnum == 2:
    ##     function2()
    ## ...
    print('---- process n°{0} ended ----'.format(pnum))

def process_manager(max_process, simultp, timegl, starttime=0, pnum=1, launchp=[]):
    # While your number of simultaneous current processes is less than simultp and
    # the historical number of processes is less than max_process
    while len(mp.active_children()) < simultp and len(launchp) < max_process:
        # Incrementation of the process number
        pnum = len(launchp) + 1
        # Start a new process
        mp.Process(target=your_process_here, args=(starttime, pnum, timegl)).start()
        # Historical of all launched unique processes
        launchp = list(set(launchp + mp.active_children()))        
    # ...
    ####### THESE 2 FOLLOWING LINES ARE TO DELETE IN OPERATIONAL CODE ############
    print('simultaneously launched processes : ', len(mp.active_children()))        
    time.sleep(3) # optionnal : This a break of 3 seconds before the next slice of process to be treated
    ##############################################################################
    if pnum < max_process:
        delay_repeat = 0.1 # 100 ms
        # If all the processes have not been launched renew the operation
        Timer(delay_repeat, process_manager, (max_process, simultp, timegl, starttime, pnum, launchp)).start()        

if __name__ == '__main__':
    max_process = 6 # maximum of processes
    simultp = 3 # maximum of simultaneous processes to save resources
    timegl = [1, 15, 1, 0.22, 6, 0.5] # Time guideline
    starttime = time.time()
    process_manager(max_process, simultp, timegl, starttime)