Python 多进程处理的方法。

Question

Python 多进程处理的方法。

pythonmultiprocessingprogress

5

我一直忙于编写我的第一份多进程代码，而它现在已经可行了，太好了。然而，现在我想要一些进度反馈，但我不确定最好的方法是什么。

我的代码（见下文）简述：

扫描目标目录中的mp4文件
每个文件由一个单独的进程分析，进程保存结果（一张图片）

我所寻找的内容可能包括：

简单

每当一个进程完成一个文件时，它会发送“完成”消息
主要代码会计算已完成的文件数

高级一些

Core 0  processing file 20 of 317 ||||||____ 60% completed
Core 1  processing file 21 of 317 |||||||||_ 90% completed
...
Core 7  processing file 18 of 317 ||________ 20% completed

我阅读了关于队列、池、tqdm等各种信息，但不确定应该选择哪种方法。请问有人能指出在这种情况下适用的方法吗？

提前感谢！

编辑：根据gsb22的建议更改了启动进程的代码。

我的代码：

# file operations
import os
import glob
# Multiprocessing
from multiprocessing import Process
# Motion detection
import cv2


# >>> Enter directory to scan as target directory
targetDirectory = "E:\Projects\Programming\Python\OpenCV\\videofiles"

def get_videofiles(target_directory):

    # Find all video files in directory and subdirectories and put them in a list
    videofiles = glob.glob(target_directory + '/**/*.mp4', recursive=True)
    # Return the list
    return videofiles


def process_file(videofile):

    '''
    What happens inside this function:
    - The video is processed and analysed using openCV
    - The result (an image) is saved to the results folder
    - Once this function receives the videofile it completes
      without the need to return anything to the main program
    '''

    # The processing code is more complex than this code below, this is just a test
    cap = cv2.VideoCapture(videofile)

    for i in range(10):
        succes, frame = cap.read()

        # cv2.imwrite('{}/_Results/{}_result{}.jpg'.format(targetDirectory, os.path.basename(videofile), i), frame)

        if succes:
            try:
                cv2.imwrite('{}/_Results/{}_result_{}.jpg'.format(targetDirectory, os.path.basename(videofile), i), frame)
            except:
                print('something went wrong')


if __name__ == "__main__":

    # Create directory to save results if it doesn't exist
    if not os.path.exists(targetDirectory + '/_Results'):
        os.makedirs(targetDirectory + '/_Results')

    # Get a list of all video files in the target directory
    all_files = get_videofiles(targetDirectory)

    print(f'{len(all_files)} video files found')

    # Create list of jobs (processes)
    jobs = []

    # Create and start processes
    for file in all_files:
        proc = Process(target=process_file, args=(file,))
        jobs.append(proc)

    for job in jobs:
        job.start()

    for job in jobs:
        job.join()

    # TODO: Print some form of progress feedback

    print('Finished :)')

- MrExplore

1

我可以给你提供一个解决方案，但是我认为 proc.start() 和 proc.join() 是错误的。你正在启动进程，然后立即加入它，这实际上会阻塞你的 for loop，所以我认为你现在并没有进行多进程处理。你能否在 process_file 方法中放置一个 10 秒的延迟，并查看最终结果是否仅延迟了 10 秒或 10 *（文件数）？ - gsb22

是的，你说得对，我测试了一下并在我的帖子中相应地编辑了代码。 - MrExplore

1个回答

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，
原文链接

- 2e0byo · Accepted Answer

我阅读了关于队列、池、tqdm等各种信息，但不确定该选择哪种方法。有人能指出在这种情况下适用的方法吗？

以下是一种非常简单的方法，可以以最小的成本获得进度指示：

from multiprocessing.pool import Pool
from random import randint
from time import sleep

from tqdm import tqdm


def process(fn) -> bool:
    sleep(randint(1, 3))
    return randint(0, 100) < 70


files = [f"file-{i}.mp4" for i in range(20)]

success = []
failed = []
NPROC = 5
pool = Pool(NPROC)


for status, fn in tqdm(zip(pool.imap(process, files), files), total=len(files)):
    if status:
        success.append(fn)
    else:
        failed.append(fn)

print(f"{len(success)} succeeded and {len(failed)} failed")

一些评论：

tqdm是一个第三方库，非常好地实现了进度条。还有其他的库可以使用。 pip install tqdm。
我们使用一个池（对于像这样简单的事情，几乎没有理由自己管理进程）NPROC流程。我们让池处理在输入数据上迭代我们的流程函数。
我们通过返回布尔值（在此示例中我们随机选择，偏向成功）来表示状态。我们不返回文件名，虽然我们可以这样做，因为它必须被序列化并从子进程发送，这是不必要的开销。
我们使用Pool.imap，它返回一个迭代器，其与我们传递的可迭代对象保持相同的顺序。因此，我们可以直接使用files迭代。由于我们使用未知大小的迭代器，tqdm需要告诉它有多长。（我们本可以使用pool.map，但没有必要提交内存 --- 虽然对于一个bool变量，可能没有太大差别。）

我故意把它写成一种食谱。您可以通过使用高级放置范例来使用多进程，并且Pool.[i]map是最有用的之一。

参考文献

https://docs.python.org/3/library/multiprocessing.html#multiprocessing.pool.Pool https://tqdm.github.io/