pyboot-dataflow 1.3.3.1
Pyboot-dataflow is an open-source Python microservice framework that simplifies the development of stand-alone production-grade Spring applications.
Requires Python: >=3.12.10
Keywords: web, framework, cli, springboot, fastapi

# PyBoot 框架全面介绍指南

## 第一章：PyBoot 框架概述

### 1.1 框架简介

PyBoot 是一个基于 Python 的现代化全栈 Web 开发框架，其设计理念深受 Java 生态中 Spring Boot 框架的启发。PyBoot 旨在为 Python 开发者提供一个开箱即用、功能完备的企业级应用开发解决方案。通过合理的架构设计和丰富的功能集成，PyBoot 显著降低了构建复杂 Python Web 应用的难度，同时保证了应用的高性能和高可维护性。

PyBoot 框架的核心设计哲学是"约定优于配置"和"开箱即用"。开发者无需花费大量时间进行繁琐的配置，框架已经为大多数常见场景提供了合理的默认配置。同时，PyBoot 保持了高度的灵活性，允许开发者在需要时进行自定义配置，以满足特定的业务需求。

### 1.2 设计理念与架构思想

PyBoot 框架的架构设计遵循了现代软件工程的多个重要原则：

**模块化设计**：PyBoot 采用高度模块化的架构，每个功能模块都相对独立，可以按需引入。这种设计不仅降低了框架的复杂性，还使得开发者能够根据项目需求灵活选择所需功能。

**依赖注入与控制反转**：框架内置了强大的依赖注入容器，实现了控制反转（IoC）的设计模式。这种机制使得组件之间的耦合度大大降低，提高了代码的可测试性和可维护性。

**面向切面编程**：PyBoot 全面支持 AOP（面向切面编程），允许开发者将横切关注点（如日志追踪记录、事务管理、安全控制等）与业务逻辑分离，实现了更好的代码组织和复用。

**配置外部化**：框架支持将配置信息从代码中分离出来，通过外部的 YAML 文件进行管理。同时支持多环境配置，使得应用在不同部署环境下的配置管理变得简单而高效。

### 1.3 核心特性总览

PyBoot 框架提供了一系列强大的核心特性，包括但不限于：

- 内嵌 FastAPI 作为 Web 容器，提供高性能的 Web 服务能力
- 完整的服务容器体系，支持依赖注入和组件生命周期管理
- 强大的定时任务调度系统
- 优化的多线程池管理
- 类似 MyBatis-Plus 的便捷数据库操作
- 完整的消息队列集成（Kafka）
- Redis 缓存和数据存储支持
- 多数据源动态路由
- 基于注解的配置系统
- YAML 配置文件和多环境配置支持
- 可扩展的自定义组件机制
- 灵活的过滤器系统
- 声明式的控制器编程模型
- 完善的数据库事务管理

### 1.4 适用场景

PyBoot 框架适用于各种规模的 Python Web 应用开发，特别适合以下场景：

- 企业级后台管理系统
- 微服务架构中的单个服务
- 高并发的 API 服务
- 需要复杂业务逻辑的数据处理应用
- 需要集成多种数据源和中间件的应用
- 需要良好可维护性和可测试性的长期项目

## 第二章：快速开始

### 2.1 环境要求与安装

在开始使用 PyBoot 之前，需要确保系统满足以下环境要求：

- Python 3.8 或更高版本
- pip 包管理工具
- 可选：虚拟环境工具（如 venv 或 conda）

安装 PyBoot 框架非常简单，可以通过 pip 命令直接安装：

```bash
pip install pyboot-framework
```

或者从源代码安装：

```bash
git clone https://github.com/pyboot/framework.git
cd framework
pip install -e .
```

### 2.2 创建第一个 PyBoot 应用

让我们创建一个简单的 "Hello World" 应用来演示 PyBoot 的基本用法：

**项目结构**：
```
myapp/
├── app.py
├── application.yaml
└── requirements.txt
```

**application.yaml**：
```yaml
app:
  name: my-first-pyboot-app
  version: 1.0.0

server:
  port: 8080
  host: "0.0.0.0"
```

**app.py**：
```python
from pyboot import PyBootApplication
from pyboot.web import controller, GetMapping

@controller
class HelloController:
    
    @GetMapping("/hello")
    def hello(self):
        return {"message": "Hello, PyBoot!"}

if __name__ == "__main__":
    app = PyBootApplication()
    app.run()
```

运行应用：
```bash
python app.py
```

访问 http://localhost:8080/hello 即可看到返回的 JSON 消息。

### 2.3 基本项目结构说明

一个标准的 PyBoot 项目通常具有以下目录结构：

