使用 class 在其方法中包含无限循环的最佳显式和隐式方法是什么？

Question

我正在尝试编写一个 class 代码，其中包含一个无限循环的方法，直到发出如下所示的显式中断请求：

# module_2

import sys
from time import sleep
from pathlib import Path


class Receiver:
    
    
    def __init__(self, *args, **kwargs):
        
        self.identified = False
        self.id_= hex(id(self))
        
        
    @property
    def get_id(self):
        
        self.identified = True
        return self.id_
        
        
    def write_id(self):
        
        self.identified = True
        with open('active_listeners.lst', 'a') as f:
            f.write(str(self.id_))
        return True
    
    
    def _is_interrupted(self):
        
        flag = Path.cwd().joinpath(Path(f'{self.id_}.interrupt'))
        
        if flag.exists():
            flag.unlink()
            return True
        return False
        
        
    def listen(self):
        
        if not self.identified:
            print("can't start listening without identification or else can't stop listening.")
            print("use 'get_id' or 'write_id' to identify me.")
            return False
        
        while True:
            try:
                print('listening...')
                sleep(5)
                result = 'foo'
                print(f'got result: {result}')
                with open(f'results_{self.id_}.tmp', 'w') as f:
                    f.write(result)
                        
                if self._is_interrupted():
                    sys.exit(0)
            
            except KeyboardInterrupt:
                sys.exit(0)
    
                            
if __name__ == '__main__':
            
            r = Receiver()
            r.write_id()
            r.listen()

Receiver 既可以作为独立对象执行，也可以通过将其导入到另一个模块中使用，例如 module_1。

我有两个问题，

在调用 Receiver.listen() 时，通过将 Receiver 导入另一个模块（比如 main）而不暂停主模块的代码执行流程，使用 Receiver 的最佳方法是什么？（听 I/O 绑定，我在 Windows）
使用 Receiver 的最佳方式是什么而不显式导入它 但从主模块启动 module_2 作为一个完全独立的进程就好像在 Windows 上的分隔 shell 中执行 'python -m module_2'。在此使用中，我需要 shell 保持打开状态以监视输出并停止使用 'ctrl+c'
收听

Answer 1

运行它在一个线程中有点像你在运行作为主线程时所做的那样：

from pathlib import Path
from threading import Thread
from module2 import Receiver

r = Receiver()
r.write_id()
t = Thread(target=r.listen)
t.start()

#do some other stuff...

#Path.cwd().joinpath(Path(f'{r.id_}.interrupt'))
#shorter way to write this: 
flag = Path.cwd() / f'{r.id_}.interrupt'
with open(flag, 'w'): #"touch" your stopflag
    pass
t.join() #wait for the thread to exit

在那个实例中使用 keyboardinterrupt 是不好的做法，因为它可以通过调用 sys.exit(0) 毫无警告地终止整个脚本。一般来说，图书馆不应该自己调用 exit。始终让主脚本执行此操作。最好从循环中简单地 break ，然后让函数 return 执行到启动它的任何地方。如果它在自己的进程中，当函数 returns 时它将到达文件末尾并自行退出。

一个简单的方法运行将它作为自己的主脚本和自己的 cmd window 是使用 os.system 调用“开始”命令（这个命令必须改变如果你运行在另一个 OS).

from sys import argv
from time import sleep
import os

if __name__ == "__main__":
    if '-child' not in argv: #don't fork bomb yourself...
        cmd = f"start cmd /k python \"{__file__}\" -child"
        #/k will keep the window open after the script completes
        #/c will close immediately after completion
        os.system(cmd)
    for i in range(10): #just a simple example to demonstrate starting a new window...
        print(i)
        sleep(1)

使用 class 在其方法中包含无限循环的最佳显式和隐式方法是什么？

What is the best explicit and implicit way to use a class contains an infinite loop in its methods?

python

multithreading

shell

multiprocessing

subprocess