Spring Boot 打造全能異步處理方案，簡單高效！

作者：路條編程 2025-01-08 10:35:26

通過本文，開發者不僅可以掌握如何在 Spring Boot 中高效實現異步任務管理，還可以學到如何編寫更加清晰、可維護、易擴展的代碼。

在系統設計中，遵循“開閉原則”是良好實踐。隨著業務不斷演化和更新，核心代碼的頻繁改動不僅可能引入更多錯誤風險，還可能影響整體系統穩定性。盡管新增的功能大多是對現有功能的擴展，但如何在保證性能和質量的前提下平穩實現這些更新，成為了一大挑戰。為了解決這一問題，我設計了一套通用的異步處理SDK，能夠高效地實現各種異步任務的處理需求。

通過該SDK，不僅可以確保方法正常執行、不影響主業務流程，還能通過可靠的兜底方案保障數據不丟失，實現最終一致性。

設計優勢

無侵入性：獨立的數據庫、定時任務、消息隊列及人工操作界面（支持統一登錄認證）。
事務安全：基于Spring的事務事件機制，異步策略解析失敗時不影響主業務流程。
完善的兜底方案：即使異步策略解析失敗（如事務提交后或回滾后），也有多層次的補償機制（除非數據庫或隊列存在問題）。

工作原理

注解緩存：容器初始化時，掃描并緩存所有帶有@AsyncExec注解的方法。
AOP切面：方法執行時通過AOP切面發布事件。
事務監聽：通過@TransactionalEventListener實現事務事件監聽，處理異步策略。

事務事件監聽核心代碼

@TransactionalEventListener(fallbackExecution = true, phase = TransactionPhase.AFTER_COMPLETION)

fallbackExecution = true：即使沒有事務，也能處理事件。
TransactionPhase.AFTER_COMPLETION：事務提交或回滾后均會處理事件。

核心組件

消息隊列：Kafka
定時任務：XXL Job
數據庫：MySQL
前端界面：基于Vue實現
設計模式：策略模式、模板方法、動態代理

以下是一個完整的 Spring Boot 項目示例，展示如何使用 Kafka 消息隊列、XXL Job 定時任務、MySQL 數據庫和 Vue 前端界面，并結合策略模式、模板方法和動態代理實現通用異步處理功能。

項目結構

核心模塊

任務狀態枚舉 (TaskStatus)：定義任務的生命周期狀態。
任務實體 (AsyncTask)：數據庫表對應的實體類，記錄任務的執行信息。
任務處理策略 (AsyncStrategy)：定義任務的執行邏輯，支持動態擴展。
任務調度器 (AsyncTaskScheduler)：調度任務執行并管理任務狀態。
任務監控器 (TaskMonitorService)：實時監控任務狀態，提供告警能力。

數據庫設計

任務表 DDL

CREATE TABLE async_task (
    id BIGINTAUTO_INCREMENTPRIMARYKEY,
    task_name VARCHAR(255)NOTNULLCOMMENT'任務名稱',
    task_status INTNOTNULLDEFAULT0COMMENT'任務狀態',
    task_param TEXTCOMMENT'任務參數',
    retry_count INTDEFAULT0COMMENT'重試次數',
    create_time DATETIMEDEFAULTCURRENT_TIMESTAMPCOMMENT'創建時間',
    update_time DATETIMEDEFAULTCURRENT_TIMESTAMPONUPDATECURRENT_TIMESTAMPCOMMENT'更新時間'
);

核心代碼實現

任務狀態枚舉

package com.icoderoad.async.enums;
public enum TaskStatus {
    INITIAL(0, "初始化"),
    PROCESSING(1, "執行中"),
    SUCCESS(2, "執行成功"),
    FAILED(3, "執行失敗");


    private final int code;
    private final String description;


    TaskStatus(int code, String description) {
        this.code = code;
        this.description = description;
    }


    public int getCode() {
        return code;
    }


    public String getDescription() {
        return description;
    }


    public static TaskStatus fromCode(int code) {
        for (TaskStatus status : values()) {
            if (status.code == code) {
                return status;
            }
        }
        throw new IllegalArgumentException("未知任務狀態代碼：" + code);
    }
}

任務實體

package com.icoderoad.async.entity;
@Data
@TableName("async_task")
public class AsyncTask {
    private Long id;
    private String taskName;
    private TaskStatus taskStatus;
    private String taskParam;
    private Integer retryCount;
    private LocalDateTime createTime;
    private LocalDateTime updateTime;
}