```
project/
├── src/                    # 源代码目录
│   ├── main/              # 主要代码
│   │   ├── python/        # Python 代码
│   │   │   ├── controller/    # 控制器层
│   │   │   ├── service/       # 服务层
│   │   │   ├── model/         # 数据模型
│   │   │   ├── config/         # 配置类
│   │   └── conf/                   # 资源文件
│   │       ├── application.yaml    # 主配置文件
│   │       ├── application-dev.yaml # 开发环境配置
│   │       └── application-prod.yaml # 生产环境配置
│   └── test/              # 测试代码
├── web/                # 静态资源
├── requirements.txt       # 依赖列表
└── README.md             # 项目说明
```

### 2.4 配置文件基础

PyBoot 使用 YAML 格式的配置文件，默认加载 `application.yaml` 文件。基本的配置项包括：

```yaml
app:
  name: "我的应用"
  version: "1.0.0"
  
server:
  port: 8080
  host: "0.0.0.0"
  static:
    path: "/static"
    directory: "./static"
  
logging:
  level: "INFO"
  format: "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
  
database:
  default:
    url: "postgresql://user:pass@localhost:5432/mydb"
    echo: false
```

## 第三章：Web 容器与 FastAPI 集成

### 3.1 内嵌 FastAPI 的优势

PyBoot 选择内嵌 FastAPI 作为其 Web 容器，主要基于以下考虑：

**高性能**：FastAPI 是基于 Starlette 和 Pydantic 的现代 Web 框架，具有极高的性能，能够与 NodeJS 和 Go 相媲美。

**异步支持**：原生支持异步请求处理，能够更好地处理高并发场景。

**自动 API 文档**：自动生成交互式 API 文档（Swagger UI 和 ReDoc），极大提高了 API 的开发效率和可用性。

**类型提示**：充分利用 Python 的类型提示系统，提供更好的代码补全和错误检查。

**易于学习**：简洁的 API 设计使得开发者能够快速上手。

### 3.2 控制器(Controller)详解

在 PyBoot 中，控制器负责处理 HTTP 请求并返回响应。通过装饰器模式，可以轻松定义路由和请求处理方法。

**基本控制器示例**：

```python
from pyboot.web import controller, GetMapping, PostMapping, RequestBody, PathVariable

@controller
class UserController:
    
    @GetMapping("/users")
    def get_all_users(self):
        # 获取所有用户
        return user_service.get_all_users()
    
    @GetMapping("/users/{user_id}")
    def get_user_by_id(self, user_id: int):
        # 根据ID获取用户
        user = user_service.get_user_by_id(user_id)
        if not user:
            return {"error": "User not found"}, 404
        return user
    
    @PostMapping("/users")
    def create_user(self, user_data: RequestBody):
        # 创建新用户
        new_user = user_service.create_user(user_data)
        return new_user, 201
    
    @GetMapping("/users/{user_id}/orders")
    def get_user_orders(self, user_id: int, page: int = 1, size: int = 10):
        # 获取用户的订单，支持分页参数
        orders = order_service.get_orders_by_user(user_id, page, size)
        return {
            "page": page,
            "size": size,
            "total": len(orders),
            "data": orders
        }
```

### 3.3 请求映射与参数处理

PyBoot 支持多种类型的请求参数绑定：

**路径参数**：
```python
@GetMapping("/users/{user_id}/orders/{order_id}")
def get_order(self, user_id: int, order_id: int):
    # 使用路径参数
    return order_service.get_order(user_id, order_id)
```

**查询参数**：
```python
@GetMapping("/users")
def search_users(self, name: str = None, age: int = None, page: int = 1):
    # 使用查询参数，带有默认值
    return user_service.search_users(name, age, page)
```

**请求体参数**：
```python
from pyboot.web import RequestBody
from pydantic import BaseModel

class UserCreateRequest(BaseModel):
    name: str
    email: str
    age: int

@PostMapping("/users")
def create_user(self, user_data: RequestBody[UserCreateRequest]):
    # 使用请求体参数，支持Pydantic模型验证
    return user_service.create_user(user_data)
```

**请求头参数**：
```python
from pyboot.web import RequestHeader

@GetMapping("/profile")
def get_profile(self, authorization: str = RequestHeader()):
    # 获取请求头
    token = authorization.replace("Bearer ", "")
    return auth_service.get_profile(token)
```

### 3.4 静态文件服务配置

PyBoot 支持静态文件服务，可以轻松托管前端资源：

**配置静态文件**：
```yaml
server:
  static:
    - path: "/static"
      directory: "./static"
    - path: "/uploads"
      directory: "./uploads"
      show_index: true
```

