从需求到上线：异步任务框架实战指南

引言

前三篇文章我们深入探讨了异步任务框架的架构设计、租约机制和 Worker 实现。本文将通过一个真实的医疗系统需求，完整演示如何接入异步任务框架，从需求分析、设计方案、代码实现到上线部署的全流程。

业务需求

场景描述

某三甲医院需要从卫健委平台批量导入患者的医保信息。系统要求：

  • 数据量：单次导入 5,000 ~ 50,000 条患者记录
  • 耗时：每条记录需要调用医保接口验证，平均耗时 200ms
  • 体验要求：用户提交后立即返回，后台异步处理，支持查看进度
  • 容错要求：部分记录失败不影响整体导入，生成失败报告
  • 审计要求：记录操作人、操作时间、导入结果
  • 重试机制：网络异常时自动重试，人工失败支持手动重试

需求分析

为什么需要异步任务？

同步处理异步处理
50,000 条 × 200ms = 2.7 小时立即返回，后台处理
HTTP 请求超时无超时问题
服务器宕机任务丢失持久化到数据库
无法查看进度实时进度反馈
无法重试支持自动/手动重试

异步任务框架如何解决？

用户操作             框架处理                 医保接口
   │                    │                        │
   │──提交导入──→       │                        │
   │                    │──持久化任务──→         │
   │←─返回任务ID─       │    （数据库）          │
   │                    │                        │
   │                    │──Worker领取──→         │
   │                    │                        │
   │                    │──────调用验证──────→   │
   │──查询进度──→       │                        │
   │←─50%完成───        │                        │
   │                    │←─────返回结果─────     │
   │                    │                        │
   │                    │──上报进度──→           │
   │──查询进度──→       │                        │
   │←─100%完成──        │                        │
   │──下载报告──→       │                        │
   │←─返回文件──        │                        │

设计方案

1. 任务类型定义

/**
 * 医保信息导入任务
 * 
 * 任务类型：MEDICAL_INSURANCE.IMPORT
 * 所属模块：患者管理
 * 执行时长：预计 10-60 分钟
 * 并发限制：单类型最多 2 个并发（避免打满医保接口）
 */
public static final String TASK_TYPE = "MEDICAL_INSURANCE.IMPORT";

2. 数据模型

Payload（任务参数）

/**
 * 医保导入任务参数
 */
@Data
public class MedicalInsuranceImportPayload {
    
    /**
     * Payload 版本（支持升级兼容）
     */
    private String schemaVersion = "1.0";
    
    /**
     * 导入文件 ID（OSS 文件引用）
     */
    private String fileId;
    
    /**
     * 导入文件名称
     */
    private String fileName;
    
    /**
     * 导入批次号（业务唯一键）
     */
    private String batchNo;
    
    /**
     * 导入配置
     */
    private ImportConfig config;
    
    @Data
    public static class ImportConfig {
        /**
         * 是否跳过已存在的患者
         */
        private boolean skipExisting = true;
        
        /**
         * 是否更新已存在的患者信息
         */
        private boolean updateExisting = false;
        
        /**
         * 每批次处理记录数
         */
        private int batchSize = 100;
    }
}

导入结果

/**
 * 医保导入结果
 */
@Data
public class MedicalInsuranceImportResult {
    /**
     * 总记录数
     */
    private long totalCount;
    
    /**
     * 成功导入数
     */
    private long successCount;
    
    /**
     * 失败记录数
     */
    private long failedCount;
    
    /**
     * 跳过记录数
     */
    private long skippedCount;
    
    /**
     * 失败报告文件 ID
     */
    private String errorReportFileId;
}

3. 数据库设计

导入记录表

CREATE TABLE medical_insurance_import_record (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    batch_no VARCHAR(64) NOT NULL COMMENT '批次号',
    task_id VARCHAR(64) NOT NULL COMMENT '异步任务ID',
    
    file_id VARCHAR(128) NOT NULL COMMENT '源文件ID',
    file_name VARCHAR(255) NOT NULL COMMENT '源文件名',
    
    total_count BIGINT NOT NULL DEFAULT 0 COMMENT '总记录数',
    success_count BIGINT NOT NULL DEFAULT 0 COMMENT '成功数',
    failed_count BIGINT NOT NULL DEFAULT 0 COMMENT '失败数',
    skipped_count BIGINT NOT NULL DEFAULT 0 COMMENT '跳过数',
    
    error_report_file_id VARCHAR(128) COMMENT '失败报告文件ID',
    
    status VARCHAR(20) NOT NULL COMMENT '状态',
    
    create_by VARCHAR(64) NOT NULL,
    create_time DATETIME NOT NULL,
    update_time DATETIME NOT NULL,
    
    UNIQUE KEY uk_batch_no (batch_no),
    KEY idx_task_id (task_id),
    KEY idx_create_by_time (create_by, create_time)
) COMMENT='医保信息导入记录';

