递归检查数组中的新元素并触发动作答案

【问题标题】：Recursively check array for new elements and trigger action递归检查数组中的新元素并触发动作
【发布时间】：2019-01-14 19:04:44
【问题描述】：

我正在运行一个脚本，它每 60 秒不断生成文件，然后我将其存储在一个数组中：

my_files = [file1, file2, ..., filen]

随着这个数组不断动态增长，我需要递归地检查这个数组是否有新文件。每次有新文件时，我都需要将它们添加到“队列”中，并通过调用其他函数等待它们被处理。

最好的方法是什么？我已经研究过看门狗，但对于我需要的东西来说似乎很复杂。我也尝试过使用以下类，这将允许我递归检查我的文件夹，但我无法遍历“获取”的文件，因为它不是list：

class RepeatedTimer(object):

    def __init__(self, interval, function, *args, **kwargs):
        self._timer = None
        self.interval = interval
        self.function = function
        self.args = args
        self.kwargs = kwargs
        self.is_running = False
        self.next_call = time.time()
        self.start()

     def _run(self):
        self.is_running = False
        self.start()
        self.function(*self.args, **self.kwargs)

     def start(self):
         if not self.is_running:
             self.next_call += self.interval
             self._timer = threading.Timer(self.next_call - time.time(), self._run)
             self._timer.start()
             self.is_running = True

     def stop(self):
         self._timer.cancel()
         self.is_running = False

以下内容不起作用，因此我无法访问文件：

my_files = [RepeatedTimer(30, pick_testing_files)]

while True:
    if len(my_files) > 0:
        files_to_be_processed = my_files.pop()
        threading.Thread(target=test, args=(files_to_be_processed)).start()

【问题讨论】：

my_files 在 Python 中被称为 list，而不是数组，顺便说一下。 file1, file2, ..., filen 的具体类型是什么？为什么检查需要递归？
@martineau 它们是 pcap 文件，每 60 秒生成一次。它不一定是递归的，但我需要每 x 分钟检查一次文件夹以获取新创建的文件。
您的意思是它们是文件名称吗？ Python 有一个名为该名称的内置类型，这就是我问的原因。
是的，比如：my_files = [file1.pcap, file2.pcap, etc.]

标签： python arrays multithreading file

【解决方案1】：

这样的事情似乎有效（至少就我自己的有限测试而言）。我添加了一个threading.Lock 来保护共享资源my_files 不被并发访问。

import random
import string
import threading
import time


class RepeatedTimer:
    def __init__(self, interval, function, *args, **kwargs):
        self._timer = None
        self.interval = interval
        self.function = function
        self.args = args
        self.kwargs = kwargs
        self.is_running = False
        self.next_call = time.time()
        self.start()

    def _run(self):
        self.is_running = False
        self.start()
        self.function(*self.args, **self.kwargs)

    def start(self):
        if not self.is_running:
            self.next_call += self.interval
            self._timer = threading.Timer(self.next_call - time.time(), self._run)
            self._timer.start()
            self.is_running = True

    def stop(self):
        self._timer.cancel()
        self.is_running = False


def pick_testing_files(filenames, lock):
    newfiles = []
    with lock:
        while True:
            try:
                filename = filenames.pop()
            except IndexError:  # List is empty.
                break
            newfiles.append(filename)

    print('files retrieved:', newfiles)

def random_filename():
    letters = []
    for _ in range(random.randint(4, 8)):
        letters.append(random.choice(string.ascii_lowercase))
    return ''.join(letters)


my_files = []
filelist_lock = threading.Lock()

watcher = RepeatedTimer(5, pick_testing_files, my_files, filelist_lock)
#watcher.start()  # Not needed because RepeatedTimer's start themselves.

for _ in range(30):  # Run test for 30 seconds.
    # Add some filenames to my_files list.
    with filelist_lock:
        for _ in range(random.randint(1, 4)):  # Generate some filenames.
            my_files.append(random_filename())
    print('new files added')
    time.sleep(1)  # Wait a little before adding more filenames.

watcher.cancel()

【讨论】：