**自定义静态文件处理器**：
```python
from pyboot.web import StaticFileConfig

@Configuration
class WebConfig:
    
    @Bean
    def static_file_config(self) -> StaticFileConfig:
        config = StaticFileConfig()
        config.add_mapping("/web", "./web-resources")
        config.add_mapping("/docs", "./documentation", show_index=True)
        return config
```

### 3.5 代理服务与流式响应

PyBoot 提供了强大的代理服务支持，包括常规代理和流式响应代理：

**普通代理服务**：
```python
from pyboot.web import ProxyService

@Bean
def user_proxy_service(self) -> ProxyService:
    service = ProxyService()
    service.add_route("/api/users", "http://user-service:8080")
    service.add_route("/api/orders", "http://order-service:8080")
    return service
```

**流式响应代理**：
```python
from pyboot.web import StreamingProxyService

@Bean
def streaming_proxy_service(self) -> StreamingProxyService:
    service = StreamingProxyService()
    
    # 代理SSE（Server-Sent Events）端点
    service.add_sse_proxy("/events", "http://event-service:8080/events")
    
    # 代理WebSocket端点
    service.add_websocket_proxy("/ws", "http://websocket-service:8080/ws")
    
    return service
```

**自定义流式处理**：
```python
import asyncio
from pyboot.web import GetMapping, controller

@controller
class StreamController:
    
    @GetMapping("/stream-data")
    async def stream_data(self):
        """流式返回数据示例"""
        async def generate_data():
            for i in range(10):
                yield f"data: Message {i}\n\n"
                await asyncio.sleep(1)
        
        return StreamingResponse(
            generate_data(),
            media_type="text/event-stream",
            headers={
                "Cache-Control": "no-cache",
                "Connection": "keep-alive",
            }
        )
```

## 第四章：服务容器与依赖注入

### 4.1 服务容器核心概念

PyBoot 的服务容器是其核心功能之一，它负责管理应用中所有组件的生命周期和依赖关系。服务容器基于依赖注入模式，实现了控制反转（IoC）原则。

**容器的基本功能**：
- 组件的注册与发现
- 依赖关系的自动解析
- 组件生命周期的管理
- 作用域管理（单例、请求作用域等）

### 4.2 组件注册与生命周期

在 PyBoot 中，有多种方式可以注册组件：

**类装饰器方式**：
```python
from pyboot.core import Component, Service, Repository

@Component
class EmailValidator:
    def validate(self, email: str) -> bool:
        return "@" in email

@Service
class UserService:
    def __init__(self, email_validator: EmailValidator):
        self.email_validator = email_validator
    
    def create_user(self, user_data):
        if not self.email_validator.validate(user_data.email):
            raise ValueError("Invalid email address")
        # 创建用户逻辑
        return new_user

@Repository
class UserRepository:
    def save(self, user):
        # 保存用户到数据库
        return saved_user
```

**配置类方式**：
```python
from pyboot.core import Configuration, Bean

@Configuration
class AppConfig:
    
    @Bean
    def data_source(self) -> DataSource:
        return PostgreSQLDataSource()
    
    @Bean
    def user_service(self, data_source: DataSource) -> UserService:
        return UserService(data_source)
```

**组件生命周期**：
```python
from pyboot.core import Component, PostConstruct, PreDestroy

@Component
class CacheManager:
    
    def __init__(self):
        self.cache = {}
    
    @PostConstruct
    def initialize(self):
        """在组件初始化后调用"""
        print("CacheManager initialized")
        # 加载初始缓存数据
        self.load_initial_data()
    
    @PreDestroy
    def cleanup(self):
        """在组件销毁前调用"""
        print("CacheManager cleaning up")
        self.cache.clear()
    
    def load_initial_data(self):
        # 加载初始数据逻辑
        pass
```

### 4.3 依赖注入的多种方式

PyBoot 支持多种依赖注入方式：

**构造函数注入**：
```python
@Service
class OrderService:
    def __init__(self, user_service: UserService, product_service: ProductService):
        self.user_service = user_service
        self.product_service = product_service
```

**属性注入**：
```python
from pyboot.core import Autowired

@Service
class PaymentService:
    
    @Autowired
    private user_service: UserService
    
    @Autowired
    private order_service: OrderService
    
    def process_payment(self, order_id: int):
        order = self.order_service.get_order(order_id)
        user = self.user_service.get_user(order.user_id)
        # 处理支付逻辑
```

**方法注入**：
```python
@Service
class ReportService:
    
    private database_connection: DatabaseConnection
    
    @Autowired
    def set_database_connection(self, connection: DatabaseConnection):
        self.database_connection = connection
```

### 4.4 条件化组件注册

PyBoot 支持基于条件的组件注册，类似于 Spring Boot 的 `@Conditional` 注解：