导入明细表（失败记录）

CREATE TABLE medical_insurance_import_detail (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    batch_no VARCHAR(64) NOT NULL COMMENT '批次号',
    
    row_number INT NOT NULL COMMENT '行号',
    patient_name VARCHAR(128) COMMENT '患者姓名',
    id_card VARCHAR(32) COMMENT '身份证号',
    insurance_no VARCHAR(64) COMMENT '医保卡号',
    
    error_code VARCHAR(32) COMMENT '错误码',
    error_message TEXT COMMENT '错误信息',
    
    create_time DATETIME NOT NULL,
    
    KEY idx_batch_no (batch_no),
    KEY idx_row_number (batch_no, row_number)
) COMMENT='医保信息导入明细';

代码实现

1. Handler 实现

package com.tudicloud.medical.patient.handler;

import com.tudicloud.framework.asynctask.core.model.AsyncTaskProgress;
import com.tudicloud.framework.asynctask.core.model.TaskExecutionResult;
import com.tudicloud.framework.asynctask.core.spi.AsyncTaskHandler;
import com.tudicloud.framework.asynctask.core.spi.TaskExecutionContext;
import com.tudicloud.medical.patient.model.MedicalInsuranceImportPayload;
import com.tudicloud.medical.patient.service.MedicalInsuranceImportService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

