python-3.x - 如果文件夹被修改，如何使用看门狗并在主线程中执行某些操作？

Question

我正在寻找一个看门狗，我发现了 this很棒的图书馆。我需要安装DBSCAN模型(如果在文件夹中创建文件)。 Joblib 用于 scikit-learn 的 DBSCAN 实现，如果 DBSCAN 运行代码不在主线程中，则 joblib 不允许使用多处理。如果我使用看门狗，DBSCAN 代码将无法在主线程中运行。我该如何解决这个问题？您可以在下面找到看门狗脚本和一个简单的函数来测试它。当我运行main_watchdog.py时并在看门狗正在监视的文件夹中添加一个文件，它会运行 simple_function.py在线程1中。同时，main_watchdog.py在主线程中运行。

PS:解决方案可以是每次调用 simple_function.py 时启动一个子进程。但我担心如果在 watchdog 文件夹中创建多个文件，这可能会导致一些问题。想象一下一次接收 10 个、100 个或 10000 个文件......

#main_watchdog.py
import time
import logging
import threading
from watchdog.observers import Observer
from watchdog.events import LoggingEventHandler
from a_function import simple_function

class Event(LoggingEventHandler):
    def on_created(self, event):
        simple_function(x)

    def on_modified(self, event):
        simple_function(x)

if __name__ == "__main__":
    x = 1
    logging.basicConfig(level=logging.INFO,
                        format='%(asctime)s - %(message)s',
                        datefmt='%Y-%m-%d %H:%M:%S')
    # path = sys.argv[1] if len(sys.argv) > 1 else '.'
    path = '/path/to/watch/the/folder'
    event_handler = Event()
    observer = Observer()
    observer.schedule(event_handler, path, recursive=False)
    observer.start()
    try:
        while True:
            time.sleep(1)
            print(threading.current_thread().name)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

#a_function.py
import threading
def simple_function(x):
    x += 1
    print(threading.current_thread().name)
    print(x)

Answer 1

如果我要正确理解这个问题，您需要业务逻辑在主线程中运行，观察者在后台线程中运行。 通过使用线程库在后台调用观察者线程，然后通过队列(线程之间通信的一种方式)将这些事件的值传递给函数调用，可以轻松解决此问题。

#main_watchdog.py
import time
import logging
import threading
from watchdog.observers import Observer
from watchdog.events import LoggingEventHandler
from a_function import simple_function
import sys
from queue import Queue

q = Queue()
x = 1
class Event(LoggingEventHandler):
    def on_created(self, event):
        q.put(x)
    def on_modified(self, event):
        q.put(x)

def run_observer():
    logging.basicConfig(level=logging.INFO,
                        format='%(asctime)s - %(message)s',
                        datefmt='%Y-%m-%d %H:%M:%S')
    path = sys.argv[1] if len(sys.argv) > 1 else '.'
    # path = '/path/to/watch/the/folder'
    event_handler = Event()
    observer = Observer()
    observer.schedule(event_handler, path, recursive=False)
    observer.start()
    while True:
        time.sleep(1)
        print(threading.currentThread().name)

if __name__ == "__main__":
    background_thread = threading.Thread(target=run_observer, args=())
    background_thread.daemon = True
    background_thread.start()
    print('Business logic')
    while True:
        val = q.get(True)
        simple_function(val)

其他功能保持不变。