```python
from pyboot.core import Configuration, Bean, ConditionalOnProperty, ConditionalOnClass

@Configuration
class ConditionalConfig:
    
    @Bean
    @ConditionalOnProperty(name="cache.enabled", having_value="true")
    def cache_manager(self) -> CacheManager:
        return RedisCacheManager()
    
    @Bean
    @ConditionalOnProperty(name="cache.enabled", having_value="false", match_if_missing=True)
    def cache_manager(self) -> CacheManager:
        return SimpleCacheManager()
    
    @Bean
    @ConditionalOnClass("redis.Redis")
    def redis_template(self) -> RedisTemplate:
        return RedisTemplate()
```

### 4.5 配置属性绑定

PyBoot 支持将配置文件中的属性绑定到组件：

```python
from pyboot.core import Component, ConfigurationProperties

@ConfigurationProperties(prefix="app.database")
class DatabaseProperties:
    url: str
    username: str
    password: str
    pool_size: int = 10
    timeout: int = 30

@Component
class DatabaseConfig:
    
    def __init__(self, properties: DatabaseProperties):
        self.properties = properties
    
    @Bean
    def data_source(self) -> DataSource:
        return create_data_source(
            url=self.properties.url,
            username=self.properties.username,
            password=self.properties.password,
            pool_size=self.properties.pool_size,
            timeout=self.properties.timeout
        )
```

对应的配置文件：
```yaml
app:
  database:
    url: "postgresql://localhost:5432/mydb"
    username: "admin"
    password: "secret"
    pool_size: 20
    timeout: 60
```

## 第五章：面向切面编程(AOP)

### 5.1 AOP 基本概念

面向切面编程（AOP）是一种编程范式，旨在将横切关注点（如日志、事务、安全等）与业务逻辑分离。PyBoot 提供了完整的 AOP 支持，让开发者能够以声明式的方式处理这些横切关注点。

**AOP 核心概念**：
- 切面（Aspect）：横切关注点的模块化
- 连接点（Join Point）：程序执行过程中的特定点
- 通知（Advice）：在连接点执行的动作
- 切点（Pointcut）：匹配连接点的谓词
- 引入（Introduction）：为现有类添加新的方法或属性
- 目标对象（Target Object）：被一个或多个切面通知的对象

### 5.2 切面定义与配置

在 PyBoot 中定义切面：

```python
from pyboot.aop import Aspect, Pointcut, Before, After, Around, AfterReturning, AfterThrowing

@Aspect
@Component
class LoggingAspect:
    
    @Pointcut("execution(* com.example.service.*.*(..))")
    def service_methods(self):
        """匹配service包下所有类的所有方法"""
        pass
    
    @Before("service_methods()")
    def log_before(self, joinpoint):
        method_name = joinpoint.method.__name__
        class_name = joinpoint.target.__class__.__name__
        print(f"Before executing {class_name}.{method_name}")
    
    @AfterReturning(pointcut="service_methods()", returning="result")
    def log_after_returning(self, joinpoint, result):
        method_name = joinpoint.method.__name__
        print(f"Method {method_name} returned: {result}")
    
    @AfterThrowing(pointcut="service_methods()", throwing="exception")
    def log_after_throwing(self, joinpoint, exception):
        method_name = joinpoint.method.__name__
        print(f"Method {method_name} threw exception: {exception}")
    
    @Around("service_methods()")
    def measure_performance(self, proceeding_joinpoint):
        import time
        start_time = time.time()
        
        try:
            result = proceeding_joinpoint.proceed()
            return result
        finally:
            execution_time = time.time() - start_time
            method_name = proceeding_joinpoint.method.__name__
            print(f"Method {method_name} executed in {execution_time:.3f}s")
```

### 5.3 五种通知类型详解

PyBoot 支持五种标准的通知类型：

**@Before** - 前置通知：
```python
@Before("execution(* UserService.*(..))")
def validate_arguments(self, joinpoint):
    # 在目标方法执行前进行参数验证
    args = joinpoint.args
    for arg in args:
        if arg is None:
            raise ValueError("Arguments cannot be None")
```

**@AfterReturning** - 返回后通知：
```python
@AfterReturning(
    pointcut="execution(* UserService.get_user(..))",
    returning="user"
)
def audit_user_access(self, joinpoint, user):
    # 在成功获取用户信息后记录审计日志
    if user:
        user_id = user.id
        accessed_by = get_current_user()
        audit_service.log_access(user_id, accessed_by)
```