/**
 * 医保信息导入任务处理器
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class MedicalInsuranceImportHandler 
    implements AsyncTaskHandler<MedicalInsuranceImportPayload> {
    
    private final MedicalInsuranceImportService importService;
    private final FileStorageService fileStorageService;
    private final MedicalInsuranceApiClient insuranceApiClient;
    
    @Override
    public String taskType() {
        return "MEDICAL_INSURANCE.IMPORT";
    }
    
    @Override
    public Class<MedicalInsuranceImportPayload> payloadType() {
        return MedicalInsuranceImportPayload.class;
    }
    
    @Override
    public void validate(MedicalInsuranceImportPayload payload) {
        // 快速校验（无副作用）
        if (payload.getFileId() == null) {
            throw new IllegalArgumentException("文件ID不能为空");
        }
        if (payload.getBatchNo() == null) {
            throw new IllegalArgumentException("批次号不能为空");
        }
    }
    
    @Override
    public TaskExecutionResult execute(
        TaskExecutionContext context,
        MedicalInsuranceImportPayload payload
    ) throws Exception {
        
        String taskId = context.taskId();
        String batchNo = payload.getBatchNo();
        
        log.info("开始执行医保导入任务, taskId={}, batchNo={}, fileId={}", 
            taskId, batchNo, payload.getFileId());
        
        // 1. 初始化导入记录
        importService.initImportRecord(batchNo, taskId, payload);
        
        // 2. 下载并解析文件
        List<PatientRow> rows = parseFile(payload.getFileId());
        long totalCount = rows.size();
        
        log.info("文件解析完成, taskId={}, totalCount={}", taskId, totalCount);
        
        // 3. 分批处理
        int batchSize = payload.getConfig().getBatchSize();
        long processedCount = 0;
        long successCount = 0;
        long failedCount = 0;
        long skippedCount = 0;
        
        List<ImportDetail> failedDetails = new ArrayList<>();
        
        for (int i = 0; i < rows.size(); i += batchSize) {
            // 检查取消标志
            context.checkCancellation();
            
            // 获取当前批次
            int endIndex = Math.min(i + batchSize, rows.size());
            List<PatientRow> batch = rows.subList(i, endIndex);
            
            // 处理批次（幂等）
            BatchResult result = processBatchIdempotently(
                taskId, batchNo, batch, payload.getConfig()
            );
            
            successCount += result.getSuccessCount();
            failedCount += result.getFailedCount();
            skippedCount += result.getSkippedCount();
            failedDetails.addAll(result.getFailedDetails());
            
            processedCount = successCount + failedCount + skippedCount;
            
            // 上报进度
            context.reportProgress(new AsyncTaskProgress(
                totalCount,
                processedCount,
                successCount,
                failedCount,
                skippedCount
            ));
            
            log.info("批次处理完成, taskId={}, batch={}/{}, progress={}/{}", 
                taskId, (i / batchSize + 1), 
                (rows.size() + batchSize - 1) / batchSize,
                processedCount, totalCount);
        }
        
        // 4. 生成失败报告
        String errorReportFileId = null;
        if (!failedDetails.isEmpty()) {
            errorReportFileId = generateErrorReport(batchNo, failedDetails);
            log.info("失败报告已生成, taskId={}, fileId={}, failedCount={}", 
                taskId, errorReportFileId, failedCount);
        }
        
        // 5. 更新导入记录
        importService.completeImportRecord(
            batchNo,
            totalCount,
            successCount,
            failedCount,
            skippedCount,
            errorReportFileId
        );
        
        log.info("医保导入任务完成, taskId={}, success={}, failed={}, skipped={}", 
            taskId, successCount, failedCount, skippedCount);
        
        // 6. 返回结果
        if (failedCount == 0) {
            return TaskExecutionResult.success(errorReportFileId);
        } else {
            return TaskExecutionResult.partialSuccess(errorReportFileId);
        }
    }
    
    @Override
    public boolean supportsManualRetry() {
        // 支持手动重新排队
        return true;
    }
    
    /**
     * 幂等处理批次
     */
    private BatchResult processBatchIdempotently(
        String taskId,
        String batchNo,
        List<PatientRow> batch,
        ImportConfig config
    ) {
        // 使用任务ID+批次起始行号作为幂等键
        int startRowNumber = batch.get(0).getRowNumber();
        String idempotencyKey = taskId + "_" + startRowNumber;
        
        // 检查是否已处理
        if (importService.isBatchProcessed(idempotencyKey)) {
            log.info("批次已处理，跳过, idempotencyKey={}", idempotencyKey);
            return importService.getBatchResult(idempotencyKey);
        }
        
        // 处理批次
        BatchResult result = new BatchResult();
        
        for (PatientRow row : batch) {
            try {
                // 检查是否跳过
                if (config.isSkipExisting() && 
                    importService.patientExists(row.getIdCard())) {
                    result.addSkipped(row);
                    continue;
                }
                
                // 调用医保接口验证
                InsuranceInfo info = insuranceApiClient.queryPatientInfo(
                    row.getIdCard(),
                    row.getInsuranceNo()
                );
                
                // 保存到数据库
                importService.savePatientInsurance(row, info);
                result.addSuccess(row);
                
            } catch (InsuranceApiException ex) {
                // 医保接口异常，记录失败明细
                log.warn("医保接口调用失败, row={}, error={}", 
                    row.getRowNumber(), ex.getErrorCode());
                result.addFailed(row, ex.getErrorCode(), ex.getMessage());
                
            } catch (Exception ex) {
                // 其他异常
                log.error("处理患者记录失败, row={}", row.getRowNumber(), ex);
                result.addFailed(row, "SYSTEM_ERROR", ex.getMessage());
            }
        }
        
        // 记录批次处理结果（幂等标记）
        importService.saveBatchResult(idempotencyKey, result);
        
        return result;
    }
    
    /**
     * 解析导入文件
     */
    private List<PatientRow> parseFile(String fileId) throws IOException {
        InputStream inputStream = fileStorageService.download(fileId);
        
        try (ExcelReader reader = ExcelUtil.getReader(inputStream)) {
            List<Map<String, Object>> rows = reader.readAll();
            
            List<PatientRow> result = new ArrayList<>();
            for (int i = 0; i < rows.size(); i++) {
                Map<String, Object> row = rows.get(i);
                result.add(PatientRow.fromMap(i + 2, row));  // +2 因为表头占1行
            }
            
            return result;
        }
    }
    
    /**
     * 生成失败报告
     */
    private String generateErrorReport(
        String batchNo,
        List<ImportDetail> failedDetails
    ) throws IOException {
        // 生成 Excel 文件
        ExcelWriter writer = ExcelUtil.getWriter();
        writer.addHeaderAlias("rowNumber", "行号");
        writer.addHeaderAlias("patientName", "患者姓名");
        writer.addHeaderAlias("idCard", "身份证号");
        writer.addHeaderAlias("insuranceNo", "医保卡号");
        writer.addHeaderAlias("errorCode", "错误码");
        writer.addHeaderAlias("errorMessage", "错误信息");
        
        writer.write(failedDetails, true);
        
        byte[] content = writer.flush().toByteArray();
        writer.close();
        
        // 上传到 OSS
        String fileName = String.format(
            "medical_insurance_import_error_%s_%s.xlsx",
            batchNo,
            LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyyMMddHHmmss"))
        );
        
        return fileStorageService.upload(fileName, content);
    }
}

