在 Pycharm 上运行同一程序的并行实例

Question

我有一个名为 Scanner.py 的程序，它从一个名为 Config.json 的配置文件加载参数——有没有一种方法可以让我在 Pycharm 上并行运行同一文件的两个实例，并通过 Config.json 传递不同的配置集。

例子：这是读入Pycharm的Python文件的Json文件。我希望它的各种组合不一个接一个地运行，而是并行运行

"Traded Instrument": "STOCK",
"Start Date": "2019/01/01",
"End Date": "2022/12/20",
"Trade Start": "09:16",
"Trade End": "15:29",
"Trade End Scalp": "09:16",

除了使用不同的机器之外，我不知道如何开始使用它，但我需要通过运行 Scanner.py 同时需要超过 3/4 个实例

Answer 1

我建议使用 multiprocessing。您可以创建一个接受配置指令的函数，并使用不同的配置运行它。

from multiprocessing import Pool

def runner(config: dict):
    # your logic
    ...

if __name__ == '__main__':
    configs = [{...}, {...}, {...}]
    with Pool(len(configs)) as p:
        p.map(runner, configs)

Answer 2

我会说在你的情况下最好制作一个多线程包装器，仅仅是因为线程使用相同的内存空间，不需要等待等，而多处理使用不同的内存空间。

from threading import Thread

def async(func: object):
    """ This is a decorator for async functions """
    def asyncFunction(*args, **kwargs):
        """ This is the wrapper function that will be returned """
        thread: Thread = Thread(target=func, args=args, kwargs=kwargs)
        thread.start()
        return thread
    return asyncFunction

@async
def runner(config: dict):
    ...
    
if __name__ == "__main__":
    configs = [{...}, {...}, {...}]
    for i in range(len(configs)):
        runner(configs[i])

注意：使用线程有一个缺点，它不是内存安全的（主要是），并且您不能在程序进行处理时中断程序，除非您在函数内部进行某种检查。

如果您希望它与键盘中断一起工作，程序将如下所示：

from threading import Thread

KeyboardInterruptDetected = False # This is a global variable that will be used to stop the program

def async(func: object):
    """ This is a decorator for async functions """
    def asyncFunction(*args, **kwargs):
        """ This is the wrapper function that will be returned """
        thread: Thread = Thread(target=func, args=args, kwargs=kwargs)
        thread.start()
        return thread
    return asyncFunction

@async
def runner(config: dict):
    if KeyboardInterruptDetected:
        return
    ... # Do something.

if __name__ == "__main__":
    configs = [{...}, {...}, {...}] # This is a list of the config dictionaries
    try:
        for _ in range(len(configs)):
            runner(configs[_])
    except KeyboardInterrupt:
        KeyboardInterruptDetected = True