向量化系统阶段二实现指南——异步任务调度

📋 概述

Phase 2 在 Phase 1 基础向量化能力之上，实现完整的异步任务调度系统，提供：

✨ 异步任务队列: 基于优先级的任务队列管理
✨ 并发控制: 灵活的线程池和并发策略
✨ 自动重试: 失败任务自动重试机制
✨ 进度推送: SSE 实时推送任务进度
✨ 任务管理: 任务取消、暂停、恢复
✨ 性能优化: 批处理优化和资源调度

🎯 实现目标

核心功能

功能	说明	优先级
优先级队列	支持高/中/低优先级任务调度	P0
异步执行	任务提交后立即返回，后台异步处理	P0
自动重试	失败任务按策略自动重试	P0
任务取消	支持取消正在执行的任务	P1
进度推送	SSE 实时推送任务执行进度	P1
并发控制	动态调整并发数量	P1
任务持久化	任务队列持久化到 Redis	P2

性能目标

支持 1000+ 并发任务队列
单任务提交响应时间 < 100ms
进度推送延迟 < 500ms
任务失败重试间隔 1s/2s/5s (指数退避)

🏗️ 架构设计

整体架构

┌─────────────────────────────────────────────────────────────┐
│                        客户端                                │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐                  │
│  │ Web UI   │  │ API调用  │  │ 定时任务  │                  │
│  └────┬─────┘  └────┬─────┘  └────┬─────┘                  │
└───────┼────────────┼────────────┼─────────────────────────┘
        │            │            │
        ▼            ▼            ▼
┌─────────────────────────────────────────────────────────────┐
│                  VectorizationController                     │
│              ┌─────────────────────────┐                     │
│              │  submitAsyncTask()      │                     │
│              │  cancelTask()           │                     │
│              │  streamProgress() SSE   │                     │
│              └───────────┬─────────────┘                     │
└────────────────────────┼───────────────────────────────────┘
                         │
                         ▼
┌─────────────────────────────────────────────────────────────┐
│              VectorizationScheduler (核心调度器)             │
│  ┌──────────────────────────────────────────────────────┐  │
│  │  submitTask()     提交任务到队列                      │  │
│  │  cancelTask()     取消任务                            │  │
│  │  retryTask()      重试失败任务                        │  │
│  │  processTasks()   从队列获取并执行任务 (@Scheduled)   │  │
│  └──────────────────────────────────────────────────────┘  │
│                         │                                    │
│  ┌─────────────┐   ┌────────────┐   ┌──────────────┐       │
│  │ TaskQueue   │   │ Executor   │   │ RetryPolicy  │       │
│  │ (优先级队列) │   │ (线程池)    │   │ (重试策略)    │       │
│  └─────────────┘   └────────────┘   └──────────────┘       │
└─────────────────────────────────────────────────────────────┘
           │                    │                    │
           ▼                    ▼                    ▼
┌─────────────────┐  ┌─────────────────┐  ┌─────────────────┐
│ Redis           │  │ BatchJobExecutor│  │ ProgressPublisher│
│ (队列持久化)     │  │ (批量执行器)     │  │ (SSE进度推送)    │
└─────────────────┘  └─────────────────┘  └─────────────────┘

📦 核心组件实现

1. VectorizationScheduler - 任务调度器

位置: com.pig4cloud.pigx.ai.rag.vector.scheduler.VectorizationScheduler

java

