问如何解决创建过多进程间管道的问题？

EN

我对Python非常陌生，所以请不要评判我:)我是一个网络工程师，从事连接600多个设备来收集特定数据的python程序。它使用的是多重处理。我已经决定基本上一次运行，因为我使用的服务器可以很好地处理这种方法，它极大地加快了整个过程。我刚刚注意到我的程序使用了大量的文件描述符(超过200 K)。我能够跟踪到进程间(我假设)正在创建的管道。看起来，每个连续产生的进程不仅打开一个管道给它的父进程(这是主程序)，而且打开给他之前所有的“兄弟”。老实说，我不知道这是正常行为还是代码出了问题。我确实需要进程将数据发送回他们的父进程，但我绝对不需要它们之间的通信。当我生成子进程时，我能做些什么来防止创建完整的管道网吗？

这表明了这个问题：

root@vf1netcat2:~# lsof -p 15531型grep管

python3 15531 Netcat4r FIFO 0,12 0t0 334887管

python3 15531网猫5w FIFO 0,12 0t0 334887管

python3 15531 Netcat6R FIFO 0,12 0t0 334888管

python3 15531网猫7w FIFO 0,12 0t0 334888管

python3 15531 Netcat9wFIFO 0,120t0 334889管

root@vf1netcat2:~# lsof -p 15532型grep管

python3 15532 Netcat4r FIFO 0,12 0t0 334887管

python3 15532网猫5w FIFO 0,12 0t0 334887管

python3 15532 Netcat6R FIFO 0,12 0t0 334888管

python3 15532网猫7w FIFO 0,12 0t0 334888管

python3 15532 Netcat8R FIFO 0,12 0t0 334889管

python3 15532网猫10w FIFO 0,12 0t0 334890管

root@vf1netcat2:~# lsof -p 15533型grep管

python3 15533 Netcat4r FIFO 0,12 0t0 334887管

python3 15533网猫5w FIFO 0,12 0t0 334887管

python3 15533 Netcat6R FIFO 0,12 0t0 334888管

python3 15533网猫7w FIFO 0,12 0t0 334888管

python3 15533 Netcat8R FIFO 0,12 0t0 334889管

python3 15533 Netcat9r FIFO 0,12 0t0 334890管

python3 15533 Netcat11w FIFO 0,120t0 334891管

..。

root@vf1netcat2:~# lsof -p 15735 \ grep管# wc -l

209

root@vf1netcat2:~# lsof -p 16035 \ grep管# wc -l

509

root@vf1netcat2:~# lsof -p 16100 \ grep管# wc -l

root@vf1netcat2:~# lsof _ wc -l

三八三八

root@vf1netcat2:~# lsof _ wc -l

184686

root@vf1netcat2:~# lsof _ wc -l

237127

root@vf1netcat2:~# lsof _ wc -l

228187

我就是这样生孩子的：

def execute_data_processing_function(data_list: List[Any], data_processing_function: Callable[..., List[Any]], *args: Any, **kwargs: Any) -> List[Any]:
    """ Execute generic data processing function in single or multiprocess manner and return merged list of results """

    if SINGLE_PROCESS_MODE:
        results = [data_processing_function(_, *args, **kwargs) for _ in data_list]

    else:
        with concurrent.futures.ProcessPoolExecutor(max_workers=len(data_list)) as executor:
            process_pool = [executor.submit(data_processing_function, _, *args, **kwargs) for _ in data_list]

        results = [_.result() for _ in process_pool if not _.exception() and _.result()]

    return [_ for __ in results for _ in __]