字节跳动Java后端面试深度解析（三）：异步任务 REST API 完整方案 — 提交、查询、结果获取与自动清理

1
PENDING(待处理) → PROCESSING(处理中) → COMPLETED(已完成)
2
       ↓                ↓                    ↓
3
   DUPLICATE(重复)  FAILED(失败)      EXPIRED(已过期)

1
public enum TaskStatus {
2
    PENDING,     // 已提交，等待处理
3
    PROCESSING,  // 处理中
4
    COMPLETED,   // 已完成，可获取结果
5
    FAILED,      // 处理失败
6
    DUPLICATE,   // 与已有任务重复
7
    EXPIRED      // 结果已过期清理
8
}

1
CREATE TABLE async_task (
2
    id              BIGINT PRIMARY KEY AUTO_INCREMENT,
3
    task_id         VARCHAR(36) NOT NULL UNIQUE,  -- UUID，对外暴露
4
    task_type       VARCHAR(50) NOT NULL,          -- 任务类型（如 VIDEO_DESCRIPTION）
5
    dedup_key       VARCHAR(128),                  -- 去重键（业务维度）
6
    request_payload JSON NOT NULL,                 -- 原始请求参数
7
    status          VARCHAR(20) NOT NULL DEFAULT 'PENDING',
8
    result_payload  JSON,                          -- 任务结果
9
    error_message   TEXT,                          -- 失败原因
10
    progress        INT DEFAULT 0,                 -- 进度百分比 0-100
11
    created_at      TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
12
    started_at      TIMESTAMP,
13
    completed_at    TIMESTAMP,
14
    expires_at      TIMESTAMP,                     -- 结果过期时间
15
    retry_count     INT DEFAULT 0,
16
    version         INT DEFAULT 0,                 -- 乐观锁版本号
17

18
    INDEX idx_dedup_key (dedup_key),
19
    INDEX idx_status_created (status, created_at),
20
    INDEX idx_expires_at (expires_at)
21
);

1
@Entity
2
@Table(name = "async_task")
3
public class AsyncTask {
4

5
    @Id
6
    @GeneratedValue(strategy = GenerationType.IDENTITY)
7
    private Long id;
8

9
    @Column(name = "task_id", unique = true, nullable = false)
10
    private String taskId;
11

12
    @Column(name = "task_type", nullable = false)
13
    private String taskType;
14

15
    @Column(name = "dedup_key")
16
    private String dedupKey;
17

18
    @Column(name = "request_payload", columnDefinition = "JSON")
19
    @JdbcTypeCode(SqlTypes.JSON)
20
    private Map<String, Object> requestPayload;
21

22
    @Enumerated(EnumType.STRING)
23
    @Column(nullable = false)
24
    private TaskStatus status = TaskStatus.PENDING;
25

26
    @Column(name = "result_payload", columnDefinition = "JSON")
27
    @JdbcTypeCode(SqlTypes.JSON)
28
    private Map<String, Object> resultPayload;
29

30
    @Column(name = "error_message")
31
    private String errorMessage;
32

33
    @Column(columnDefinition = "INT DEFAULT 0")
34
    private int progress = 0;
35

36
    @Column(name = "created_at")
37
    private LocalDateTime createdAt = LocalDateTime.now();
38

39
    @Column(name = "started_at")
40
    private LocalDateTime startedAt;
41

42
    @Column(name = "completed_at")
43
    private LocalDateTime completedAt;
44

45
    @Column(name = "expires_at")
46
    private LocalDateTime expiresAt;
47

48
    @Column(name = "retry_count")
49
    private int retryCount = 0;
50

51
    @Version
52
    private int version; // 乐观锁
53

54
    // 状态流转方法
55
    public void startProcessing() {
56
        this.status = TaskStatus.PROCESSING;
57
        this.startedAt = LocalDateTime.now();
58
    }
59

60
    public void complete(Map<String, Object> result) {
61
        this.status = TaskStatus.COMPLETED;
62
        this.resultPayload = result;
63
        this.progress = 100;
64
        this.completedAt = LocalDateTime.now();
65
        this.expiresAt = LocalDateTime.now().plusHours(24);
66
    }
67

68
    public void fail(String errorMessage) {
69
        this.status = TaskStatus.FAILED;
70
        this.errorMessage = errorMessage;
71
        this.completedAt = LocalDateTime.now();
72
        this.expiresAt = LocalDateTime.now().plusHours(24);
73
    }
74

75
    public boolean isExpired() {
76
        return expiresAt != null && LocalDateTime.now().isAfter(expiresAt);
77
    }
78
}