2. Controller 实现

package com.tudicloud.medical.patient.controller;

import com.tudicloud.framework.asynctask.core.model.*;
import com.tudicloud.framework.asynctask.core.service.AsyncTaskQueryService;
import com.tudicloud.framework.asynctask.core.service.AsyncTaskService;
import com.tudicloud.medical.common.model.BaseResult;
import com.tudicloud.medical.patient.model.MedicalInsuranceImportPayload;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.tags.Tag;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.security.access.prepost.PreAuthorize;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.multipart.MultipartFile;

import java.util.UUID;

/**
 * 医保信息导入 Controller
 */
@Slf4j
@RestController
@RequestMapping("/api/v1/medical-insurance/import")
@RequiredArgsConstructor
@Tag(name = "医保信息导入")
public class MedicalInsuranceImportController {
    
    private final AsyncTaskService asyncTaskService;
    private final AsyncTaskQueryService asyncTaskQueryService;
    private final FileStorageService fileStorageService;
    private final SecurityHelper securityHelper;
    
    /**
     * 提交导入任务
     */
    @PostMapping("/submit")
    @PreAuthorize("hasAuthority('patient:insurance:import')")
    @Operation(summary = "提交医保信息导入任务")
    public BaseResult<ImportTaskView> submitImportTask(
        @RequestParam("file") MultipartFile file,
        @RequestParam(value = "skipExisting", defaultValue = "true") boolean skipExisting,
        @RequestParam(value = "updateExisting", defaultValue = "false") boolean updateExisting
    ) {
        // 1. 权限校验（已通过 @PreAuthorize）
        
        // 2. 文件校验
        validateFile(file);
        
        // 3. 上传文件到 OSS
        String fileId = fileStorageService.upload(
            file.getOriginalFilename(),
            file.getInputStream()
        );
        
        // 4. 生成批次号
        String batchNo = generateBatchNo();
        
        // 5. 构造任务参数
        MedicalInsuranceImportPayload payload = new MedicalInsuranceImportPayload();
        payload.setFileId(fileId);
        payload.setFileName(file.getOriginalFilename());
        payload.setBatchNo(batchNo);
        
        MedicalInsuranceImportPayload.ImportConfig config = 
            new MedicalInsuranceImportPayload.ImportConfig();
        config.setSkipExisting(skipExisting);
        config.setUpdateExisting(updateExisting);
        payload.setConfig(config);
        
        // 6. 提交异步任务
        AsyncTaskSubmission submission = new AsyncTaskSubmission(
            "MEDICAL_INSURANCE.IMPORT",
            payload,
            securityHelper.getCurrentUserId(),  // ownerId
            batchNo,                             // idempotencyKey（批次号唯一）
            3                                    // maxAttempts
        );
        
        AsyncTask task = asyncTaskService.submit(submission);
        
        log.info("医保导入任务已提交, taskId={}, batchNo={}, userId={}, fileName={}", 
            task.id(), batchNo, securityHelper.getCurrentUserId(), 
            file.getOriginalFilename());
        
        // 7. 返回任务信息
        return BaseResult.success(toImportTaskView(task));
    }
    
    /**
     * 查询任务状态
     */
    @GetMapping("/{taskId}")
    @PreAuthorize("hasAuthority('patient:insurance:import')")
    @Operation(summary = "查询导入任务状态")
    public BaseResult<ImportTaskView> getTask(@PathVariable String taskId) {
        // 查询任务（带 owner 数据范围）
        AsyncTask task = asyncTaskQueryService.getTask(
            taskId,
            AsyncTaskQueryScope.owner(securityHelper.getCurrentUserId())
        );
        
        if (task == null) {
            return BaseResult.error("任务不存在或无权访问");
        }
        
        return BaseResult.success(toImportTaskView(task));
    }
    
