在 python 中将队列传递给看门狗线程

Question

我有一个从主线程实例化的class Indexer，这个class的实例存储在一个变量中，比如说indexer。 watchdog.observers.Observer() 监视目录的变化，这些变化发生在另一个线程中。我尝试将这个 indexer 变量从主线程通过我的处理程序 Vigilante 传递给 ob.schedule(Vigilante(indexer)) 以及来自主线程的其他一些变量。由于在不同的线程中，我无法访问 Vigilante class 中的 indexer 变量。我知道我可以使用 Queue 但我不知道如何将 Queue 传递给看门狗的线程。

这是来自主线程的代码：

if watch:
    import watchdog.observers
    from .utils import vigilante
    class callbacks:
        def __init__(self):
            pass
        @staticmethod
        def build(filename, response):
            return _build(filename, response)
        @staticmethod
        def renderer(src, mode):
            return render(src, mode)
    handler = vigilante.Vigilante(_filter, ignore, Indexer, callbacks, Mode)
    path_to_watch = os.path.normpath(os.path.join(workspace, '..'))
    ob = watchdog.observers.Observer()
    ob.schedule(handler, path=path_to_watch, recursive=True)
    ob.start()
    import time
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        ob.stop()
        Indexer.close()
    ob.join()

Indexer class 用于从实例化 Indexer 的代码的另一部分写入数据库。

这是看门狗线程中 Vigilante class 运行的代码：

class Vigilante(PatternMatchingEventHandler):
    """Helps to watch files, directories for changes"""
    def __init__(self, pattern, ignore, indexer, callback, mode):
        pattern.append("*.yml")
        self.Callback = callback
        self.Mode = mode
        self.Indexer = indexer
        super(Vigilante, self).__init__(patterns=pattern, ignore_directories=ignore)

    def vigil(self, event):
        print(event.src_path, 'modified')
        IndexReader = self.Indexer.get_index_on(event.src_path)
        dep = IndexReader.read_index()
        print(dep.next(), 'dependency')
        feedout = self.Callback.build(
            os.path.basename(event.src_path)
            ,self.Callback.renderer(event.src_path, self.Mode.FILE_MODE)
        )

    def on_modified(self, event):
        self.vigil(event)

    def on_created(self, event):
        self.vigil(event)

我只需要一种方法，通过 Vigilante class

将这些变量从主线程传递到看门狗线程

Answer 1

您可以尝试 Observer pattern（没有双关语意），即让 Observer class 有一个侦听器列表，它将通知它看到的任何更改。然后让索引器向观察者宣布它的兴趣。

在我的示例中，观察者期望订阅者是接收更改的可调用对象。然后你可以这样做：

from queue import Queue
class Observable:
    def __init__(self):
        self.listeners = []
    def subscribe(listener):
        self.listeners.append(listener)
    def onNotify(change):
        for listener in self.listeners:
            listener(change)

class Indexer(Thread):
    def __init__(self, observer):
        Thread.__init__(self)
        self.q = Queue()
        observer.subscribe(self.q.put)
    def run(self):
        while True:
            change = self.q.get()

因为标准 Queue 是完全线程安全的，所以这可以正常工作。

Answer 2

我终于找到了一种方法，无需像以前那样过多地交叉线程，其想法源自@EvertW 的回答。我将主线程中的 Queue 传递给另一个线程中的 Vigilante class，因此每个修改后的文件都将放入 Queue 中，然后从主线程中，我从队列中修改了文件，从 Indexer 数据库中读取，并且 Vigilante.vigil 方法需要执行的所有其他任务都移到了主线程，因为这些任务取决于修改后的文件和读取的内容索引器数据库。

这个错误消失了：

SQLite objects created in a thread can only be used in that same thread.The object was created in thread id 9788 and this is thread id 4288.

这是我所做的一个片段：

....
q = Queue.LifoQueue(10)
handler = vigilante.Vigilante(q, _filter, ignore)
path_to_watch = os.path.normpath(os.path.join(workspace, '..'))
ob = watchdog.observers.Observer()
ob.schedule(handler, path=path_to_watch, recursive=True)
ob.start()
import time
try:
    while True:
        if not q.empty():
            modified = q.get()
            IndexReader = Indexer.get_index_on(modified)
            deps = IndexReader.read_index()
            print(deps.next(), 'dependency')
            # TODO
        else:
            print('q is empty')
            time.sleep(1)
except KeyboardInterrupt:
    ob.stop()
    Indexer.close()
ob.join()

义警class:

class Vigilante(PatternMatchingEventHandler):
    """Helps to watch files, directories for changes"""
    def __init__(self, q, pattern, ignore):
        self.q = q
        super(Vigilante, self).__init__(
            patterns=pattern, 
            ignore_patterns=ignore, 
            ignore_directories=True
        )

    def vigil(self, event):
        print(event.src_path, 'modified')
        self.q.put(event.src_path)

    def on_modified(self, event):
        self.vigil(event)

    def on_created(self, event):
        self.vigil(event)
....

PS：忠告：我对在看门狗中遇到这种线程问题的人的忠告是； "Don't trust the watchdog's thread to do tasks on modified files, just get the modified files out and do whatever you like with them, except the task is a simple one."

在 python 中将队列传递给看门狗线程

Pass a Queue to watchdog thread in python

python

multithreading

queue

watchdog