Docker中Python程序运行时动态创建Scanner对象的实现问询

核心思路 #

• 用线程安全的容器存储Scanner对象，避免多线程操作时的竞态问题
• 单独启动一个线程负责FTP文件轮询，检测到新输入时创建Scanner并加入容器
• 主遍历逻辑每次从容器获取最新对象列表，兼容动态新增的实例

代码实现 #

1. 基础定义（Scanner类与构建器）

import threading
import time
from ftplib import FTP
import queue

# 你的Scanner类示例
class Scanner:
    def __init__(self, name):
        self.name = name
        self.running = True
    
    def scan(self):
        # 替换为实际扫描逻辑
        print(f"[{time.strftime('%H:%M:%S')}] Scanner {self.name} is running")
        time.sleep(1)

# 你的Scanner构建函数
def scanner_class_builder(name):
    return Scanner(name)

2. 线程安全的Scanner管理器

class ScannerManager:
    def __init__(self):
        # 用Queue保证线程安全，避免直接操作list的竞态问题
        self.scanner_queue = queue.Queue()
        # 初始化默认Scanner（可选）
        self.scanner_queue.put(scanner_class_builder("default_scanner"))
    
    def add_scanner(self, name):
        """添加新的Scanner实例"""
        new_scanner = scanner_class_builder(name)
        self.scanner_queue.put(new_scanner)
    
    def get_current_scanners(self):
        """获取当前所有Scanner实例，同时保留队列内容"""
        scanners = []
        while not self.scanner_queue.empty():
            scanner = self.scanner_queue.get()
            scanners.append(scanner)
            self.scanner_queue.put(scanner)
        return scanners

3. FTP轮询线程

def ftp_poller(manager, ftp_settings, interval=10):
    """定期轮询FTP服务器上的用户输入文件"""
    last_content = ""
    # 建立FTP连接
    ftp = FTP(ftp_settings["host"])
    ftp.login(user=ftp_settings["user"], passwd=ftp_settings["passwd"])
    ftp.cwd(ftp_settings["remote_dir"])

    while True:
        # 获取文件内容
        content_lines = []
        ftp.retrlines(f"RETR {ftp_settings['filename']}", content_lines.append)
        current_content = "\n".join(content_lines).strip()

        # 检测内容变更
        if current_content != last_content and current_content:
            # 按行解析用户输入，支持批量添加
            for name in current_content.split("\n"):
                name = name.strip()
                if name:
                    print(f"New scanner added: {name}")
                    manager.add_scanner(name)
            last_content = current_content
        
        time.sleep(interval)

4. 主执行逻辑

def main():
    # 配置FTP参数，替换为你的实际信息
    ftp_config = {
        "host": "your-web-server-ftp-host",
        "user": "ftp-username",
        "passwd": "ftp-password",
        "remote_dir": "/path/to/input-file-dir",
        "filename": "user_inputs.txt"
    }

    manager = ScannerManager()

    # 启动FTP轮询线程（设为守护线程，随主进程退出）
    poll_thread = threading.Thread(target=ftp_poller, args=(manager, ftp_config))
    poll_thread.daemon = True
    poll_thread.start()

    # 主循环：遍历所有Scanner执行任务
    while True:
        current_scanners = manager.get_current_scanners()
        for scanner in current_scanners:
            if scanner.running:
                scanner.scan()
        time.sleep(2)  # 主循环间隔，按需调整

if __name__ == "__main__":
    main()

关键注意事项 #

• 线程安全：使用queue.Queue而非普通list，因为Queue内置锁机制，避免多线程读写冲突
• FTP文件处理：如果用户输入是单行文本，可直接取current_content作为名称；如果是多行则批量添加，可根据实际需求调整解析逻辑
• Docker网络配置：确保Docker容器能访问Web服务器的FTP服务，若在同一内网使用内网IP，公网环境需开放对应端口并配置容器网络权限
• 资源管理：若需要停止Scanner，可给Scanner类添加stop()方法，主循环中过滤已停止的实例，或从队列中移除过期对象

内容的提问来源于stack exchange，提问作者StrangerDanger