    /**
     * 查询任务列表
     */
    @GetMapping("/list")
    @PreAuthorize("hasAuthority('patient:insurance:import')")
    @Operation(summary = "查询导入任务列表")
    public BaseResult<AsyncTaskPage<ImportTaskView>> listTasks(
        @RequestParam(defaultValue = "1") int pageNumber,
        @RequestParam(defaultValue = "20") int pageSize
    ) {
        // 构造查询条件
        AsyncTaskPageQuery query = AsyncTaskPageQuery.builder()
            .taskType("MEDICAL_INSURANCE.IMPORT")
            .pageNumber(pageNumber)
            .pageSize(pageSize)
            .build();
        
        // 查询任务列表（带 owner 数据范围）
        AsyncTaskPage<AsyncTask> page = asyncTaskQueryService.pageTasks(
            query,
            AsyncTaskQueryScope.owner(securityHelper.getCurrentUserId())
        );
        
        // 转换为视图对象
        AsyncTaskPage<ImportTaskView> viewPage = page.map(this::toImportTaskView);
        
        return BaseResult.success(viewPage);
    }
    
    /**
     * 取消任务
     */
    @PostMapping("/{taskId}/cancel")
    @PreAuthorize("hasAuthority('patient:insurance:import')")
    @Operation(summary = "取消导入任务")
    public BaseResult<Void> cancelTask(@PathVariable String taskId) {
        asyncTaskService.requestCancellation(
            taskId,
            AsyncTaskQueryScope.owner(securityHelper.getCurrentUserId())
        );
        
        log.info("医保导入任务已取消, taskId={}, userId={}", 
            taskId, securityHelper.getCurrentUserId());
        
        return BaseResult.success();
    }
    
    /**
     * 手动重试
     */
    @PostMapping("/{taskId}/retry")
    @PreAuthorize("hasAuthority('patient:insurance:import')")
    @Operation(summary = "手动重试导入任务")
    public BaseResult<ImportTaskView> retryTask(@PathVariable String taskId) {
        // 生成新的幂等键
        String newIdempotencyKey = UUID.randomUUID().toString();
        
        AsyncTask newTask = asyncTaskService.requeue(
            taskId,
            AsyncTaskQueryScope.owner(securityHelper.getCurrentUserId()),
            newIdempotencyKey
        );
        
        log.info("医保导入任务已重新排队, oldTaskId={}, newTaskId={}, userId={}", 
            taskId, newTask.id(), securityHelper.getCurrentUserId());
        
        return BaseResult.success(toImportTaskView(newTask));
    }
    
    /**
     * 下载失败报告
     */
    @GetMapping("/{taskId}/error-report")
    @PreAuthorize("hasAuthority('patient:insurance:import')")
    @Operation(summary = "下载失败报告")
    public void downloadErrorReport(
        @PathVariable String taskId,
        HttpServletResponse response
    ) throws IOException {
        // 查询任务（重新鉴权）
        AsyncTask task = asyncTaskQueryService.getTask(
            taskId,
            AsyncTaskQueryScope.owner(securityHelper.getCurrentUserId())
        );
        
        if (task == null) {
            throw new BusinessException("任务不存在或无权访问");
        }
        
        String artifactId = task.artifactId();
        if (artifactId == null) {
            throw new BusinessException("任务暂无失败报告");
        }
        
        // 下载文件
        FileMetadata metadata = fileStorageService.getMetadata(artifactId);
        InputStream inputStream = fileStorageService.download(artifactId);
        
        // 设置响应头
        response.setContentType("application/vnd.openxmlformats-officedocument.spreadsheetml.sheet");
        response.setHeader("Content-Disposition", 
            "attachment; filename=" + URLEncoder.encode(metadata.getFileName(), "UTF-8"));
        
        // 写入响应
        IOUtils.copy(inputStream, response.getOutputStream());
        response.flushBuffer();
    }
    
    /**
     * 文件校验
     */
    private void validateFile(MultipartFile file) {
        if (file.isEmpty()) {
            throw new IllegalArgumentException("上传文件不能为空");
        }
        
        String fileName = file.getOriginalFilename();
        if (fileName == null || !fileName.endsWith(".xlsx")) {
            throw new IllegalArgumentException("只支持 .xlsx 格式文件");
        }
        
        // 限制文件大小：10MB
        if (file.getSize() > 10 * 1024 * 1024) {
            throw new IllegalArgumentException("文件大小不能超过 10MB");
        }
    }
    
    /**
     * 生成批次号
     */
    private String generateBatchNo() {
        return "INSURANCE_" + 
            LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyyMMddHHmmss")) +
            "_" + UUID.randomUUID().toString().substring(0, 8);
    }
    
