多线程。Pool.apply_async() 立即返回并且不执行回调

Question

函数sync_lister()：

的控制台输出

20203.161 ms
19930.7166 ms
21279.1429 ms
18988.5079 ms
17724.5125 ms
11358.8549 ms
36164.6077 ms
20183.306099999998 ms
14238.174599999998 ms
20383.551 ms
Process finished with exit code 0

很好，这就是我所期望的。现在，我想将其放入工人池中，并尽快完成大约4倍的完成。

我以为我确实做到了。为什么我没有输出？

运行async_lister()时没有控制台输出，并且应用程序立即返回。

在下面完成代码。让我知道您是否需要ffprobe包装器或其他任何内容的来源。

#! /usr/bin/env python3
# -*- coding: utf-8 -*-
import os
from utilities import glob_from
from controller import run_ffprobe
from multiprocessing import Pool
INPUT_DIR = "C:/Users/spike/Music/Deezloader"

def get_duration(metadata):
    return float(metadata.get("format", {}).get("duration", 0)) * 100

def example_callback(metadata):
    print(get_duration(metadata), "ms")

args_list = [(os.path.join(INPUT_DIR, path),) for path in glob_from(INPUT_DIR, "flac")[0:10]]
# Example: [('C:/Users/spike/Music/Deezloader\AWOLNATION - Handyman\1 - AWOLNATION - Handyman.flac',), ...]

def sync_lister():
    for args in args_list:
        example_callback(run_ffprobe(*args))

def async_lister():
    pool = Pool(4)  # Creates four threads, four items from "args" will be run with "run_ffprobe" at a time
    pool.apply_async(run_ffprobe, args=args_list, callback=example_callback)
    # Set the target to "run_ffprobe", and each item in "args_list" will be unpacked as arguments
    # and sent to that function when a pool worker is free
    pool.close()  # Close the pool, no more process can be added
    pool.join()  # Wait for all of the processes to complete

if __name__ == "__main__":
    sync_lister()  # Working
    async_lister()  # Not working

Answer 1

正如apply()/apply_async()文档中所述的那样，每个调用apply()或apply_async()的调用等于一个 call call call to to y y y y y y y y y as the ass the aS第一个参数(即。run_ffprobe(。

在同步代码中，您将args_list的每个元素传递给单独的run_ffprobe()调用：

for args in args_list:
    example_callback(run_ffprobe(*args))

在异步代码中，您要求池进行计算：

example_callback(run_ffprobe(args_list))

这不是您想要的。您需要在循环中调用apply_async()，就像您在循环中称为run_ffprobe()一样。另外，您可以使用map_async()，但这只会调用您的回调一次，其中整个结果列表。这似乎与您在同步代码中所做的事情不匹配，但是您可以重写回调以适合此模式，也可以使用functools.partial(map, example_callback)之类的内容作为map_async()的回调。