package com.pig4cloud.pigx.ai.rag.vector.scheduler;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 向量化任务调度器
 *
 * 核心职责:
 * 1. 管理优先级任务队列
 * 2. 调度任务到线程池执行
 * 3. 处理任务取消和重试
 * 4. 监控任务执行状态
 *
 * @author pigx
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class VectorizationScheduler {

    private final VectorizationService vectorizationService;
    private final VectorizationTaskMapper taskMapper;
    private final TaskProgressPublisher progressPublisher;
    private final RetryPolicyConfig retryConfig;

    /**
     * 优先级任务队列（内存队列）
     * 优先级: 数字越小优先级越高
     */
    private final PriorityBlockingQueue<VectorizationTaskWrapper> taskQueue =
        new PriorityBlockingQueue<>(100, Comparator.comparing(VectorizationTaskWrapper::getPriority));

    /**
     * 任务执行线程池
     */
    private final ThreadPoolExecutor executor;

    /**
     * 正在执行的任务
     */
    private final ConcurrentHashMap<String, Future<?>> runningTasks = new ConcurrentHashMap<>();

    /**
     * 任务取消标记
     */
    private final ConcurrentHashMap<String, Boolean> cancelledTasks = new ConcurrentHashMap<>();

    /**
     * 构造线程池
     */
    @PostConstruct
    public void initThreadPool() {
        executor = new ThreadPoolExecutor(
            5,                              // 核心线程数
            10,                             // 最大线程数
            60L,                            // 空闲线程存活时间
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(100), // 工作队列
            new ThreadFactory() {
                private final AtomicInteger counter = new AtomicInteger(0);
                @Override
                public Thread newThread(Runnable r) {
                    Thread thread = new Thread(r, "vectorization-worker-" + counter.incrementAndGet());
                    thread.setDaemon(false);
                    return thread;
                }
            },
            new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略：调用者执行
        );
    }

    /**
     * 提交异步任务
     *
     * @param documentIds 文档ID列表
     * @param config 配置
     * @param priority 优先级 (1-10, 1最高)
     * @return 任务ID
     */
    public String submitAsyncTask(List<Long> documentIds, VectorizationConfig config, Integer priority) {
        // 创建任务实体
        VectorizationTaskEntity task = createTask(documentIds, config, priority);

        // 保存到数据库
        taskMapper.insert(task);

        // 包装任务并加入队列
        VectorizationTaskWrapper wrapper = new VectorizationTaskWrapper(task, priority);
        taskQueue.offer(wrapper);

        log.info("任务已提交到队列: taskId={}, priority={}, queueSize={}",
            task.getTaskId(), priority, taskQueue.size());

        return task.getTaskId();
    }

    /**
     * 定时处理任务队列
     * 每秒检查一次队列，获取任务并提交到线程池执行
     */
    @Scheduled(fixedDelay = 1000)
    public void processTasks() {
        // 检查线程池是否有空闲
        int activeCount = executor.getActiveCount();
        int maxPoolSize = executor.getMaximumPoolSize();
        int availableSlots = maxPoolSize - activeCount;

        if (availableSlots <= 0) {
            return; // 线程池已满，等待下次调度
        }

        // 从队列获取任务（最多获取 availableSlots 个）
        for (int i = 0; i < availableSlots; i++) {
            VectorizationTaskWrapper wrapper = taskQueue.poll();
            if (wrapper == null) {
                break; // 队列为空
            }

            VectorizationTaskEntity task = wrapper.getTask();

            // 检查任务是否已取消
            if (cancelledTasks.containsKey(task.getTaskId())) {
                updateTaskStatus(task.getTaskId(), TaskStatus.CANCELLED);
                cancelledTasks.remove(task.getTaskId());
                continue;
            }

            // 提交到线程池执行
            Future<?> future = executor.submit(() -> executeTask(task));
            runningTasks.put(task.getTaskId(), future);

            log.info("任务开始执行: taskId={}, activeThreads={}", task.getTaskId(), executor.getActiveCount());
        }
    }

    /**
     * 执行单个任务
     */
    private void executeTask(VectorizationTaskEntity task) {
        String taskId = task.getTaskId();

        try {
            // 更新状态为 RUNNING
            task.setStatus(TaskStatus.RUNNING);
            task.setStartedAt(LocalDateTime.now());
            taskMapper.updateById(task);

            // 解析文档ID列表
            List<Long> documentIds = parseDocumentIds(task.getDocumentIds());
            VectorizationConfig config = parseConfig(task.getConfig());

            // 执行向量化（批量处理）
            int totalDocs = documentIds.size();
            int processedDocs = 0;
            List<String> errors = new ArrayList<>();

            for (Long documentId : documentIds) {
                // 检查任务是否被取消
                if (cancelledTasks.containsKey(taskId)) {
                    updateTaskStatus(taskId, TaskStatus.CANCELLED);
                    return;
                }

                try {
                    // 向量化单个文档
                    VectorizationResult result = vectorizationService.vectorizeDocument(documentId, config);

                    if (result.getSuccess()) {
                        processedDocs++;

                        // 更新进度
                        task.setProcessedDocuments(processedDocs);
                        task.setTotalChunks(task.getTotalChunks() + result.getTotalChunks());
                        taskMapper.updateById(task);

                        // 推送进度
                        publishProgress(taskId, processedDocs, totalDocs);
                    } else {
                        errors.add("文档 " + documentId + " 向量化失败: " + result.getErrorMessage());
                    }

                } catch (Exception e) {
                    log.error("文档向量化失败: documentId={}", documentId, e);
                    errors.add("文档 " + documentId + " 向量化异常: " + e.getMessage());
                }
            }

            // 计算耗时
            long durationMs = Duration.between(task.getStartedAt(), LocalDateTime.now()).toMillis();

            // 更新最终状态
            task.setProcessedDocuments(processedDocs);
            task.setFailedDocuments(totalDocs - processedDocs);
            task.setStatus(errors.isEmpty() ? TaskStatus.SUCCESS : TaskStatus.FAILED);
            task.setErrorMessage(errors.isEmpty() ? null : String.join("; ", errors));
            task.setDurationMs(durationMs);
            task.setCompletedAt(LocalDateTime.now());
            taskMapper.updateById(task);

            log.info("任务执行完成: taskId={}, processed={}/{}, duration={}ms",
                taskId, processedDocs, totalDocs, durationMs);

        } catch (Exception e) {
            log.error("任务执行异常: taskId={}", taskId, e);
            handleTaskFailure(task, e);

        } finally {
            runningTasks.remove(taskId);
        }
    }

    /**
     * 处理任务失败（重试逻辑）
     */
    private void handleTaskFailure(VectorizationTaskEntity task, Exception e) {
        Integer retryCount = task.getRetryCount() != null ? task.getRetryCount() : 0;
        Integer maxRetries = task.getMaxRetries() != null ? task.getMaxRetries() : 3;

        if (retryCount < maxRetries) {
            // 可以重试
            task.setRetryCount(retryCount + 1);
            task.setStatus(TaskStatus.PENDING);
            task.setErrorMessage("第 " + (retryCount + 1) + " 次重试: " + e.getMessage());
            taskMapper.updateById(task);

            // 计算退避时间（指数退避: 1s, 2s, 4s, 8s, ...）
            long backoffMs = (long) Math.pow(2, retryCount) * 1000;

            log.warn("任务将在 {}ms 后重试: taskId={}, retryCount={}/{}",
                backoffMs, task.getTaskId(), retryCount + 1, maxRetries);

            // 延迟重新提交到队列
            CompletableFuture.delayedExecutor(backoffMs, TimeUnit.MILLISECONDS)
                .execute(() -> {
                    VectorizationTaskWrapper wrapper = new VectorizationTaskWrapper(task, task.getPriority());
                    taskQueue.offer(wrapper);
                });

        } else {
            // 达到最大重试次数，标记为失败
            task.setStatus(TaskStatus.FAILED);
            task.setErrorMessage("达到最大重试次数: " + e.getMessage());
            task.setCompletedAt(LocalDateTime.now());
            taskMapper.updateById(task);

            log.error("任务最终失败: taskId={}, retries={}", task.getTaskId(), maxRetries);
        }
    }

    /**
     * 取消任务
     */
    public boolean cancelTask(String taskId) {
        // 标记为取消
        cancelledTasks.put(taskId, true);

        // 如果正在执行，尝试中断
        Future<?> future = runningTasks.get(taskId);
        if (future != null && !future.isDone()) {
            boolean cancelled = future.cancel(true);
            log.info("取消正在执行的任务: taskId={}, cancelled={}", taskId, cancelled);
            return cancelled;
        }

        // 从队列中移除
        taskQueue.removeIf(wrapper -> wrapper.getTask().getTaskId().equals(taskId));

        // 更新状态
        updateTaskStatus(taskId, TaskStatus.CANCELLED);

        log.info("任务已取消: taskId={}", taskId);
        return true;
    }

    /**
     * 推送进度
     */
    private void publishProgress(String taskId, int processed, int total) {
        double progress = (processed * 100.0) / total;

        TaskProgress progressData = TaskProgress.builder()
            .taskId(taskId)
            .processedDocuments(processed)
            .totalDocuments(total)
            .progress(progress)
            .timestamp(LocalDateTime.now())
            .build();

        progressPublisher.publish(taskId, progressData);
    }

    /**
     * 更新任务状态
     */
    private void updateTaskStatus(String taskId, TaskStatus status) {
        VectorizationTaskEntity task = taskMapper.selectOne(
            new LambdaQueryWrapper<VectorizationTaskEntity>()
                .eq(VectorizationTaskEntity::getTaskId, taskId)
        );

        if (task != null) {
            task.setStatus(status);
            if (status == TaskStatus.CANCELLED) {
                task.setCompletedAt(LocalDateTime.now());
            }
            taskMapper.updateById(task);
        }
    }

    /**
     * 获取队列统计信息
     */
    public SchedulerStats getStats() {
        return SchedulerStats.builder()
            .queueSize(taskQueue.size())
            .activeThreads(executor.getActiveCount())
            .completedTasks(executor.getCompletedTaskCount())
            .poolSize(executor.getPoolSize())
            .maxPoolSize(executor.getMaximumPoolSize())
            .build();
    }

    /**
     * 任务包装类（用于优先级队列）
     */
    @Data
    @AllArgsConstructor
    private static class VectorizationTaskWrapper {
        private VectorizationTaskEntity task;
        private Integer priority;
    }
}