    /**
     * 转换为视图对象
     */
    private ImportTaskView toImportTaskView(AsyncTask task) {
        ImportTaskView view = new ImportTaskView();
        view.setTaskId(task.id());
        view.setStatus(task.status().name());
        view.setBatchNo(task.idempotencyKey());
        
        // 进度信息
        if (task.progress() != null) {
            view.setTotalCount(task.progress().totalCount());
            view.setProcessedCount(task.progress().processedCount());
            view.setSuccessCount(task.progress().successCount());
            view.setFailedCount(task.progress().failedCount());
            view.setSkippedCount(task.progress().skippedCount());
            
            // 计算百分比
            if (task.progress().totalCount() > 0) {
                view.setProgressPercent(
                    (int) (task.progress().processedCount() * 100 / task.progress().totalCount())
                );
            }
        }
        
        // 时间信息
        view.setCreatedAt(task.createdAt());
        view.setFinishedAt(task.finishedAt());
        
        // 结果文件
        view.setErrorReportFileId(task.artifactId());
        
        return view;
    }
}

3. 前端页面（Vue 3）

<template>
  <div class="import-page">
    <!-- 上传区域 -->
    <el-card class="upload-card">
      <template #header>
        <span>医保信息导入</span>
      </template>
      
      <el-upload
        ref="uploadRef"
        :auto-upload="false"
        :on-change="handleFileChange"
        :limit="1"
        accept=".xlsx"
        drag
      >
        <el-icon class="el-icon--upload"><upload-filled /></el-icon>
        <div class="el-upload__text">
          将文件拖到此处，或<em>点击上传</em>
        </div>
        <template #tip>
          <div class="el-upload__tip">
            只能上传 .xlsx 文件，且不超过 10MB
          </div>
        </template>
      </el-upload>
      
      <el-form :model="form" label-width="120px" style="margin-top: 20px;">
        <el-form-item label="导入策略">
          <el-checkbox v-model="form.skipExisting">
            跳过已存在的患者
          </el-checkbox>
          <el-checkbox v-model="form.updateExisting">
            更新已存在的患者信息
          </el-checkbox>
        </el-form-item>
      </el-form>
      
      <el-button 
        type="primary" 
        :loading="uploading"
        @click="submitImport"
      >
        开始导入
      </el-button>
    </el-card>
    
    <!-- 任务列表 -->
    <el-card class="task-list-card">
      <template #header>
        <span>导入任务列表</span>
        <el-button 
          type="text" 
          @click="refreshTaskList"
          style="float: right;"
        >
          刷新
        </el-button>
      </template>
      
      <el-table :data="taskList" style="width: 100%">
        <el-table-column prop="batchNo" label="批次号" width="200" />
        
        <el-table-column label="状态" width="120">
          <template #default="{ row }">
            <el-tag :type="getStatusType(row.status)">
              {{ getStatusText(row.status) }}
            </el-tag>
          </template>
        </el-table-column>
        
        <el-table-column label="进度" width="300">
          <template #default="{ row }">
            <div v-if="row.status === 'RUNNING' || row.status === 'QUEUED'">
              <el-progress 
                :percentage="row.progressPercent" 
                :status="getProgressStatus(row.status)"
              />
              <div class="progress-text">
                {{ row.processedCount }} / {{ row.totalCount }}
                (成功: {{ row.successCount }}, 
                 失败: {{ row.failedCount }}, 
                 跳过: {{ row.skippedCount }})
              </div>
            </div>
            <div v-else>
              成功: {{ row.successCount }}, 
              失败: {{ row.failedCount }}, 
              跳过: {{ row.skippedCount }}
            </div>
          </template>
        </el-table-column>
        
        <el-table-column label="创建时间" width="180">
          <template #default="{ row }">
            {{ formatTime(row.createdAt) }}
          </template>
        </el-table-column>
        
        <el-table-column label="操作" width="250">
          <template #default="{ row }">
            <el-button 
              v-if="row.status === 'RUNNING' || row.status === 'QUEUED'"
              type="text" 
              size="small"
              @click="cancelTask(row.taskId)"
            >
              取消
            </el-button>
            
            <el-button 
              v-if="row.status === 'FAILED'"
              type="text" 
              size="small"
              @click="retryTask(row.taskId)"
            >
              重试
            </el-button>
            
            <el-button 
              v-if="row.errorReportFileId"
              type="text" 
              size="small"
              @click="downloadErrorReport(row.taskId)"
            >
              下载失败报告
            </el-button>
          </template>
        </el-table-column>
      </el-table>
      
      <el-pagination
        v-model:current-page="pagination.pageNumber"
        v-model:page-size="pagination.pageSize"
        :total="pagination.total"
        layout="total, prev, pager, next"
        @current-change="handlePageChange"
      />
    </el-card>
  </div>
</template>