**@AfterThrowing** - 异常通知：
```python
@AfterThrowing(
    pointcut="execution(* OrderService.*(..))",
    throwing="ex"
)
def handle_order_errors(self, joinpoint, ex):
    # 处理订单服务中的异常
    error_msg = f"Error in {joinpoint.method.__name__}: {str(ex)}"
    error_service.report_error(error_msg, severity="HIGH")
```

**@After** - 后置通知：
```python
@After("execution(* DatabaseService.*(..))")
def cleanup_resources(self, joinpoint):
    # 无论方法执行成功与否，都进行资源清理
    database_connection.cleanup_temp_resources()
```

**@Around** - 环绕通知：
```python
@Around("execution(* ExternalService.call_api(..))")
def retry_on_failure(self, proceeding_joinpoint):
    # 在调用外部API时实现重试机制
    max_attempts = 3
    last_exception = None
    
    for attempt in range(max_attempts):
        try:
            result = proceeding_joinpoint.proceed()
            return result
        except TemporaryError as ex:
            last_exception = ex
            if attempt < max_attempts - 1:
                time.sleep(2 ** attempt)  # 指数退避
            continue
    
    # 所有重试都失败了
    raise ServiceUnavailableError("Service unavailable after retries") from last_exception
```

### 5.4 切点表达式语法

PyBoot 使用强大的切点表达式来匹配连接点：

**执行表达式**：
```python
# 匹配特定方法
@Pointcut("execution(public String com.example.UserService.getUserName(int))")

# 匹配包下所有方法
@Pointcut("execution(* com.example.service.*.*(..))")

# 匹配特定类所有方法
@Pointcut("execution(* com.example.UserService.*(..))")

# 匹配所有public方法
@Pointcut("execution(public * *(..))")

# 匹配所有以get开头的方法
@Pointcut("execution(* *.get*(..))")
```

**Within 表达式**：
```python
# 匹配包内所有方法
@Pointcut("within(com.example.service..*)")

# 匹配特定类
@Pointcut("within(com.example.UserService)")

# 匹配注解标注的类
@Pointcut("@within(com.example.Transactional)")
```

**注解表达式**：
```python
# 匹配带有特定注解的方法
@Pointcut("@annotation(com.example.Cacheable)")

# 匹配带有特定注解的参数
@Pointcut("@args(com.example.Validated)")

# 匹配带有特定注解的类的方法
@Pointcut("@within(com.example.Secured)")
```

### 5.5 实际应用场景

**事务管理切面**：
```python
@Aspect
@Component
class TransactionAspect:
    
    @Autowired
    def set_transaction_manager(self, tx_manager: TransactionManager):
        self.tx_manager = tx_manager
    
    @Around("@annotation(transactional)")
    def manage_transaction(self, proceeding_joinpoint, transactional):
        tx = self.tx_manager.begin_transaction(
            isolation=transactional.isolation,
            read_only=transactional.read_only
        )
        
        try:
            result = proceeding_joinpoint.proceed()
            self.tx_manager.commit_transaction(tx)
            return result
        except Exception as ex:
            self.tx_manager.rollback_transaction(tx)
            raise ex
```

**缓存切面**：
```python
@Aspect
@Component
class CacheAspect:
    
    @Autowired
    def set_cache_manager(self, cache_manager: CacheManager):
        self.cache_manager = cache_manager
    
    @Around("@annotation(cacheable)")
    def cache_result(self, proceeding_joinpoint, cacheable):
        # 生成缓存键
        cache_key = self.generate_cache_key(
            proceeding_joinpoint.method,
            proceeding_joinpoint.args
        )
        
        # 尝试从缓存获取
        cached_result = self.cache_manager.get(cache_key)
        if cached_result is not None:
            return cached_result
        
        # 执行方法并缓存结果
        result = proceeding_joinpoint.proceed()
        self.cache_manager.set(
            cache_key, 
            result, 
            ttl=cacheable.ttl
        )
        
        return result
    
    @AfterReturning("@annotation(cache_evict)")
    def evict_cache(self, joinpoint, cache_evict):
        # 根据配置清除缓存
        if cache_evict.all_entries:
            self.cache_manager.clear()
        else:
            cache_key = self.generate_cache_key(
                joinpoint.method,
                joinpoint.args
            )
            self.cache_manager.delete(cache_key)
```

## 第六章：定时任务调度

### 6.1 定时任务基础

PyBoot 提供了强大的定时任务调度功能，支持多种类型的任务调度需求。定时任务系统基于 cron 表达式和简单间隔，可以轻松配置周期性执行的任务。

**启用定时任务**：
```python
from pyboot.scheduling import EnableScheduling

@EnableScheduling
@PyBootApplication
class MyApplication:
    def main(self):
        app = PyBootApplication(MyApplication)
        app.run()
```

### 6.2 多种调度方式

