spooling技术模拟实现（虚拟打印画廊）

虚拟打印画廊系统设计与实现

一、项目结构说明

虚拟打印画廊/
├── __init__.py               # 包初始化文件
├── main.py                   # 主程序入口
├── queue_display.py          # 任务队列展示模块
├── load_tasks_to_queue.py    # 任务加载与调度模块
├── task_submission.py        # 任务提交模块
├── user_processing_procedure.py # 用户进程模拟模块
└── show_output_images.py     # 图片展示模块

二、核心模块代码解析

1. 主程序入口（main.py）

from 虚拟打印画廊 import queue_display

if __name__ == "__main__":
    queue_display.root.mainloop()  # 启动任务队列展示界面

2. 任务队列展示（queue_display.py）

import tkinter as tk
from load_tasks_to_queue import print_queue, load_tasks_to_queue
from task_submission import input_spool_dir
from show_output_images import create_show_images_interface
from user_processing_procedure import output_spool_dir, user_processing_procedure

root = tk.Toplevel()
root.title("任务队列展示界面")

queue_text = tk.Text(root)
load_button = tk.Button(root, text="加载并展示任务队列", command=lambda: display_queue())

def display_queue():
    queue_text.delete('1.0', tk.END)  # 清空文本框
    print_queue = load_tasks_to_queue(input_spool_dir)  # 从输入井加载任务

    if not print_queue:
        queue_text.insert(tk.END, '任务队列暂时没有内容')
        return

    # 模拟用户进程处理任务
    user = user_processing_procedure(print_queue)
    user.process_tasks()  # 处理后存入输出井

    output_buffer = load_tasks_to_queue(output_spool_dir)  # 加载输出井任务
    for task in output_buffer:
        # 格式化任务信息并展示
        task_info = f"任务ID: {task['task_id']}\n" \
                    f"文件路径: {task['file_path']}\n" \
                    f"标题: {task['title']}\n" \
                    f"作者: {task['author']}\n" \
                    f"描述: {task['description']}\n" \
                    f"时间戳: {task['timestamp']}\n\n"
        queue_text.insert(tk.END, task_info)
    
    create_show_images_interface(output_buffer)  # 打开图片展示界面

if __name__ == "__main__":
    root.mainloop()

3. 任务提交模块（task_submission.py）

import tkinter as tk
from tkinter import filedialog
import os
import json
import time

input_spool_dir = os.path.join(os.getcwd(), 'input_spool')  # 输入井目录

class PrintTask:
    def __init__(self, task_id, file_path, title, author, description):
        self.task_id = task_id
        self.file_path = file_path
        self.title = title
        self.author = author
        self.description = description
        self.status = "等待"

def submit_artwork():
    file_path = filedialog.askopenfilename()
    if not (title_entry.get() and author_entry.get() and description_entry.get()):
        status_label.config(text="请填写完整作品标题、作者和描述信息后再提交！")
        return

    task_id = len(os.listdir(input_spool_dir)) if os.path.exists(input_spool_dir) else 0
    task = {
        "task_id": task_id,
        "file_path": file_path,
        "title": title_entry.get(),
        "author": author_entry.get(),
        "description": description_entry.get(),
        "timestamp": time.time()
    }

    # 保存任务到输入井
    task_file_path = os.path.join(input_spool_dir, f'task_{task_id}.json')
    with open(task_file_path, 'w') as f:
        json.dump(task, f)
    
    status_label.config(text=f"作品已提交，任务ID：{task['task_id']}")
    title_entry.delete(0, tk.END)  # 清空输入框

# 创建提交界面
root = tk.Tk()
root.title("虚拟打印画廊提交界面")
root.geometry("200x200")
root.attributes('-toolwindow', True)  # Windows下去除最大化/最小化按钮

# 输入框与按钮布局
title_label = tk.Label(root, text="作品标题：").pack()
title_entry = tk.Entry(root).pack()
# 作者和描述输入框类似...
submit_button = tk.Button(root, text="提交作品", command=submit_artwork).pack()
status_label = tk.Label(root).pack()

if not os.path.exists(input_spool_dir):
    os.makedirs(input_spool_dir)

if __name__ == "__main__":
    root.mainloop()

4. 任务加载与调度（load_tasks_to_queue.py）

import os
import json
from task_submission import input_spool_dir

output_spool_dir = os.path.join(os.getcwd(), 'output_spool')
print_queue = []  # 全局打印队列

def load_tasks_to_queue(spool):
    global print_queue
    print_queue.clear()
    task_files = os.listdir(spool)
    for task_file in task_files:
        with open(os.path.join(spool, task_file), 'r') as f:
            task_info = json.load(f)
            print_queue.append(task_info)
    return print_queue

def prioritize_queue_fcfs():
    """先来先服务（FCFS）调度算法"""
    return print_queue  # 按文件读取顺序自然有序