2. TaskProgressPublisher - 进度推送器（SSE）

位置: com.pig4cloud.pigx.ai.rag.vector.scheduler.TaskProgressPublisher

java

package com.pig4cloud.pigx.ai.rag.vector.scheduler;

import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 任务进度推送器（SSE）
 *
 * @author pigx
 */
@Slf4j
@Service
public class TaskProgressPublisher {

    /**
     * 存储每个任务的SSE连接
     * key: taskId
     * value: SseEmitter实例
     */
    private final ConcurrentHashMap<String, SseEmitter> emitters = new ConcurrentHashMap<>();

    /**
     * 创建SSE连接
     *
     * @param taskId 任务ID
     * @return SseEmitter
     */
    public SseEmitter createEmitter(String taskId) {
        // 设置超时时间: 30分钟
        SseEmitter emitter = new SseEmitter(30 * 60 * 1000L);

        // 连接完成或超时时移除
        emitter.onCompletion(() -> {
            log.info("SSE连接完成: taskId={}", taskId);
            emitters.remove(taskId);
        });

        emitter.onTimeout(() -> {
            log.warn("SSE连接超时: taskId={}", taskId);
            emitters.remove(taskId);
        });

        emitter.onError((e) -> {
            log.error("SSE连接异常: taskId={}", taskId, e);
            emitters.remove(taskId);
        });

        // 保存连接
        emitters.put(taskId, emitter);

        log.info("SSE连接已创建: taskId={}", taskId);
        return emitter;
    }