**固定速率调度**：
```python
from pyboot.scheduling import Scheduled, Component

@Component
class FixedRateTasks:
    
    @Scheduled(fixed_rate=5000)  # 每5秒执行一次
    def report_metrics(self):
        """每5秒报告一次系统指标"""
        metrics = system_metrics_collector.collect()
        metrics_reporter.report(metrics)
    
    @Scheduled(fixed_rate=60000, initial_delay=10000)  # 启动后10秒开始，每60秒执行
    def cleanup_temp_files(self):
        """每分钟清理一次临时文件"""
        temp_file_cleaner.cleanup()
```

**固定延迟调度**：
```python
@Component
class FixedDelayTasks:
    
    @Scheduled(fixed_delay=30000)  # 上次执行完成后30秒再执行
    def process_batch_data(self):
        """批处理数据，确保每次执行间隔至少30秒"""
        batch_processor.process_next_batch()
```

**Cron 表达式调度**：
```python
@Component
class CronTasks:
    
    @Scheduled(cron="0 0 * * * *")  # 每小时执行一次
    def hourly_backup(self):
        """每小时执行一次数据备份"""
        backup_service.create_hourly_backup()
    
    @Scheduled(cron="0 0 2 * * *")  # 每天凌晨2点执行
    def daily_report(self):
        """每天生成日报"""
        report_generator.generate_daily_report()
    
    @Scheduled(cron="0 0 0 * * MON")  # 每周一凌晨执行
    def weekly_cleanup(self):
        """每周执行一次大清理"""
        system_cleaner.perform_weekly_cleanup()
```

### 6.3 动态定时任务

PyBoot 支持动态创建和管理定时任务：

```python
from pyboot.scheduling import TaskScheduler, Trigger, TaskRegistrar

@Component
class DynamicTaskManager:
    
    def __init__(self, task_scheduler: TaskScheduler):
        self.task_scheduler = task_scheduler
        self.scheduled_tasks = {}
    
    def schedule_task(self, task_id: str, cron_expression: str, task_function):
        """动态调度任务"""
        trigger = Trigger.cron(cron_expression)
        task = self.task_scheduler.schedule_task(task_function, trigger)
        self.scheduled_tasks[task_id] = task
    
    def reschedule_task(self, task_id: str, new_cron_expression: str):
        """重新调度任务"""
        if task_id in self.scheduled_tasks:
            self.cancel_task(task_id)
            # 重新创建任务
            task_function = self.scheduled_tasks[task_id].function
            self.schedule_task(task_id, new_cron_expression, task_function)
    
    def cancel_task(self, task_id: str):
        """取消任务"""
        if task_id in self.scheduled_tasks:
            self.task_scheduler.cancel_scheduled_task(self.scheduled_tasks[task_id])
            del self.scheduled_tasks[task_id]
```

### 6.4 任务执行配置

**配置任务执行器**：
```python
from pyboot.scheduling import SchedulingConfig, TaskExecutor

@Configuration
class SchedulingConfiguration:
    
    @Bean
    def task_executor(self) -> TaskExecutor:
        from concurrent.futures import ThreadPoolExecutor
        return ThreadPoolExecutor(
            max_workers=10,
            thread_name_prefix="scheduled-task-"
        )
```

**任务异常处理**：
```python
@Component
class ScheduledTaskErrorHandler:
    
    @EventListener
    def handle_task_exception(self, event: TaskExecutionExceptionEvent):
        exception = event.exception
        task_method = event.task_method
        logger.error(f"Task {task_method} failed with exception: {exception}")
        
        # 发送告警
        alert_service.send_alert(
            f"Scheduled task failed: {task_method}",
            severity="ERROR"
        )
```

## 第七章：多线程池管理

### 7.1 线程池配置与管理

PyBoot 提供了强大的线程池管理功能，可以轻松配置和管理多种用途的线程池。

**线程池配置类**：
```python
from pyboot.executor import ThreadPoolConfig, EnableThreadPools

@Configuration
@EnableThreadPools
class ExecutorConfig:
    
    @Bean
    def io_thread_pool(self) -> ThreadPoolConfig:
        """I/O密集型任务线程池"""
        return ThreadPoolConfig(
            name="io-executor",
            core_pool_size=20,
            max_pool_size=100,
            queue_capacity=1000,
            keep_alive_seconds=60,
            thread_name_prefix="io-worker-"
        )
    
    @Bean
    def cpu_thread_pool(self) -> ThreadPoolConfig:
        """CPU密集型任务线程池"""
        return ThreadPoolConfig(
            name="cpu-executor",
            core_pool_size=4,  # 通常设置为CPU核心数
            max_pool_size=8,
            queue_capacity=100,
            keep_alive_seconds=30,
            thread_name_prefix="cpu-worker-"
        )
    
    @Bean
    def scheduled_thread_pool(self) -> ThreadPoolConfig:
        """定时任务线程池"""
        return ThreadPoolConfig(
            name="scheduled-executor",
            core_pool_size=5,
            max_pool_size=20,
            queue_capacity=500,
            thread_name_prefix="scheduled-"
        )
```