1
@RestController
2
@RequestMapping("/api/v1/tasks")
3
public class AsyncTaskController {
4

5
    private final AsyncTaskService taskService;
6

7
    /**
8
     * 提交异步任务
9
     * POST /api/v1/tasks
10
     */
11
    @PostMapping
12
    public ResponseEntity<TaskSubmitResponse> submitTask(
13
            @Valid @RequestBody TaskSubmitRequest request) {
14

15
        TaskSubmitResponse response = taskService.submitTask(request);
16

17
        if (response.isDuplicate()) {
18
            // 去重命中，返回已有任务信息
19
            return ResponseEntity.status(HttpStatus.CONFLICT)
20
                .header("X-Task-Deduplicated", "true")
21
                .body(response);
22
        }
23

24
        // 202 Accepted：任务已接受，正在处理
25
        URI taskLocation = ServletUriComponentsBuilder.fromCurrentRequest()
26
            .path("/{taskId}")
27
            .buildAndExpand(response.getTaskId())
28
            .toUri();
29

30
        return ResponseEntity.accepted()
31
            .location(taskLocation)
32
            .body(response);
33
    }
34

35
    /**
36
     * 查询任务状态
37
     * GET /api/v1/tasks/{taskId}
38
     */
39
    @GetMapping("/{taskId}")
40
    public ResponseEntity<TaskStatusResponse> getTaskStatus(
41
            @PathVariable String taskId) {
42

43
        TaskStatusResponse response = taskService.getTaskStatus(taskId);
44

45
        return switch (response.getStatus()) {
46
            case COMPLETED -> ResponseEntity.ok(response);
47
            case PENDING, PROCESSING -> ResponseEntity.status(HttpStatus.OK)
48
                .header("Retry-After", "3")
49
                .body(response);
50
            case FAILED -> ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
51
                .body(response);
52
            case EXPIRED -> ResponseEntity.status(HttpStatus.GONE).body(response);
53
            default -> ResponseEntity.ok(response);
54
        };
55
    }
56

57
    /**
58
     * 获取任务结果
59
     * GET /api/v1/tasks/{taskId}/result
60
     */
61
    @GetMapping("/{taskId}/result")
62
    public ResponseEntity<TaskResultResponse> getTaskResult(
63
            @PathVariable String taskId) {
64

65
        TaskResultResponse result = taskService.getTaskResult(taskId);
66

67
        if (result.isExpired()) {
68
            return ResponseEntity.status(HttpStatus.GONE)
69
                .body(result);
70
        }
71
        if (result.getStatus() != TaskStatus.COMPLETED) {
72
            return ResponseEntity.status(HttpStatus.ACCEPTED)
73
                .header("Retry-After", "5")
74
                .body(result);
75
        }
76
        return ResponseEntity.ok(result);
77
    }
78

79
    /**
80
     * 取消任务
81
     * DELETE /api/v1/tasks/{taskId}
82
     */
83
    @DeleteMapping("/{taskId}")
84
    public ResponseEntity<Void> cancelTask(@PathVariable String taskId) {
85
        boolean cancelled = taskService.cancelTask(taskId);
86
        if (cancelled) {
87
            return ResponseEntity.noContent().build();
88
        }
89
        return ResponseEntity.status(HttpStatus.CONFLICT).build();
90
    }
91
}

1
public record TaskSubmitRequest(
2
    @NotBlank String taskType,
3
    @NotNull Map<String, Object> payload,
4
    String dedupKey,  // 可选，不填则按 payload hash 去重
5
    Integer priority  // 可选，优先级
6
) {}
7

8
public record TaskSubmitResponse(
9
    String taskId,
10
    TaskStatus status,
11
    boolean duplicate,
12
    String message
13
) {}
14

15
public record TaskStatusResponse(
16
    String taskId,
17
    TaskStatus status,
18
    int progress,
19
    LocalDateTime createdAt,
20
    LocalDateTime startedAt,
21
    LocalDateTime completedAt,
22
    String errorMessage
23
) {}
24

25
public record TaskResultResponse(
26
    String taskId,
27
    TaskStatus status,
28
    Map<String, Object> result,
29
    boolean expired
30
) {}

