使用的多处理限制线程

Question

我有一些包含数据的大型列表，需要处理。我正在使用多处理来更快地完成工作，除非列表包含超过 100 行，否则它工作得很好，因为它开始使用大量 CPU 和内存。

我提供了一些示例数据：

changed_devices = [[3036c360, 013-HX.ITEM-UA-01, DESCRIPTION 6, 172.29.6.2, Location#All Locations#013-LOCATION-1, IPSEC#Is IPSEC Device#No, Device Type#All Device Types#Switch, ISE Config#ISE Config#With ISE Config Monitor Mode, ISE_enable#ISE_enable#YES - Monitor Mode],
[306fd4c0, 013-HX.ITEM-UA-02, DESCRIPTION 7, 172.29.6.3, Location#All Locations#013-LOCATION-1, IPSEC#Is IPSEC Device#No, Device Type#All Device Types#Switch, ISE Config#ISE Config#With ISE Config Monitor Mode, ISE_enable#ISE_enable#YES - Monitor Mode],
[30a8bf10, 013-UX.ITEM-UA-01, DESCRIPTION 8, 172.29.6.10, Location#All Locations#013-LOCATION-1, IPSEC#Is IPSEC Device#No, Device Type#All Device Types#Switch, ISE Config#ISE Config#With ISE Config Monitor Mode, ISE_enable#ISE_enable#YES - Monitor Mode],
[30e293c0, 013-UX1.ITEM-UA-01, DESCRIPTION 1, 172.29.6.4, Location#All Locations#013-LOCATION-1, IPSEC#Is IPSEC Device#No, Device Type#All Device Types#Switch, ISE Config#ISE Config#With ISE Config Monitor Mode, ISE_enable#ISE_enable#YES - Monitor Mode],
[311cb690, 013-UX2.ITEM-UA-01, DESCRIPTION 2, 172.29.6.5, Location#All Locations#013-LOCATION-1, IPSEC#Is IPSEC Device#No, Device Type#All Device Types#Switch, ISE Config#ISE Config#With ISE Config Monitor Mode, ISE_enable#ISE_enable#YES - Monitor Mode],
[3155ef00, 013-UX3.ITEM-UA-01, DESCRIPTION 3, 172.29.6.6, Location#All Locations#013-LOCATION-1, IPSEC#Is IPSEC Device#No, Device Type#All Device Types#Switch, ISE Config#ISE Config#With ISE Config Monitor Mode, ISE_enable#ISE_enable#YES - Monitor Mode],
[318ed950, 013-UX4.ITEM-UA-01, DESCRIPTION 4, 172.29.6.7, Location#All Locations#013-LOCATION-1, IPSEC#Is IPSEC Device#No, Device Type#All Device Types#Switch, ISE Config#ISE Config#With ISE Config Monitor Mode, ISE_enable#ISE_enable#YES - Monitor Mode],
[31c8d510, 013-UX5.ITEM-UA-01, DESCRIPTION 5, 172.29.6.8, Location#All Locations#013-LOCATION-1, IPSEC#Is IPSEC Device#No, Device Type#All Device Types#Switch, ISE Config#ISE Config#With ISE Config Monitor Mode, ISE_enable#ISE_enable#YES - Monitor Mode],
[3202f7e0, 013-UX5.ITEM-UA-02, DESCRIPTION 5, 172.29.6.9, Location#All Locations#013-LOCATION-1, IPSEC#Is IPSEC Device#No, Device Type#All Device Types#Switch, ISE Config#ISE Config#With ISE Config Monitor Mode, ISE_enable#ISE_enable#YES - Monitor Mode]]

我当前的代码：

if __name__ == "__main__":
    for chg in range(len(changed_devices)):
        p = multiprocessing.Process(target=start_exscript, args=(accounts,changed_devices[chg][0],changed_devices[chg][1],changed_devices[chg][2],changed_devices[chg][3],changed_devices[chg][4],changed_devices[chg][5],changed_devices[chg][6],changed_devices[chg][7],changed_devices[chg][8],changed_devices[chg][9],changed_devices[chg][10],))
        processes.append(p)
        p.start()

    for process in processes:
        process.join()

列表可以包含超过 5000 个条目，所以我希望脚本能够解决这个问题，并且一次只能启动 100 个线程。有人有想法吗？

更新：

我仍然无法让它工作，所以我会提供一些额外的细节。

我使用的是 Python 2.7，但无法让 concurrent.futures 工作。

start_exscript 只是另一个函数，它看起来像这样：

def start_exscript(accounts, host, ena, ise_nwdvs_id, ise_nwdvs_name, ise_nwdvs_ip, ise_nwdvs_desc, ise_nwdvs_loc, ise_nwdvs_ipsec, ise_nwdvs_type, ise_nwdvs_cfg, ise_nwdvs_ena):
    try:
        start(accounts, host, bind(update_config_cisco_device, ena), verbose=0) # starts Python Exscript
        updateisedvsdata = [ise_nwdvs_id, ise_nwdvs_name, ise_nwdvs_ip,  ise_nwdvs_desc, ise_nwdvs_loc, ise_nwdvs_ipsec, ise_nwdvs_type, ise_nwdvs_cfg, ise_nwdvs_ena]
        ise_update_nw_dvs = ise_update_network_device(updateisedvsdata) # Start update function
        if ise_update_nw_dvs:
            logger.critical("[ROOT]: Updating ISE config variable on: [ "+ise_nwdvs_name+" ] [ OK ]")
        else:
            logger.critical("[ROOT]: Updating ISE config variable on: [ "+ise_nwdvs_name+" ] [ Failed ]")
    except Exception as e:
        logger.critical(e.message, e.args)

Python Exscript只是一个可以登录网络设备的插件。

我在使用 Pool 传递所有参数时遇到问题。

我试过了，但它不会启动：

cpus = multiprocessing.cpu_count() - 1
pool = Pool(processes=cpus)

for chg in range(len(changed_devices)):
    pool.apply_async(start_exscript,args=(accounts,changed_devices[chg][0],changed_devices[chg][1],changed_devices[chg][2],changed_devices[chg][3],changed_devices[chg][4],changed_devices[chg][5],changed_devices[chg][6],changed_devices[chg][7],changed_devices[chg][8],changed_devices[chg][9],changed_devices[chg][10]))

pool.close()
pool.join()

这没有任何反应。我什至没有收到错误，这让我更加困惑。

更新 2

任何向外看的人的最终工作结果。非常感谢 SyntaxVoid。

def start_exscript(args):
    try:
        # --- PLEASE READ ---
        # Setup Switch/Router CLI access - remember to change
        cliuser = "x"
        clipass = "x"
        accounts = Account(cliuser,clipass,needs_lock=False)

        host = args[0]
        ena = args[1]
        ise_nwdvs_id = args[2]
        ise_nwdvs_name = args[3]
        ise_nwdvs_ip = args[4]
        ise_nwdvs_desc = args[5]
        ise_nwdvs_loc = args[6]
        ise_nwdvs_ipsec = args[7]
        ise_nwdvs_type = args[8]
        ise_nwdvs_cfg = args[9]
        ise_nwdvs_ena = args[10]

        start(accounts, host, bind(update_config_cisco_device, ena), verbose=0) # Start Exscript process
        updateisedvsdata = [ise_nwdvs_id, ise_nwdvs_name, ise_nwdvs_ip,  ise_nwdvs_desc, ise_nwdvs_loc, ise_nwdvs_ipsec, ise_nwdvs_type, ise_nwdvs_cfg, ise_nwdvs_ena] # Update data
        ise_update_nw_dvs = ise_update_network_device(updateisedvsdata) # Start update process
        if ise_update_nw_dvs: # If update went well
            logger.critical("[ROOT]: Updating ISE config variable on: [ "+ise_nwdvs_name+" ] [ OK ]")
        else:
            logger.critical("[ROOT]: Updating ISE config variable on: [ "+ise_nwdvs_name+" ] [ Failed ]")
    except Exception as e:
        logger.critical(e.message, e.args)

if __name__ == "__main__":
    cpus = multiprocessing.cpu_count() - 1
    pool = Pool(processes=cpus)
    results = pool.map(start_exscript, changed_devices)

Answer 1

这是一个使用multiprocessing 模块的示例。

编辑：最初这是手动处理队列，但正如 Roland 指出的那样，Pool 已经为您设计。

在if __name__ == "__main__" 块中，我们使用我们希望允许的最大进程数创建池，并将map 它添加到我们的输入参数列表中。对于此示例，它将计算 0 到 24 的平方。由于map 的工作方式，results 的顺序将与您的输入参数的顺序相匹配，即使它们可能会乱序执行。辅助函数中的随机处理时间有助于证明这一点。

代码：

from multiprocessing import Pool
import os
import time
import random

def worker(x):
    pid = os.getpid()
    print(f"PID: {pid} got {x}. Evaluating now...")
    processing_time = random.uniform(0.2, 5)
    time.sleep(processing_time)    
    output = x**2
    print(f"PID: {pid} done evaluating in {processing_time:.2f}s. {x}**2 = {output}.")
    return output


if __name__ == "__main__":
    max_processes = 5
    with Pool(max_processes) as pool:
        results = pool.map(worker, range(25))
    print(results)

编辑：

为帮助将其应用于您的代码，请查看pool.map 文档。

ma​​p(func, iterable[, chunksize])

map() 内置函数的并行等效项（尽管它仅支持一个可迭代参数）。它会一直阻塞，直到结果准备好。

此方法将可迭代对象分割成多个块，将它们作为单独的任务提交给进程池。这些块的（近似）大小可以通过将 chunksize 设置为正整数来指定。

您传递的可迭代对象的每个元素都应该是每个函数调用所需的参数集。不幸的是，我认为您不能为 start_exscript 函数使用关键字参数，因为您只能传递一个对象。有几种方法可以解决这个问题，但我认为这是最直接的方法：

1. 像这样重新定义start_exscript 函数：

def start_exscript(args):
    accounts, host, ena, ise_nwdvs_id, ise_nwdvs_name, ise_nwdvs_ip,
    ise_nwdvs_desc, ise_nwdvs_loc, ise_nwdvs_ipsec, ise_nwdvs_type,
    ise_nwdvs_cfg, ise_nwdvs_ena = args
    # The rest of your function can stay the same

如果您这样做，那么您用作传递给 map 的可迭代对象应该看起来像一个列表列表，其中每个内部列表都是传递给 start_exscript 的参数。喜欢：

pass_this_to_map = [ [ARGUMENTS_FOR_FIRST_CALL], [ARGUMENTS_FOR_SECOND_CALL], ... ]

如果我正确理解了您的代码，我认为您只需像上面那样重新定义您的函数后，将您的changed_devices 列表传递给映射：pool.map(start_exscript, changed_devices)。

如果您无权访问start_exscript 的源代码（如，您无法更改它），那么您必须编写这样的包装器：

def start_exscript_wrapper(f):
    def inner(args):
        return f(*args)
    return inner

然后做

start_exscript = start_exscript_wrapper(start_exscript)

在你调用它之前。

Answer 2

在这种情况下，你应该做的是使用有限数量的进程，让它们处理列表中的所有项目。

幸运的是，Python 已经涵盖了这种情况。

基本上有三种解决方案：

• multiprocessing.Pool
• concurrent.futures.ThreadPoolExecutor
• concurrent.futures.ProcessPoolExecutor

您当前工作函数的名称start_exscript 暗示您正在使用例如subprocess 启动外部程序。

如果正确，我建议使用concurrent.futures.ThreadPoolExecutor。这是我的一个程序作为一个完整的工作示例。它使用ffmpeg 将视频批量转换为 MKV 格式：

from functools import partial
import argparse
import concurrent.futures as cf
import logging
import os
import subprocess as sp
import sys

__version__ = '1.4.1'


def main(argv):
    """
    Entry point for vid2mkv.

    Arguments:
        argv: Command line arguments.
    """
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        '-q', '--videoquality', type=int, default=6, help='video quality (0-10, default 6)'
    )
    parser.add_argument(
        '-a', '--audioquality', type=int, default=3, help='audio quality (0-10, default 3)'
    )
    parser.add_argument(
        '--log',
        default='warning',
        choices=['debug', 'info', 'warning', 'error'],
        help="logging level (defaults to 'warning')"
    )
    parser.add_argument('-v', '--version', action='version', version=__version__)
    parser.add_argument("files", metavar='file', nargs='+', help="one or more files to process")
    args = parser.parse_args(argv)
    logging.basicConfig(
        level=getattr(logging, args.log.upper(), None), format='%(levelname)s: %(message)s'
    )
    logging.debug(f'command line arguments = {args}')
    logging.debug(f'parsed arguments = {args}')
    # Check for required programs.
    try:
        sp.run(['ffmpeg'], stdout=sp.DEVNULL, stderr=sp.DEVNULL)
        logging.debug('found “ffmpeg”')
    except FileNotFoundError:
        logging.error('the “ffmpeg” program cannot be found')
        sys.exit(1)
    # Work starts here.
    starter = partial(runencoder, vq=args.videoquality, aq=args.audioquality)
    with cf.ThreadPoolExecutor(max_workers=os.cpu_count()) as tp:
        fl = [tp.submit(starter, t) for t in args.files]
        for fut in cf.as_completed(fl):
            fn, rv = fut.result()
            if rv == 0:
                logging.info(f'finished "{fn}"')
            elif rv < 0:
                logging.warning(f'file "{fn}" has unknown extension, ignoring it.')
            else:
                logging.error(f'conversion of "{fn}" failed, return code {rv}')


def runencoder(fname, vq, aq):
    """
    Convert a video file to Theora/Vorbis streams in a Matroska container.

    Arguments:
        fname: Name of the file to convert.
        vq : Video quality. See ffmpeg docs.
        aq: Audio quality. See ffmpeg docs.

    Returns:
        (fname, return value)
    """
    basename, ext = os.path.splitext(fname)
    known = [
        '.mp4', '.avi', '.wmv', '.flv', '.mpg', '.mpeg', '.mov', '.ogv', '.mkv', '.webm', '.gif'
    ]
    if ext.lower() not in known:
        return (fname, -1)
    ofn = basename + '.mkv'
    args = [
        'ffmpeg', '-i', fname, '-c:v', 'libtheora', '-q:v',
        str(vq), '-c:a', 'libvorbis', '-q:a',
        str(aq), '-sn', '-y', ofn
    ]
    logging.debug(' '.join(args))
    logging.info(f'starting conversion of "{fname}".')
    cp = sp.run(args, stdout=sp.DEVNULL, stderr=sp.DEVNULL)
    return fname, cp.returncode


if __name__ == '__main__':
    main(sys.argv[1:])

注意：

• 我使用functools.partial 预先提供在调用runencoder 之间不会改变的参数。
• 由于某些转换可能需要比其他转换更长的时间，我在Futures 的列表中使用as_completed。这将在结果可用后立即产生。
• 我将返回文件名和结果，因为结果出现的顺序可能不是它们提交的顺序。
• 当您运行multiprocessing.Pool 或concurrent.futures.ProcessPoolExecutor 或手动运行多个子进程时，通常没有意义 启动比您的CPU 内核更多的进程。正如您所发现的，您会遇到很多进程争夺 CPU 资源。

如果所有编码器的运行时间大致相同，我可以使用Executor 的map 方法，该方法稍微简单一些。这将按照提交的顺序返回结果。

如果你没有在工作函数中使用subprocess，而是在Python中做实际工作，只需将ThreadPoolExecutor更改为ProcessPoolExecutor，然后你会没事的。

编辑：由于您使用的是 2.7，因此 concurrent.futures 的 backport 可用于 2.7。 但在您的用例中，multiprocessing.Pool 应该可以正常工作。