### 7.2 线程池使用示例

**注入和使用线程池**：
```python
from pyboot.executor import ThreadPoolExecutor, Async

@Service
class DataProcessingService:
    
    def __init__(self, io_executor: ThreadPoolExecutor):
        self.io_executor = io_executor
    
    def process_large_dataset(self, dataset: List[Data]) -> List[ProcessedData]:
        """使用线程池并行处理大数据集"""
        futures = []
        
        # 分批提交任务
        batch_size = 100
        for i in range(0, len(dataset), batch_size):
            batch = dataset[i:i + batch_size]
            future = self.io_executor.submit(self.process_batch, batch)
            futures.append(future)
        
        # 收集结果
        results = []
        for future in futures:
            try:
                batch_result = future.result(timeout=300)  # 5分钟超时
                results.extend(batch_result)
            except TimeoutError:
                logger.error("Batch processing timeout")
        
        return results
    
    def process_batch(self, batch: List[Data]) -> List[ProcessedData]:
        """处理单个数据批次"""
        return [self.process_single_item(item) for item in batch]
```

**异步方法执行**：
```python
@Service
class AsyncService:
    
    @Async("io_executor")  # 指定使用I/O线程池
    def async_process_data(self, data: Data) -> ProcessedData:
        """异步处理数据"""
        # 模拟耗时操作
        time.sleep(2)
        return self.process_data(data)
    
    @Async  # 使用默认线程池
    def async_send_notification(self, user: User, message: str):
        """异步发送通知"""
        notification_service.send(user, message)
    
    def process_multiple_async(self, items: List[Data]) -> List[ProcessedData]:
        """并行处理多个异步任务"""
        futures = [self.async_process_data(item) for item in items]
        
        # 等待所有任务完成
        results = []
        for future in futures:
            try:
                result = future.get(timeout=30)
                results.append(result)
            except Exception as e:
                logger.error(f"Async task failed: {e}")
        
        return results
```

### 7.3 线程池监控与管理

**线程池监控**：
```python
@Component
class ThreadPoolMonitor:
    
    def __init__(self, thread_pool_manager: ThreadPoolManager):
        self.thread_pool_manager = thread_pool_manager
    
    @Scheduled(fixed_rate=30000)  # 每30秒监控一次
    def monitor_thread_pools(self):
        """监控所有线程池状态"""
        pools = self.thread_pool_manager.get_all_pools()
        
        for pool_name, pool in pools.items():
            stats = pool.get_statistics()
            
            # 记录监控指标
            self.record_metrics(pool_name, stats)
            
            # 检查异常情况
            if stats.active_count > stats.max_pool_size * 0.8:
                self.alert_high_usage(pool_name, stats)
            
            if stats.queue_size > stats.queue_capacity * 0.9:
                self.alert_queue_full(pool_name, stats)
    
    def record_metrics(self, pool_name: str, stats: ThreadPoolStats):
        metrics.record_gauge(f"thread_pool.{pool_name}.active_count", stats.active_count)
        metrics.record_gauge(f"thread_pool.{pool_name}.queue_size", stats.queue_size)
        metrics.record_gauge(f"thread_pool.{pool_name}.completed_count", stats.completed_count)
```

**动态线程池调整**：
```python
@Service
class DynamicThreadPoolManager:
    
    def __init__(self, thread_pool_manager: ThreadPoolManager):
        self.thread_pool_manager = thread_pool_manager
    
    def adjust_thread_pool_size(self, pool_name: str, new_core_size: int, new_max_size: int):
        """动态调整线程池大小"""
        pool = self.thread_pool_manager.get_thread_pool(pool_name)
        if pool:
            pool.set_core_pool_size(new_core_size)
            pool.set_maximum_pool_size(new_max_size)
            logger.info(f"Adjusted {pool_name} to core={new_core_size}, max={new_max_size}")
    
    @EventListener
    def handle_high_load_event(self, event: SystemHighLoadEvent):
        """根据系统负载事件自动调整线程池"""
        if event.metric == "cpu_usage" and event.value > 0.8:
            # CPU使用率高，减少CPU密集型线程池
            self.adjust_thread_pool_size("cpu-executor", 2, 4)
        elif event.metric == "io_wait" and event.value > 0.5:
            # I/O等待高，增加I/O线程池
            self.adjust_thread_pool_size("io-executor", 30, 150)
```