任務策略接口

package com.icoderoad.async.strategy;
public interface AsyncStrategy {
    void execute(String taskParam);
}

任務執行服務

package com.icoderoad.async.service;


import com.icoderoad.async.entity.AsyncTask;
import com.icoderoad.async.enums.TaskStatus;
import com.icoderoad.async.repository.AsyncTaskRepository;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;


import java.time.LocalDateTime;
import java.util.List;
import java.util.Optional;


@Service
public class AsyncTaskService {


    private final AsyncTaskRepository asyncTaskRepository;


    @Autowired
    public AsyncTaskService(AsyncTaskRepository asyncTaskRepository) {
        this.asyncTaskRepository = asyncTaskRepository;
    }


    /**
     * 獲取所有異步任務
     */
    public List<AsyncTask> getAllTasks() {
        return asyncTaskRepository.findAll();
    }


    /**
     * 重試任務
     *
     * @param taskId 任務ID
     * @return 是否重試成功
     */
    @Transactional
    public boolean retryTask(Long taskId) {
        Optional<AsyncTask> optionalTask = asyncTaskRepository.findById(taskId);
        if (optionalTask.isPresent()) {
            AsyncTask task = optionalTask.get();


            // 檢查任務是否允許重試
            if (task.getTaskStatus() == TaskStatus.FAILED) {
                task.setTaskStatus(TaskStatus.INITIAL); // 將狀態重置為初始化
                task.setRetryCount(task.getRetryCount() + 1); // 增加重試次數
                task.setUpdateTime(LocalDateTime.now());
                asyncTaskRepository.save(task);
                return true;
            }
        }
        return false; // 任務不存在或狀態異常
    }


    /**
     * 創建新異步任務
     *
     * @param taskName  任務名稱
     * @param taskParam 任務參數
     */
    @Transactional
    public void createAsyncTask(String taskName, String taskParam) {
        AsyncTask newTask = new AsyncTask();
        newTask.setTaskName(taskName);
        newTask.setTaskParam(taskParam);
        newTask.setTaskStatus(TaskStatus.INITIAL);
        newTask.setRetryCount(0);
        newTask.setCreateTime(LocalDateTime.now());
        newTask.setUpdateTime(LocalDateTime.now());
        asyncTaskRepository.save(newTask);
    }


    /**
     * 獲取待執行的任務（供調度器使用）
     */
    public List<AsyncTask> getPendingTasks() {
        return asyncTaskRepository.findByTaskStatus(TaskStatus.INITIAL);
    }


    /**
     * 更新任務狀態
     *
     * @param task 更新后的任務
     */
    @Transactional
    public void updateTask(AsyncTask task) {
        task.setUpdateTime(LocalDateTime.now());
        asyncTaskRepository.save(task);
    }
}

任務調度器

@Component
public class AsyncTaskScheduler {


    private final AsyncTaskService asyncTaskService;
    private final TaskMonitorService taskMonitorService;


    @Scheduled(fixedRate = 5000)
    public void scheduleTasks() {
        List<AsyncTask> tasks = asyncTaskService.getPendingTasks();
        tasks.forEach(task -> {
            try {
                asyncTaskService.executeTask(task);
                task.setTaskStatus(TaskStatus.SUCCESS);
            } catch (Exception e) {
                task.setTaskStatus(TaskStatus.FAILED);
                taskMonitorService.alertOnFailedTask(task);
            }
            asyncTaskService.updateTask(task);
        });
    }
}

配置文件

application.yml

spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: async-tasks-group
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

Controller 類代碼

package com.icoderoad.async.controller;


import com.icoderoad.async.dto.AsyncTaskDto;
import com.icoderoad.async.entity.AsyncTask;
import com.icoderoad.async.service.AsyncTaskService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;


import java.util.List;
import java.util.stream.Collectors;


@RestController
@RequestMapping("/api/tasks")
public class AsyncTaskController {


    private final AsyncTaskService asyncTaskService;


    @Autowired
    public AsyncTaskController(AsyncTaskService asyncTaskService) {
        this.asyncTaskService = asyncTaskService;
    }