1
@Service
2
@Transactional
3
public class AsyncTaskService {
4

5
    private final AsyncTaskRepository taskRepository;
6
    private final TaskQueueProducer queueProducer;
7
    private final RedisTemplate<String, String> redisTemplate;
8

9
    private static final Duration DEDUP_WINDOW = Duration.ofMinutes(5);
10

11
    public TaskSubmitResponse submitTask(TaskSubmitRequest request) {
12
        // 1. 生成去重键
13
        String dedupKey = resolveDedupKey(request);
14

15
        // 2. 检查去重（Redis + DB 双重检查）
16
        if (dedupKey != null) {
17
            // Redis 快速检查
18
            String existingTaskId = redisTemplate.opsForValue()
19
                .get("task:dedup:" + dedupKey);
20
            if (existingTaskId != null) {
21
                return new TaskSubmitResponse(
22
                    existingTaskId,
23
                    TaskStatus.DUPLICATE,
24
                    true,
25
                    "Duplicate task found: " + existingTaskId
26
                );
27
            }
28

29
            // DB 检查（防止 Redis 数据丢失）
30
            taskRepository.findRecentByDedupKey(dedupKey, LocalDateTime.now().minusMinutes(5))
31
                .ifPresent(existing -> {
32
                    throw new DuplicateTaskException(existing.getTaskId());
33
                });
34
        }
35

36
        // 3. 创建任务
37
        AsyncTask task = new AsyncTask();
38
        task.setTaskId(UUID.randomUUID().toString());
39
        task.setTaskType(request.taskType());
40
        task.setDedupKey(dedupKey);
41
        task.setRequestPayload(request.payload());
42
        task.setStatus(TaskStatus.PENDING);
43
        taskRepository.save(task);
44

45
        // 4. 设置 Redis 去重标记
46
        if (dedupKey != null) {
47
            redisTemplate.opsForValue().set(
48
                "task:dedup:" + dedupKey,
49
                task.getTaskId(),
50
                DEDUP_WINDOW
51
            );
52
        }
53

54
        // 5. 投递到消息队列
55
        queueProducer.sendTaskMessage(task);
56

57
        return new TaskSubmitResponse(
58
            task.getTaskId(),
59
            TaskStatus.PENDING,
60
            false,
61
            "Task submitted successfully"
62
        );
63
    }
64

65
    private String resolveDedupKey(TaskSubmitRequest request) {
66
        if (request.dedupKey() != null) {
67
            return request.dedupKey();
68
        }
69
        // 按 payload 内容生成 hash 去重
70
        String payloadJson = JsonUtil.toJson(request.payload());
71
        return DigestUtils.sha256Hex(request.taskType() + ":" + payloadJson);
72
    }
73
}

1
@Component
2
public class AsyncTaskExecutor {
3

4
    private final AsyncTaskRepository taskRepository;
5
    private final Map<String, TaskProcessor> processors;
6
    private final AsyncTaskProgressCallback progressCallback;
7

8
    public AsyncTaskExecutor(List<TaskProcessor> processorList) {
9
        this.processors = processorList.stream()
10
            .collect(Collectors.toUnmodifiableMap(
11
                TaskProcessor::taskType,
12
                Function.identity()
13
            ));
14
    }
15

16
    /**
17
     * 异步执行任务，由消息队列消费者调用
18
     */
19
    @Async("taskExecutor")
20
    public void execute(String taskId) {
21
        AsyncTask task = taskRepository.findByTaskId(taskId)
22
            .orElseThrow(() -> new TaskNotFoundException(taskId));
23

24
        // CAS 更新状态：PENDING → PROCESSING
25
        boolean started = taskRepository.casUpdateStatus(
26
            taskId, TaskStatus.PENDING, TaskStatus.PROCESSING);
27
        if (!started) {
28
            // 状态已经不是 PENDING，可能被取消或重复消费
29
            return;
30
        }
31

32
        try {
33
            TaskProcessor processor = processors.get(task.getTaskType());
34
            if (processor == null) {
35
                throw new UnsupportedTaskTypeException(task.getTaskType());
36
            }
37

38
            // 执行业务逻辑
39
            Map<String, Object> result = processor.process(
40
                task.getRequestPayload(),
41
                progress -> updateProgress(taskId, progress)  // 进度回调
42
            );
43

44
            // 更新为完成状态
45
            taskRepository.casUpdateStatusAndResult(
46
                taskId, TaskStatus.PROCESSING, TaskStatus.COMPLETED, result);
47

48
        } catch (Exception e) {
49
            taskRepository.updateStatusWithError(
50
                taskId, TaskStatus.FAILED, e.getMessage());
51
        }
52
    }
53

54
    private void updateProgress(String taskId, int progress) {
55
        taskRepository.updateProgress(taskId, progress);
56
        progressCallback.onProgress(taskId, progress);
57
    }
58
}