5. 用户进程模拟（user_processing_procedure.py）

import os
import json

output_spool_dir = os.path.join(os.getcwd(), 'output_spool')  # 输出井目录

class user_processing_procedure:
    def __init__(self, print_queue):
        self.print_queue = print_queue

    def process_tasks(self):
        if not os.path.exists(output_spool_dir):
            os.makedirs(output_spool_dir)
        self.clear_output_spool()  # 清空输出井

        for task_info in self.print_queue:
            task_info["title"] += "| 假设用户进程处理"  # 修改任务标题模拟处理
            task_id = len(os.listdir(output_spool_dir))
            task_file_path = os.path.join(output_spool_dir, f'task_{task_id}.json')
            with open(task_file_path, 'w') as f:
                json.dump(task_info, f)

    def clear_output_spool(self):
        """清空输出井内容（保留文件夹）"""
        if os.path.exists(output_spool_dir):
            for root, dirs, files in os.walk(output_spool_dir):
                for file in files:
                    os.remove(os.path.join(root, file))
                for dir in dirs:
                    os.rmdir(os.path.join(root, dir))
            print("output_spool文件夹下内容已清空")

6. 图片展示模块（show_output_images.py）

import tkinter as tk
from PIL import Image, ImageTk
import tkinter.messagebox as msgbox

def create_card(frame, task_info):
    card_frame = tk.Frame(frame, bd=2, relief=tk.SOLID)
    task_info_text = f"任务ID: {task_info['task_id']}\n文件路径: {task_info['file_path']}\n..."
    tk.Label(card_frame, text=task_info_text).pack()

    # 处理图片展示
    file_extension = os.path.splitext(task_info['file_path'])[1]
    if file_extension in ['.png', '.jpg', '.gif']:
        try:
            image = Image.open(task_info['file_path'])
            # 缩放图片
            max_width = 150
            scale = max_width / image.size[0]
            image = image.resize((int(image.size[0]*scale), int(image.size[1]*scale)), Image.Resampling.LANCZOS)
            photo = ImageTk.PhotoImage(image)
            tk.Label(card_frame, image=photo).pack()
            card_frame.image = photo  # 防止GC回收
        except Exception as e:
            msgbox.showerror("图片加载错误", f"错误：{e}")
    return card_frame

def create_show_images_interface(output_buffer):
    root = tk.Toplevel()
    root.title("图片展示界面")

    main_frame = tk.Frame(root)
    scrollbar = tk.Scrollbar(main_frame)
    content_canvas = tk.Canvas(main_frame, bg='white')
    scrollbar.config(command=content_canvas.yview)
    content_canvas.config(yscrollcommand=scrollbar.set)

    # 添加卡片布局
    vertical_offset = 0
    for task_info in output_buffer:
        card = create_card(content_canvas, task_info)
        card.pack(padx=5, pady=5)
        vertical_offset = max(vertical_offset, card.winfo_height())

    content_canvas.config(scrollregion=(0, 0, 0, vertical_offset))
    main_frame.pack(fill=tk.BOTH, expand=True)

三、系统流程说明

1. 任务提交：用户通过task_submission.py界面上传文件并填写信息，任务以 JSON 格式保存到input_spool目录。
2. 任务加载：queue_display.py点击 “加载” 按钮后，从input_spool读取任务到print_queue（FCFS 调度）。
3. 任务处理：user_processing_procedure.py模拟用户进程处理任务，修改标题后保存到output_spool目录。
4. 队列展示：处理后的任务信息显示在文本框，同时调用show_output_images.py生成带图片预览的卡片式展示界面。

四、关键技术点

1. 输入 / 输出井设计：
   • input_spool：存放待处理任务（JSON 文件），支持文件系统持久化。
   • output_spool：存放处理后任务，每次处理前自动清空（clear_output_spool方法）。
2. 任务调度算法： 采用 ** 先来先服务（FCFS）** 策略，按任务提交顺序处理，通过load_tasks_to_queue保证顺序性。
3. 图形界面（Tkinter）：
   • queue_display.py使用Text组件展示任务列表，Button触发加载逻辑。
   • show_output_images.py实现带滚动条的图片画廊，支持 PNG/JPG/GIF 格式预览，自动缩放图片保持比例。
4. 异常处理：
   • 任务提交时校验必填字段，防止空数据。
   • 图片加载失败时弹出错误提示，避免程序崩溃。

五、运行与测试

1. 启动系统：

   python main.py  # 打开任务队列展示界面

2. 提交任务： 运行task_submission.py，填写信息并上传图片文件，任务 ID 自动生成。

3. 查看队列与图片： 在队列界面点击 “加载并展示任务队列”，处理后的任务信息和图片会分别显示在文本框和新窗口中。

六、项目地址

Gitee 仓库 （含完整代码及资源文件）