<script setup>
import { ref, onMounted, onUnmounted } from 'vue'
import { ElMessage } from 'element-plus'
import * as api from '@/api/medical-insurance'

const uploadRef = ref()
const uploading = ref(false)
const form = ref({
  skipExisting: true,
  updateExisting: false
})

const taskList = ref([])
const pagination = ref({
  pageNumber: 1,
  pageSize: 20,
  total: 0
})

let pollingTimer = null

// 提交导入
const submitImport = async () => {
  const files = uploadRef.value.uploadFiles
  if (files.length === 0) {
    ElMessage.warning('请先选择文件')
    return
  }
  
  uploading.value = true
  
  try {
    const formData = new FormData()
    formData.append('file', files[0].raw)
    formData.append('skipExisting', form.value.skipExisting)
    formData.append('updateExisting', form.value.updateExisting)
    
    await api.submitImportTask(formData)
    
    ElMessage.success('导入任务已提交')
    uploadRef.value.clearFiles()
    refreshTaskList()
  } catch (error) {
    ElMessage.error('提交失败: ' + error.message)
  } finally {
    uploading.value = false
  }
}

// 刷新任务列表
const refreshTaskList = async () => {
  try {
    const res = await api.listTasks({
      pageNumber: pagination.value.pageNumber,
      pageSize: pagination.value.pageSize
    })
    
    taskList.value = res.data.items
    pagination.value.total = res.data.total
  } catch (error) {
    console.error('刷新任务列表失败', error)
  }
}

// 取消任务
const cancelTask = async (taskId) => {
  try {
    await api.cancelTask(taskId)
    ElMessage.success('任务已取消')
    refreshTaskList()
  } catch (error) {
    ElMessage.error('取消失败: ' + error.message)
  }
}

// 重试任务
const retryTask = async (taskId) => {
  try {
    await api.retryTask(taskId)
    ElMessage.success('任务已重新提交')
    refreshTaskList()
  } catch (error) {
    ElMessage.error('重试失败: ' + error.message)
  }
}

// 下载失败报告
const downloadErrorReport = (taskId) => {
  window.open(`/api/v1/medical-insurance/import/${taskId}/error-report`)
}

// 状态映射
const getStatusType = (status) => {
  const map = {
    QUEUED: 'info',
    RUNNING: 'primary',
    SUCCESS: 'success',
    PARTIAL_SUCCESS: 'warning',
    FAILED: 'danger',
    CANCELLED: 'info'
  }
  return map[status] || 'info'
}

const getStatusText = (status) => {
  const map = {
    QUEUED: '排队中',
    RUNNING: '执行中',
    SUCCESS: '成功',
    PARTIAL_SUCCESS: '部分成功',
    FAILED: '失败',
    CANCELLED: '已取消'
  }
  return map[status] || status
}

// 轮询刷新（有运行中的任务时）
const startPolling = () => {
  pollingTimer = setInterval(() => {
    const hasRunning = taskList.value.some(
      task => task.status === 'RUNNING' || task.status === 'QUEUED'
    )
    
    if (hasRunning) {
      refreshTaskList()
    }
  }, 3000)  // 每 3 秒刷新一次
}

const stopPolling = () => {
  if (pollingTimer) {
    clearInterval(pollingTimer)
    pollingTimer = null
  }
}

onMounted(() => {
  refreshTaskList()
  startPolling()
})

onUnmounted(() => {
  stopPolling()
})
</script>

配置调优

1. 应用配置

tudicloud:
  async-task:
    enabled: true
    worker:
      enabled: true
      local-max-concurrency: 4
      poll-interval: 1s
      heartbeat-interval: 10s
      lease-timeout: 60s
      recovery-interval: 60s
      max-execution-time: 1h        # 单任务最长 1 小时
      slow-task-threshold: 10m      # 10 分钟以上算慢任务
    
    limits:
      default-global-concurrency: 10
      default-type-concurrency: 2    # 医保导入限制 2 并发
      default-max-attempts: 3
      max-payload-bytes: 65536

2. 动态策略配置

// 初始化医保导入任务的并发策略
@Bean
public CommandLineRunner initMedicalInsurancePolicy(
    AsyncTaskPolicyService policyService
) {
    return args -> {
        policyService.save(new AsyncTaskPolicy(
            AsyncTaskPolicy.typePolicyKey("MEDICAL_INSURANCE.IMPORT"),
            "MEDICAL_INSURANCE.IMPORT",
            2,                          // 最多 2 个并发
            3,                          // 最多重试 3 次
            Duration.ofMinutes(5),      // 重试延迟 5 分钟
            true
        ));
    };
}