## 第八章：数据库操作与 MyBatis-Plus 风格功能

### 8.1 数据访问层配置

PyBoot 提供了类似 MyBatis-Plus 的便捷数据库操作功能，支持多种数据库和灵活的查询方式。

**数据库配置**：
```yaml
database:
  default:
    url: "postgresql://user:pass@localhost:5432/mydb"
    driver: "postgresql"
    host: "localhost"
    port: 5432
    database: "mydb"
    username: "user"
    password: "pass"
    pool:
      max_connections: 20
      min_connections: 5
      max_lifetime: 3600
  read_replica:
    url: "postgresql://user:pass@replica:5432/mydb"
    read_only: true
```

**实体类定义**：
```python
from pyboot.data import Entity, Table, Column, Id, GeneratedValue

@Table(name = "users")
class User:
    
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    @Column(name = "id")
    def id(self) -> int:
        return self._id
    
    @Column(name = "username", length = 50, unique = True, nullable = False)
    def username(self) -> str:
        return self._username
    
    @Column(name = "email", length = 100, unique = True, nullable = False)
    def email(self) -> str:
        return self._email
    
    @Column(name = "created_at")
    def created_at(self) -> datetime:
        return self._created_at
    
    @Column(name = "updated_at")
    def updated_at(self) -> datetime:
        return self._updated_at
```

### 8.2 通用 Mapper 功能

PyBoot 的通用 Mapper 提供了丰富的 CRUD 操作方法：

**基础 Repository**：
```python
from pyboot.data import Repository, BaseMapper

@Repository
class UserRepository(BaseMapper[User, int]):
    """用户数据访问接口"""
    
    def find_by_username(self, username: str) -> Optional[User]:
        return self.select_one(
            self.table.username == username
        )
    
    def find_by_email(self, email: str) -> Optional[User]:
        return self.select_one(
            self.table.email == email
        )
    
    def find_active_users(self) -> List[User]:
        return self.select_list(
            self.table.status == UserStatus.ACTIVE
        )
    
    def find_by_create_time_range(self, start: datetime, end: datetime) -> List[User]:
        return self.select_list(
            (self.table.created_at >= start) & 
            (self.table.created_at <= end)
        )
```

**复杂查询方法**：
```python
@Repository
class OrderRepository(BaseMapper[Order, int]):
    """订单数据访问接口"""
    
    def find_user_orders(self, user_id: int, page: int = 1, size: int = 10) -> Page[Order]:
        return self.select_page(
            page=page,
            size=size,
            where=self.table.user_id == user_id,
            order_by=[self.table.created_at.desc()]
        )
    
    def find_orders_with_details(self, order_ids: List[int]) -> List[Order]:
        # 使用连接查询获取订单详情
        return self.select_list(
            self.table.id.in_(order_ids)
        ).join(OrderItem).on(
            self.table.id == OrderItem.order_id
        ).fetch_all()
    
    def calculate_user_total_spent(self, user_id: int) -> float:
        result = self.select(
            func.sum(Order.total_amount)
        ).where(
            self.table.user_id == user_id,
            self.table.status == OrderStatus.COMPLETED
        ).scalar()
        
        return result or 0.0
```

### 8.3 查询构造器

PyBoot 提供了强大的查询构造器，支持复杂的查询条件：

**查询构造器示例**：
```python
@Service
class UserService:
    
    def __init__(self, user_repository: UserRepository):
        self.user_repository = user_repository
    
    def search_users(self, criteria: UserSearchCriteria) -> Page[User]:
        """复杂用户搜索"""
        query = self.user_repository.query_builder()
        
        # 动态添加查询条件
        if criteria.username:
            query = query.where(self.user_repository.table.username.like(f"%{criteria.username}%"))
        
        if criteria.email:
            query = query.where(self.user_repository.table.email.like(f"%{criteria.email}%"))
        
        if criteria.min_age:
            query = query.where(self.user_repository.table.age >= criteria.min_age)
        
        if criteria.max_age:
            query = query.where(self.user_repository.table.age <= criteria.max_age)
        
        if criteria.roles:
            query = query.where(self.user_repository.table.role.in_(criteria.roles))
        
        # 排序和分页
        return query.order_by(
            self.user_repository.table.created_at.desc()
        ).page(
            page=criteria.page, 
            size=criteria.size
        )
    
    def get_user_statistics(self) -> UserStatistics:
        """用户统计信息"""
        total_users = self.user_repository.select_count()
        active_users = self.user_repository.select_count(
            self.user_repository.table.status == UserStatus.ACTIVE
        )
        new_today = self.user_repository.select_count(
            self.user_reposi