    /**
     * 推送进度
     *
     * @param taskId 任务ID
     * @param progress 进度数据
     */
    public void publish(String taskId, TaskProgress progress) {
        SseEmitter emitter = emitters.get(taskId);

        if (emitter != null) {
            try {
                emitter.send(SseEmitter.event()
                    .name("progress")
                    .data(progress)
                );

                log.debug("进度推送成功: taskId={}, progress={}", taskId, progress.getProgress());

            } catch (IOException e) {
                log.error("进度推送失败: taskId={}", taskId, e);
                emitters.remove(taskId);
                emitter.completeWithError(e);
            }
        }
    }

    /**
     * 推送任务完成事件
     *
     * @param taskId 任务ID
     * @param status 最终状态
     */
    public void publishComplete(String taskId, TaskStatus status) {
        SseEmitter emitter = emitters.get(taskId);

        if (emitter != null) {
            try {
                emitter.send(SseEmitter.event()
                    .name("complete")
                    .data(Map.of("taskId", taskId, "status", status))
                );

                emitter.complete();

                log.info("任务完成事件已推送: taskId={}, status={}", taskId, status);

            } catch (IOException e) {
                log.error("完成事件推送失败: taskId={}", taskId, e);
            } finally {
                emitters.remove(taskId);
            }
        }
    }
}