1
@GetMapping(value = "/{taskId}/progress", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
2
public Flux<ServerSentEvent<TaskProgressEvent>> streamProgress(
3
        @PathVariable String taskId) {
4

5
    return Flux.create(sink -> {
6
        // 先推送当前状态
7
        TaskStatusResponse current = taskService.getTaskStatus(taskId);
8
        sink.next(ServerSentEvent.builder(
9
            new TaskProgressEvent(taskId, current.status(), current.progress()))
10
            .build());
11

12
        // 订阅进度更新
13
        Disposable subscription = progressEventBus.subscribe(taskId, event -> {
14
            sink.next(ServerSentEvent.builder(event).build());
15
            if (event.status() == TaskStatus.COMPLETED
16
                || event.status() == TaskStatus.FAILED) {
17
                sink.complete();
18
            }
19
        });
20

21
        sink.onDispose(subscription::dispose);
22
    });
23
}

1
客户端请求 → Redis 快速检查（毫秒级）
2
                ↓ miss
3
            数据库唯一约束检查（10ms 级）
4
                ↓ miss
5
            创建任务 + Redis 设置标记

1
// Redis Lua 脚本：原子性检查并设置
2
private static final String DEDUP_SCRIPT = """
3
    local key = KEYS[1]
4
    local taskId = ARGV[1]
5
    local ttl = tonumber(ARGV[2])
6
    local existing = redis.call('GET', key)
7
    if existing then
8
        return existing
9
    end
10
    redis.call('SETEX', key, ttl, taskId)
11
    return nil
12
    """;

1
@Component
2
public class AsyncTaskCleanupJob {
3

4
    private final AsyncTaskRepository taskRepository;
5
    private final RedisTemplate<String, String> redisTemplate;
6

7
    /**
8
     * 每小时清理一次过期任务
9
     */
10
    @Scheduled(cron = "0 0 * * * *")
11
    @Transactional
12
    public void cleanupExpiredTasks() {
13
        LocalDateTime cutoff = LocalDateTime.now().minusHours(24);
14

15
        // 1. 批量查询过期任务
16
        List<AsyncTask> expiredTasks = taskRepository
17
            .findByStatusInAndCompletedAtBefore(
18
                List.of(TaskStatus.COMPLETED, TaskStatus.FAILED),
19
                cutoff
20
            );
21

22
        if (expiredTasks.isEmpty()) return;
23

24
        // 2. 分批处理，避免大事务
25
        Lists.partition(expiredTasks, 100).forEach(batch -> {
26
            batch.forEach(task -> {
27
                // 删除结果数据（保留任务记录用于审计）
28
                task.setResultPayload(null);
29
                task.setStatus(TaskStatus.EXPIRED);
30
            });
31
            taskRepository.saveAll(batch);
32
        });
33

34
        log.info("Cleaned up {} expired tasks", expiredTasks.size());
35
    }
36

37
    /**
38
     * 清理 Redis 中的过期去重键
39
     * Redis 会自动过期，这里只是清理可能残留的异常数据
40
     */
41
    @Scheduled(cron = "0 30 * * * *")
42
    public void cleanupExpiredDedupKeys() {
43
        // Redis 的 TTL 机制会自动清理，此处为兜底逻辑
44
        Set<String> keys = redisTemplate.keys("task:dedup:*");
45
        if (keys == null || keys.isEmpty()) return;
46

47
        int cleaned = 0;
48
        for (String key : keys) {
49
            String taskId = redisTemplate.opsForValue().get(key);
50
            if (taskId != null) {
51
                Optional<AsyncTask> task = taskRepository.findByTaskId(taskId);
52
                if (task.isEmpty() || task.get().isExpired()) {
53
                    redisTemplate.delete(key);
54
                    cleaned++;
55
                }
56
            }
57
        }
58
        log.info("Cleaned {} stale dedup keys from Redis", cleaned);
59
    }
60
}

1
@Component
2
public class TaskResultStorage {
3

4
    private final S3Client s3Client;
5
    private final String resultBucket;
6

7
    /**
8
     * 存储大结果到 S3，数据库只保留引用
9
     */
10
    public String storeResult(String taskId, Map<String, Object> result) {
11
        String key = "task-results/" + taskId + ".json";
12
        String json = JsonUtil.toJson(result);
13

14
        s3Client.putObject(PutObjectRequest.builder()
15
            .bucket(resultBucket)
16
            .key(key)
17
            .contentType("application/json")
18
            .build(), RequestBody.fromString(json));
19

20
        return key;
21
    }
22

23
    /**
24
     * 获取结果（优先从 S3，降级从数据库）
25
     */
26
    public Map<String, Object> getResult(AsyncTask task) {
27
        if (task.getResultStorageKey() != null) {
28
            // 从 S3 获取
29
            ResponseInputStream<GetObjectResponse> response = s3Client.getObject(
30
                GetObjectRequest.builder()
31
                    .bucket(resultBucket)
32
                    .key(task.getResultStorageKey())
33
                    .build()
34
            );
35
            return JsonUtil.fromJson(response.readAllBytes(), Map.class);
36
        }
37
        // 降级：从数据库获取
38
        return task.getResultPayload();
39
    }
40
}