PySide6 桌面应用实践项目

这是一个基于 PySide6 和 QFluentWidgets 开发的桌面应用程序，重点展示了 queue_sqlite 在桌面应用中的实践应用，实现了高效的异步任务处理和消息队列机制。

项目特点

• 🖥️ 现代化界面：采用 Fluent Design 设计风格
• 🧩 模块化架构：功能模块清晰分离，便于扩展和维护
• 🔄 异步任务处理：基于 queue_sqlite 的高效任务队列系统
• 📡 消息队列机制：可靠的消息传递和处理框架
• 🌐 前后端通信：通过队列系统实现前后端解耦

核心技术：queue_sqlite 的应用

队列调度器初始化

在 app/scheduler_manager/__init__.py 中初始化队列调度器：

from queue_sqlite.scheduler import QueueScheduler

scheduler = QueueScheduler()

任务定义与挂载

在 app/scheduler_manager/students/add_student.py 中定义任务：

from queue_sqlite.mounter.task_mounter import TaskMounter
from queue_sqlite.model import MessageItem

@TaskMounter.task(meta={"task_name": "add_student"})
def add_student(message_item: MessageItem):
    # 模拟耗时操作
    time.sleep(3)
    print("点击了添加按钮")
    return {"message": "学生添加成功"}

监听器定义

在 app/scheduler_manager/students/students_listen.py 中定义监听器：

from queue_sqlite.mounter.listen_mounter import ListenMounter

@ListenMounter.listener()
def students(str_students_list: str):
    print("students signal received")
    listen_signals.listening_students_signal.emit(json.loads(str_students_list))

消息发送与处理

在 app/view/students/student_interface.py 中发送消息：

def add_student(self):
    self.addButton.setEnabled(False)
    scheduler.send_message(
        MessageItem(content={}, destination="add_student"),
        self.add_student_callback,
    )

前端与队列系统的交互

在 app/view/web_view/entrance.py 中通过 WebChannel 暴露队列操作给前端：

class WebViewQueueSqliteOperations(QObject):
    @Slot(str, str, result=str)
    def send_message(self, message: str, store_key: str):
        def callback(result: MessageItem):
            global_signals.update_queue_state.emit(result.to_dict(), store_key)
        
        scheduler.send_message(MessageItem.from_dict(json.loads(message)), callback)

项目结构

项目根目录/
├── main.py                 # 应用入口点，启动队列调度器
├── app/                    # 应用主要代码
│   ├── config/            # 配置管理
│   ├── scheduler_manager/ # 任务调度管理（queue_sqlite核心）
│   │   ├── __init__.py    # 初始化队列调度器
│   │   └── students/      # 学生相关任务和监听器
│   ├── signal/            # 信号与事件处理
│   ├── style/             # 样式定义
│   └── view/              # 界面视图
│       ├── camera/        # 摄像头界面
│       ├── students/      # 学生管理界面（使用队列）
│       ├── video/         # 视频播放界面
│       └── web_view/      # Web视图界面（与队列交互）
└── resources/             # 资源文件
    └── webview/           # Web前端资源
        ├── entrance/      # 主门户界面（调用队列接口）
        └── new_plugin/    # 插件模板

queue_sqlite 实践详解

1. 任务调度器管理

在应用启动时初始化队列调度器，并在应用退出时正确关闭：

# main.py
def main():
    # 启动队列调度器
    scheduler.start_queue_scheduler()
    
    # ... 应用逻辑 ...
    
    # 停止队列调度器
    scheduler.stop_queue_scheduler()

2. 异步任务处理

通过队列系统处理耗时操作，保持UI响应：

# 发送消息到队列
scheduler.send_message(
    MessageItem(content={}, destination="add_student"),
    self.add_student_callback,  # 回调函数
)

# 处理回调
def add_student_callback(self, message: MessageItem):
    global_signals.show_add_dialog_signal.emit(message.to_dict())

3. 前后端解耦

通过消息队列实现前后端解耦，前端发送消息，后端异步处理：

// 前端发送消息到队列
queueSqlite.send_message(
    JSON.stringify({
        "content": {"num": 1}, 
        "destination": "test"
    }), 
    "taskStatus"
)

4. 状态管理

通过队列系统的回调机制更新前端状态：

def update_queue_state(self, result: dict, store_key: str):
    json_str = json.dumps(result)
    js_code = f"""
    try {{
        if (typeof window.queueStore.updateQueueData === 'function') {{
            window.queueStore.updateQueueData('{store_key}', {json_str});
        }}
    }} catch (error) {{
        console.error('Error in updateQueueData:', error);
    }}
    """
    self.page.runJavaScript(js_code)

5. 错误处理与重试

队列系统内置错误处理和重试机制，提高系统可靠性。

快速开始

环境要求

• Python 3.8+
• PySide6
• queue-sqlite

安装依赖

# 初始化项目
mkdir sqlite_queue_demo
cd sqlite_queue_demo
uv init
# 安装依赖
uv add PySide6 qfluentwidgets opencv-python
# 克隆 queue_sqlite 项目
mkdir lib
cd lib
git clone https://gitee.com/cai-xinpenge/pyside_queue_sqlite.git
或
git clone http://124.71.68.6:3000/chakcy_code_repository/queue_sqlite.git
# 安装 queue_sqlite 依赖
cd queue_sqlite
uv sync
# 激活环境
.venv/Scripts/activate
# 构建 Rust 核心模块
cd src/core
maturin develop --release
# 将会在 src/core/target/release 目录下生成 core.dll 或 core.so 文件。 将该文件复制到 queue_sqlite/core 目录下（dll文件需改名为pyd后缀）。
mv target/release/core.dll ../queue_sqlite/core/core.pyd
deactivate
# 安装 queue_sqlite 项目
cd ../../../../
uv add lib/queue_sqlite

运行应用

python main.py

queue_sqlite 最佳实践

1. 任务定义

• 使用装饰器标记任务函数
• 明确任务名称便于管理和调试
• 合理设计任务参数和返回值

2. 消息设计

• 使用 MessageItem 封装消息内容
• 明确消息目的地（destination）
• 包含必要的元数据

3. 回调处理

• 为每个消息指定回调函数
• 在回调中更新UI状态
• 处理可能的错误情况

4. 监听器使用

• 使用监听器响应数据变化
• 通过信号机制通知UI更新
• 保持监听器简洁高效

扩展开发

添加新任务

1. 在 scheduler_manager 下创建新模块
2. 使用 @TaskMounter.task 装饰器定义任务
3. 在界面中发送消息到该任务

前端集成

1. 通过 WebChannel 暴露队列操作接口
2. 在前端调用队列接口发送消息
3. 通过回调更新前端状态

性能优化建议

1. 合理设置并发数：根据任务类型调整队列并发数量
2. 任务优先级：为重要任务设置更高优先级
3. 结果缓存：对频繁请求的任务结果进行缓存
4. 批量处理：对相似任务进行批量处理提高效率

学习价值

本项目重点展示了：

• queue_sqlite 在桌面应用中的集成方式
• 异步任务处理的最佳实践
• 前后端通过消息队列解耦的方案
• 基于回调的状态管理机制
• 错误处理和重试策略

后续计划

• 增加队列监控界面
• 实现任务优先级管理
• 添加任务持久化存储
• 优化队列性能指标
• 增加分布式队列支持

贡献指南

欢迎提交 Issue 和 Pull Request 来帮助改进这个项目，特别是与 queue_sqlite 相关的优化和实践。

许可证

MIT License