    /**
     * 獲取所有異步任務
     */
    @GetMapping
    public ResponseEntity<List<AsyncTaskDto>> getAllTasks() {
        List<AsyncTask> tasks = asyncTaskService.getAllTasks();
        List<AsyncTaskDto> taskDtos = tasks.stream()
                .map(task -> new AsyncTaskDto(task.getId(), task.getTaskName(), task.getTaskStatus(), task.getTaskParam()))
                .collect(Collectors.toList());
        return ResponseEntity.ok(taskDtos);
    }


    /**
     * 根據 ID 重試任務
     */
    @PostMapping("/{id}/retry")
    public ResponseEntity<String> retryTask(@PathVariable Long id) {
        boolean result = asyncTaskService.retryTask(id);
        return result ? ResponseEntity.ok("任務重試成功") : ResponseEntity.badRequest().body("任務重試失敗，任務可能不存在或狀態異常");
    }


    /**
     * 創建新異步任務
     */
    @PostMapping
    public ResponseEntity<String> createTask(@RequestParam String taskName, @RequestParam String taskParam) {
        asyncTaskService.createAsyncTask(taskName, taskParam);
        return ResponseEntity.ok("異步任務創建成功");
    }
}

DTO 類代碼

用于返回給前端的任務數據傳輸對象。

package com.icoderoad.async.dto;


import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;


@Data
@AllArgsConstructor
@NoArgsConstructor
public class AsyncTaskDto {
    private Long id;
    private String taskName;
    private Integer taskStatus;
    private String taskParam;
}

前端界面 (Vue.js)

異步任務列表

<template>
  <div>
    <h1>異步任務管理</h1>
    <table>
      <thead>
        <tr>
          <th>任務名稱</th>
          <th>狀態</th>
          <th>參數</th>
          <th>操作</th>
        </tr>
      </thead>
      <tbody>
        <tr v-for="task in tasks" :key="task.id">
          <td>{{ task.taskName }}</td>
          <td>{{ task.taskStatus }}</td>
          <td>{{ task.taskParam }}</td>
          <td><button @click="retryTask(task.id)">重試</button></td>
        </tr>
      </tbody>
    </table>
  </div>
</template>

<script>
export default {
  data() {
    return {
      tasks: [],
    };
  },
  methods: {
    fetchTasks() {
      axios.get('/api/tasks').then(response => {
        this.tasks = response.data;
      });
    },
    retryTask(taskId) {
      axios.post(`/api/tasks/${taskId}/retry`);
    },
  },
  mounted() {
    this.fetchTasks();
  },
};
</script>

總結

本文深入探討了基于 Spring Boot 開發異步任務管理功能的實現方法，涵蓋了從控制器設計到服務層邏輯優化的全過程。通過清晰的代碼示例和詳細的講解，讀者可以輕松掌握以下關鍵內容：

1.異步任務管理的核心功能：

實現了任務的增刪改查、狀態管理，以及失敗任務的重試機制，確保異步任務生命周期的完整性。

2.面向業務場景的邏輯優化：

針對任務狀態進行了明確的校驗與約束，通過 TaskStatus 枚舉提升代碼的可讀性和維護性。
重試邏輯中考慮了任務的狀態異常場景，避免因錯誤操作導致任務狀態混亂。

3.面向開發實踐的細節增強：

使用 Spring 的 @Transactional 注解確保數據操作的事務安全，避免并發修改導致數據不一致。
在任務的創建、更新、重試操作中添加 createTime 和 updateTime 字段的動態更新，確保時間戳的準確性。
提供了調度器支持的擴展方法，為后續的任務調度和自動化運行奠定基礎。

4.代碼解耦與擴展性設計：

通過服務層和數據層的職責分離，實現了業務邏輯和數據訪問的解耦。
使用 Spring Data JPA 提供的數據倉庫方法，使代碼更簡潔高效，并易于擴展其他查詢需求。

5.可靠性與易用性兼備：

在功能實現的同時，確保代碼的健壯性和高可維護性。無論是單獨的功能測試，還是集成到更復雜的業務流程中，都能穩定運行。

通過本文，開發者不僅可以掌握如何在 Spring Boot 中高效實現異步任務管理，還可以學到如何編寫更加清晰、可維護、易擴展的代碼。希望這篇文章能夠為您在實際開發中提供參考，并助力您設計出更優雅的任務管理系統！

責任編輯：武曉燕來源：路條編程

代碼開發者 Spring