3. Controller API 扩展

位置: com.pig4cloud.pigx.ai.rag.vector.controller.VectorizationController

在现有 Controller 基础上添加 Phase 2 的 API：

java

/**
 * 异步向量化（提交到任务队列）
 */
@PostMapping("/async")
@Operation(summary = "异步向量化", description = "提交任务到队列，立即返回任务ID")
public R<String> asyncVectorize(@RequestBody AsyncVectorizeRequest request) {
    String taskId = vectorizationScheduler.submitAsyncTask(
        request.getDocumentIds(),
        request.getConfig(),
        request.getPriority() != null ? request.getPriority() : 5  // 默认优先级5
    );

    return R.ok(taskId, "任务已提交，请通过SSE接口实时获取进度");
}

/**
 * SSE推送任务进度
 */
@GetMapping(value = "/task/{taskId}/progress", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
@Operation(summary = "实时获取任务进度", description = "使用SSE推送任务执行进度")
public SseEmitter streamTaskProgress(@PathVariable String taskId) {
    return progressPublisher.createEmitter(taskId);
}

/**
 * 取消任务
 */
@PostMapping("/task/{taskId}/cancel")
@Operation(summary = "取消任务", description = "取消正在执行或队列中的任务")
public R<Boolean> cancelTask(@PathVariable String taskId) {
    boolean cancelled = vectorizationScheduler.cancelTask(taskId);
    return R.ok(cancelled);
}

/**
 * 重试失败任务
 */
@PostMapping("/task/{taskId}/retry")
@Operation(summary = "重试失败任务", description = "为失败的任务创建重试任务")
public R<String> retryTask(@PathVariable String taskId) {
    // 查询原任务
    VectorizationTaskEntity originalTask = vectorizationService.getTaskStatus(taskId);

    if (originalTask == null) {
        return R.failed("任务不存在");
    }

    if (originalTask.getStatus() != TaskStatus.FAILED) {
        return R.failed("只能重试失败的任务");
    }

    // 创建新的重试任务
    List<Long> documentIds = parseDocumentIds(originalTask.getDocumentIds());
    VectorizationConfig config = parseConfig(originalTask.getConfig());

    String newTaskId = vectorizationScheduler.submitAsyncTask(
        documentIds,
        config,
        originalTask.getPriority()
    );

    return R.ok(newTaskId, "重试任务已创建");
}

/**
 * 获取调度器统计信息
 */
@GetMapping("/scheduler/stats")
@Operation(summary = "获取调度器统计信息")
public R<SchedulerStats> getSchedulerStats() {
    return R.ok(vectorizationScheduler.getStats());
}

📊 数据库变更

Phase 2 需要在 kb_vectorization_task 表中添加以下字段（如果 Phase 1 未添加）：

sql

ALTER TABLE `kb_vectorization_task`
ADD COLUMN `priority` int DEFAULT 5 COMMENT '优先级（1-10，1最高）' AFTER `processed_chunks`,
ADD COLUMN `retry_count` int DEFAULT 0 COMMENT '重试次数' AFTER `priority`,
ADD COLUMN `max_retries` int DEFAULT 3 COMMENT '最大重试次数' AFTER `retry_count`,
ADD INDEX `idx_priority` (`priority`),
ADD INDEX `idx_status_priority` (`status`, `priority`);

⚙️ 配置

application.yml

yaml

vectorization:
  scheduler:
    # 线程池配置
    thread-pool:
      core-pool-size: 5        # 核心线程数
      max-pool-size: 10        # 最大线程数
      queue-capacity: 100      # 队列容量
      keep-alive-seconds: 60   # 空闲线程存活时间
      thread-name-prefix: "vectorization-worker-"

    # 重试配置
    retry:
      max-attempts: 3          # 最大重试次数
      backoff-delay: 1000      # 初始退避时间（毫秒）
      backoff-multiplier: 2.0  # 退避倍数（指数退避）
      max-backoff-delay: 30000 # 最大退避时间

    # 任务调度配置
    task:
      process-interval: 1000   # 任务处理间隔（毫秒）
      default-priority: 5      # 默认优先级

🚀 部署步骤

1. 数据库升级

bash

# 执行Phase 2的数据库变更脚本
mysql -u root -p < db/vectorization_phase2_alter.sql

2. 代码部署

bash

# 拉取最新代码
git pull origin dev-rag

# 编译打包
cd muotai-backend/pigx-ai
mvn clean package -DskipTests

# 重启服务
./restart-pigx-ai.sh

3. 验证部署

bash

# 检查服务状态
curl http://localhost:6060/actuator/health

# 查看调度器统计
curl http://localhost:6060/ai/vectorization/scheduler/stats

📖 使用示例

1. 提交异步任务

bash

curl -X POST http://localhost:6060/ai/vectorization/async \
  -H "Content-Type: application/json" \
  -d '{
    "documentIds": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10],
    "priority": 1,
    "config": {
      "embeddingModel": "text-embedding-ada-002",
      "chunkSize": 512,
      "batchSize": 10
    }
  }'