3. 数据库索引优化

-- 医保导入相关索引
CREATE INDEX idx_insurance_import_status 
ON system_async_task (status, task_type, create_time)
WHERE task_type = 'MEDICAL_INSURANCE.IMPORT';

CREATE INDEX idx_insurance_import_owner 
ON system_async_task (owner_id, create_time DESC)
WHERE task_type = 'MEDICAL_INSURANCE.IMPORT';

上线检查清单

1. 功能验证

  • [ ] 提交导入任务成功
  • [ ] 任务持久化到数据库
  • [ ] Worker 成功领取任务
  • [ ] 进度实时更新
  • [ ] 失败报告正确生成
  • [ ] 取消功能正常
  • [ ] 手动重试功能正常
  • [ ] 幂等性验证（重复执行不产生副作用）

2. 性能验证

  • [ ] 5,000 条记录导入时长 < 30 分钟
  • [ ] 50,000 条记录导入时长 < 3 小时
  • [ ] 并发 2 个导入任务互不影响
  • [ ] 进度刷新频率合理（不过于频繁）

3. 容错验证

  • [ ] Worker 宕机后任务自动恢复
  • [ ] 数据库连接中断后自动重连
  • [ ] 医保接口超时自动重试
  • [ ] 租约续期失败时任务被其他 Worker 接管

4. 权限验证

  • [ ] 普通用户只能查看自己的任务
  • [ ] 管理员可以查看所有任务
  • [ ] 失败报告下载需要重新鉴权
  • [ ] 跨用户访问被正确拦截

监控与告警

1. 业务监控

// 自定义监控指标
@Component
public class MedicalInsuranceImportMetrics {
    
    private final MeterRegistry meterRegistry;
    
    // 导入成功率
    public void recordImportSuccess(long successCount, long totalCount) {
        meterRegistry.counter(
            "medical.insurance.import.success_rate",
            "result", "success"
        ).increment(successCount);
        
        meterRegistry.counter(
            "medical.insurance.import.success_rate",
            "result", "total"
        ).increment(totalCount);
    }
    
    // 医保接口调用延迟
    public void recordApiLatency(Duration latency) {
        meterRegistry.timer(
            "medical.insurance.api.latency"
        ).record(latency);
    }
}

2. Grafana 看板

# 导入任务成功率
sum(rate(medical_insurance_import_success_rate{result="success"}[5m])) /
sum(rate(medical_insurance_import_success_rate{result="total"}[5m]))

# 正在执行的导入任务数
async_task_active_count{task_type="MEDICAL_INSURANCE.IMPORT"}

# 导入任务平均执行时长
histogram_quantile(0.5, 
  rate(async_task_execution_seconds_bucket{task_type="MEDICAL_INSURANCE.IMPORT"}[5m])
)

3. 告警规则

# 导入成功率过低
- alert: MedicalInsuranceImportLowSuccessRate
  expr: |
    sum(rate(medical_insurance_import_success_rate{result="success"}[10m])) /
    sum(rate(medical_insurance_import_success_rate{result="total"}[10m])) < 0.8
  for: 5m
  annotations:
    summary: "医保导入成功率低于 80%"
    
# 导入任务堆积
- alert: MedicalInsuranceImportTaskBacklog
  expr: |
    count(async_task_status{task_type="MEDICAL_INSURANCE.IMPORT", status="QUEUED"}) > 10
  for: 10m
  annotations:
    summary: "医保导入任务排队数超过 10 个"

总结

本文通过一个完整的医疗系统需求，展示了异步任务框架的接入全流程：

  • ✅ 需求分析：明确业务场景和技术要求
  • ✅ 方案设计：定义任务类型、数据模型、接口设计
  • ✅ 代码实现：Handler、Controller、前端页面
  • ✅ 配置调优：并发控制、动态策略、索引优化
  • ✅ 上线验证：功能、性能、容错、权限
  • ✅ 监控告警：业务指标、系统指标、告警规则

关键要点：

  1. Handler 必须实现幂等性
  2. 分批处理 + 定期上报进度
  3. 敏感操作需要重新鉴权
  4. 配置合理的并发和超时参数
  5. 完善的监控和告警体系

在下一篇文章中，我们将探讨异步任务框架的监控、运维与故障排查，敬请期待！

作者： 无声源语架构团队
发布日期： 2026-08-28
相关文章：

最后修改：2026 年 08 月 25 日
© 允许规范转载
如果觉得我的文章对你有用，请随意赞赏