响应:

json

{
  "code": 0,
  "msg": "任务已提交，请通过SSE接口实时获取进度",
  "data": "550e8400-e29b-41d4-a716-446655440000"
}

2. 实时获取进度（SSE）

前端代码:

javascript

const taskId = '550e8400-e29b-41d4-a716-446655440000';
const eventSource = new EventSource(`http://localhost:6060/ai/vectorization/task/${taskId}/progress`);

// 监听进度事件
eventSource.addEventListener('progress', (event) => {
  const progress = JSON.parse(event.data);
  console.log(`进度: ${progress.progress}%`);
  console.log(`已处理: ${progress.processedDocuments}/${progress.totalDocuments}`);

  // 更新UI进度条
  updateProgressBar(progress.progress);
});

// 监听完成事件
eventSource.addEventListener('complete', (event) => {
  const result = JSON.parse(event.data);
  console.log(`任务完成: status=${result.status}`);

  eventSource.close(); // 关闭连接
  showCompletionMessage(result.status);
});

// 错误处理
eventSource.onerror = (error) => {
  console.error('SSE连接错误:', error);
  eventSource.close();
};

3. 取消任务

bash

curl -X POST http://localhost:6060/ai/vectorization/task/550e8400-e29b-41d4-a716-446655440000/cancel

4. 重试失败任务

bash

curl -X POST http://localhost:6060/ai/vectorization/task/550e8400-e29b-41d4-a716-446655440000/retry

5. 查看调度器统计

bash

curl http://localhost:6060/ai/vectorization/scheduler/stats

响应:

json

{
  "code": 0,
  "data": {
    "queueSize": 5,
    "activeThreads": 3,
    "completedTasks": 127,
    "poolSize": 5,
    "maxPoolSize": 10
  }
}

🔍 监控和调试

1. 查看任务队列

bash

# 查看待处理任务
curl "http://localhost:6060/ai/vectorization/tasks?status=PENDING&page=1&size=20"

2. 查看运行中任务

bash

# 查看正在执行的任务
curl "http://localhost:6060/ai/vectorization/tasks?status=RUNNING&page=1&size=20"

3. 日志监控

bash

# 查看调度器日志
tail -f logs/pigx-ai.log | grep "VectorizationScheduler"

# 查看任务执行日志
tail -f logs/pigx-ai.log | grep "executeTask"

⚠️ 注意事项

1. 线程池配置

核心线程数: 根据CPU核心数设置，建议 CPU核心数 * 1.5
最大线程数: 不超过 CPU核心数 * 2
队列容量: 根据内存大小调整，建议 100-1000

2. 优先级设置

高优先级 (1-3): 用户实时请求、紧急任务
中优先级 (4-6): 常规批量任务
低优先级 (7-10): 定时同步、后台任务

3. 重试策略

默认最大重试 3 次
指数退避: 1s → 2s → 4s → 8s
网络错误、超时错误适合重试
参数错误、权限错误不应重试

4. SSE连接管理

设置合理的超时时间（建议30分钟）
前端需要处理连接断开重连
任务完成后及时关闭连接

📈 性能优化

1. 批量处理优化

java

// 批量提交任务（共享线程池）
List<String> taskIds = new ArrayList<>();
for (int i = 0; i < 10; i++) {
    String taskId = scheduler.submitAsyncTask(batchDocumentIds, config, priority);
    taskIds.add(taskId);
}

2. 动态调整并发数

java

// 根据系统负载动态调整线程池大小
@Scheduled(fixedRate = 60000) // 每分钟调整一次
public void adjustThreadPool() {
    int queueSize = taskQueue.size();
    int activeThreads = executor.getActiveCount();

    if (queueSize > 50 && activeThreads < executor.getMaximumPoolSize()) {
        // 队列积压，增加线程
        executor.setCorePoolSize(Math.min(executor.getCorePoolSize() + 1, 10));
    } else if (queueSize < 10 && activeThreads > 5) {
        // 队列空闲，减少线程
        executor.setCorePoolSize(Math.max(executor.getCorePoolSize() - 1, 5));
    }
}

🐛 故障排查

1. 任务一直在PENDING状态

可能原因:

线程池已满
任务优先级过低
调度器未启动

解决方法:

bash

# 检查调度器状态
curl http://localhost:6060/ai/vectorization/scheduler/stats

# 调整任务优先级
# 检查线程池配置

2. SSE连接频繁断开

可能原因:

网络不稳定
超时时间过短
Nginx/代理超时配置

解决方法:

nginx

# Nginx配置
location /ai/vectorization/task/ {
    proxy_read_timeout 1800s;  # 30分钟
    proxy_buffering off;
}

3. 任务重试次数过多

可能原因:

外部依赖不稳定（Milvus、MinIO）
配置错误
资源不足

解决方法:

检查外部依赖健康状态
调整重试策略
增加资源配额

🔄 下一步：阶段三

Phase 3 将实现：

✨ 增量更新检测
✨ 定时增量同步
✨ 文档变更监听
✨ 智能增量策略

最后更新: 2025-12-12 版本: Phase 2 v1.0.0

向量化系统阶段二实现指南——异步任务调度 ​

📋 概述 ​

🎯 实现目标 ​

核心功能 ​

性能目标 ​

🏗️ 架构设计 ​

整体架构 ​

📦 核心组件实现 ​

1. VectorizationScheduler - 任务调度器 ​

2. TaskProgressPublisher - 进度推送器（SSE） ​

3. Controller API 扩展 ​

📊 数据库变更 ​

⚙️ 配置 ​

application.yml ​

🚀 部署步骤 ​

1. 数据库升级 ​

2. 代码部署 ​

3. 验证部署 ​

📖 使用示例 ​

1. 提交异步任务 ​

2. 实时获取进度（SSE） ​

3. 取消任务 ​

4. 重试失败任务 ​

5. 查看调度器统计 ​

🔍 监控和调试 ​

1. 查看任务队列 ​

2. 查看运行中任务 ​

3. 日志监控 ​

⚠️ 注意事项 ​

1. 线程池配置 ​

2. 优先级设置 ​

3. 重试策略 ​

4. SSE连接管理 ​

📈 性能优化 ​

1. 批量处理优化 ​

2. 动态调整并发数 ​

🐛 故障排查 ​

1. 任务一直在PENDING状态 ​

2. SSE连接频繁断开 ​

3. 任务重试次数过多 ​

🔄 下一步：阶段三 ​

向量化系统阶段二实现指南——异步任务调度

📋 概述

🎯 实现目标

核心功能

性能目标

🏗️ 架构设计

整体架构

📦 核心组件实现

1. VectorizationScheduler - 任务调度器

2. TaskProgressPublisher - 进度推送器（SSE）

3. Controller API 扩展

📊 数据库变更

⚙️ 配置

application.yml

🚀 部署步骤

1. 数据库升级

2. 代码部署

3. 验证部署

📖 使用示例

1. 提交异步任务

2. 实时获取进度（SSE）

3. 取消任务

4. 重试失败任务

5. 查看调度器统计

🔍 监控和调试

1. 查看任务队列

2. 查看运行中任务

3. 日志监控

⚠️ 注意事项

1. 线程池配置

2. 优先级设置

3. 重试策略

4. SSE连接管理

📈 性能优化

1. 批量处理优化

2. 动态调整并发数

🐛 故障排查

1. 任务一直在PENDING状态

2. SSE连接频繁断开

3. 任务重试次数过多

🔄 下一步：阶段三