1. 长时推理任务的挑战
2. 异步任务架构设计

   2.1 任务生命周期管理

   2.2 消息队列选型

1. 任务队列设计与实现

   3.1 任务模型与持久化

   3.2 任务生产端设计

   3.3 任务消费端设计

1. 回调与通知机制

   4.1 Webhook回调设计

   4.2 消息推送通知

1. 幂等性与可靠投递
2. 任务优先级与调度策略
3. 监控与可观测性
4. 总结

大模型推理并非总是瞬间完成。当面对复杂任务时，如长篇文档生成、代码审查分析、多轮对话深度推理等场景，模型推理可能耗时数十秒甚至数分钟。如果将这些请求同步阻塞地等待结果，将导致一组严重问题。

首先，同步等待占用宝贵的连接资源。HTTP连接线程被阻塞，连接池快速耗尽，系统吞吐量急剧下降。在Tomcat默认200线程的配置下，如果每个请求等待30秒，系统的最大并发处理能力仅为每秒约7个请求。其次，客户端体验极差。用户看到的是长时间无响应的加载状态，不知道任务是否在执行，也无法中途取消。第三，缺乏弹性伸缩能力。所有任务都在请求线程中执行，无法根据任务积压情况动态扩缩容。

将长时推理任务从同步请求响应模式中解耦出来，转变为异步任务模式，是解决这些问题的根本之道。异步任务服务的核心思想是"提交即返回"，客户端提交任务后立即获得任务标识，后续通过轮询或回调获取结果。

这种架构将系统的可用性和吞吐量提升了一个数量级，但同时也带来了新的复杂性问题：任务状态的可靠管理、失败重试的策略、回调机制的可靠性、分布式环境下的幂等性保证等。本章将对这些问题逐一深入探讨。

2.1 任务生命周期管理

异步任务从创建到完成经历多个状态转换。一个设计良好的任务状态机是系统可靠性的基础。典型的任务生命周期包含以下状态：

• PENDING：任务已创建但尚未提交到队列，或已持久化等待处理
• QUEUED：任务已入队，等待工作节点取走执行
• PROCESSING：任务正在被工作节点处理
• COMPLETED：任务已成功完成，结果可用
• FAILED：任务执行失败，可能包含错误信息
• CANCELLED：任务被主动取消

状态转换遵循严格的规则：PENDING可转换为QUEUED或CANCELLED，QUEUED可转换为PROCESSING或CANCELLED，PROCESSING可转换为COMPLETED、FAILED或CANCELLED。

public enum TaskStatus {
    PENDING("待处理"),
    QUEUED("已入队"),
    PROCESSING("处理中"),
    COMPLETED("已完成"),
    FAILED("失败"),
    CANCELLED("已取消");
    private final String description;
    public boolean canTransitionTo(TaskStatus target) {
        return switch (this) {
            case PENDING -> Set.of(QUEUED, CANCELLED).contains(target);
            case QUEUED -> Set.of(PROCESSING, CANCELLED).contains(target);
            case PROCESSING -> Set.of(COMPLETED, FAILED, CANCELLED).contains(target);
            case COMPLETED, FAILED, CANCELLED -> false;
        };
    }
}

任务模型包含业务所需的核心字段：

@Data
@Builder
public class AsyncTask {
    private String taskId;
    private String taskType;
    private String userId;
    private String modelId;
    private TaskStatus status;
    private String requestPayload;     // JSON格式的请求参数
    private String resultPayload;      // JSON格式的执行结果
    private String errorMessage;
    private Integer priority;
    private Integer progress;          // 0-100
    private LocalDateTime createdAt;
    private LocalDateTime startedAt;
    private LocalDateTime completedAt;
    private Integer retryCount;
    private Integer maxRetries;
    private String callbackUrl;
    private Map<String, String> metadata;
}

2.2 消息队列选型

消息队列是异步任务架构的神经中枢。在大模型推理场景下，选型主要考虑RocketMQ、RabbitMQ和Apache Kafka三种方案。

RabbitMQ基于AMQP协议，支持丰富的路由模式和消息确认机制，延迟低至毫秒级。对于任务量在每天数十万级别、需要灵活路由和可靠投递的场景非常合适。其管理界面直观，运维门槛低。

RocketMQ由阿里巴巴开源，在Java生态中集成度极高。支持事务消息、顺序消息和定时消息，对于需要精确调度和事务保证的场景是理想选择。延时消息能力可以直接用于任务的重试调度。

Kafka是高吞吐量的分布式流平台，单机可支撑数十万TPS。对于超大规模系统（每日千万级任务），Kafka的持久化日志存储和分区机制提供了极佳的扩展性。但其消息投递语义为至少一次，需要业务层实现幂等性。

在实际项目中，建议根据任务量级和团队技术储备选择。中小规模项目使用RocketMQ既能满足功能需求又不会过度复杂。

集成RocketMQ的依赖配置：

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.2.3</version>
</dependency>

配置文件中设置生产者组和消费者组：

rocketmq:
  name-server: 127.0.0.1:9876
  producer:
    group: llm-task-producer
    send-message-timeout: 5000
    retry-times-when-send-failed: 3
  consumer:
    group: llm-task-consumer
    max-reconsume-times: 3

3.1 任务模型与持久化

任务数据需要可靠持久化，一方面用于任务状态追踪和历史查询，另一方面作为消息队列的数据补偿。当消息丢失时，可以从数据库恢复任务重新投递。

使用MySQL存储任务数据，表结构设计如下：

CREATE TABLE llm_async_task (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    task_id VARCHAR(64) NOT NULL UNIQUE,
    task_type VARCHAR(32) NOT NULL,
    user_id VARCHAR(64) NOT NULL,
    model_id VARCHAR(64),
    status VARCHAR(20) NOT NULL DEFAULT 'PENDING',
    request_payload JSON,
    result_payload JSON,
    error_message TEXT,
    priority INT DEFAULT 0,
    progress INT DEFAULT 0,
    retry_count INT DEFAULT 0,
    max_retries INT DEFAULT 3,
    callback_url VARCHAR(512),
    metadata JSON,
    created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
    started_at DATETIME,
    completed_at DATETIME,
    updated_at DATETIME ON UPDATE CURRENT_TIMESTAMP,
    INDEX idx_user_status (user_id, status),
    INDEX idx_status_created (status, created_at),
    INDEX idx_task_type_status (task_type, status)
);

Repository层封装数据库操作：

@Repository
public interface AsyncTaskRepository extends JpaRepository<AsyncTaskEntity, Long> {
    Optional<AsyncTaskEntity> findByTaskId(String taskId);
    List<AsyncTaskEntity> findByUserIdAndStatus(
        String userId, String status, Pageable pageable);
    @Modifying
    @Query("UPDATE AsyncTaskEntity t SET t.status = :newStatus, t.startedAt = NOW() " +
           "WHERE t.taskId = :taskId AND t.status = :expectedStatus")
    int updateStatusToProcessing(
        @Param("taskId") String taskId,
        @Param("expectedStatus") String expectedStatus,
        @Param("newStatus") String newStatus);
    @Modifying
    @Query("UPDATE AsyncTaskEntity t SET t.status = :newStatus, t.progress = :progress " +
           "WHERE t.taskId = :taskId")
    int updateProgress(
        @Param("taskId") String taskId,
        @Param("progress") int progress);
    List<AsyncTaskEntity> findByStatusAndCreatedAtBefore(
        String status, LocalDateTime before);
}

乐观锁用于防止分布式环境下的状态并发冲突。通过status字段作为条件，使用updateStatusToProcessing方法确保只有当前状态为QUEUED的任务才能被标记为PROCESSING。这避免了多个消费者同时获取同一任务的竞态条件。

3.2 任务生产端设计

任务生产端负责接收客户端的任务提交请求，完成验证、持久化和消息投递。其核心职责是快速响应——在毫秒级完成这些操作后立即返回taskId给客户端，不等待任务实际执行。

@Service
@Slf4j
public class AsyncTaskService {
    private final AsyncTaskRepository repository;
    private final RocketMQTemplate rocketMQTemplate;
    public AsyncTaskService(AsyncTaskRepository repository,
                            RocketMQTemplate rocketMQTemplate) {
        this.repository = repository;
        this.rocketMQTemplate = rocketMQTemplate;
    }
    @Transactional
    public TaskSubmitResponse submitTask(TaskSubmitRequest request) {
        String taskId = generateTaskId();
        AsyncTaskEntity entity = AsyncTaskEntity.builder()
            .taskId(taskId)
            .taskType(request.getTaskType())
            .userId(request.getUserId())
            .modelId(request.getModelId())
            .status(TaskStatus.PENDING.name())
            .requestPayload(JsonUtil.toJson(request.getParams()))
            .priority(request.getPriority())
            .maxRetries(request.getMaxRetries())
            .callbackUrl(request.getCallbackUrl())
            .createdAt(LocalDateTime.now())
            .build();
        repository.save(entity);
        TaskMessage message = TaskMessage.builder()
            .taskId(taskId)
            .taskType(request.getTaskType())
            .modelId(request.getModelId())
            .payload(request.getParams())
            .priority(request.getPriority())
            .build();
        // 事务消息：确保持久化和消息投递的原子性
        rocketMQTemplate.sendMessageInTransaction(
            "llm-task-topic",
            MessageBuilder.withPayload(message)
                .setHeader("taskId", taskId)
                .build(),
            entity.getId());
        log.info("Task submitted: taskId={}, type={}", taskId, request.getTaskType());
        return TaskSubmitResponse.builder()
            .taskId(taskId)
            .status(TaskStatus.PENDING.name())
            .build();
    }
    public TaskQueryResponse queryTask(String taskId) {
        AsyncTaskEntity entity = repository.findByTaskId(taskId)
            .orElseThrow(() -> new TaskNotFoundException(taskId));
        return TaskQueryResponse.builder()
            .taskId(entity.getTaskId())
            .status(entity.getStatus())
            .progress(entity.getProgress())
            .result(entity.getResultPayload() != null ?
                JsonUtil.fromJson(entity.getResultPayload()) : null)
            .errorMessage(entity.getErrorMessage())
            .createdAt(entity.getCreatedAt())
            .completedAt(entity.getCompletedAt())
            .build();
    }
}

事务消息是保障可靠性的关键。在任务持久化到数据库的同时，通过RocketMQ的事务消息机制确保消息一定被投递。如果消息投递失败，事务回滚，数据库中不会留下孤儿记录。如果数据库写入成功但消息发送确认阶段失败，RocketMQ会回调检查器进行状态确认。

3.3 任务消费端设计

消费端从消息队列中获取任务，执行实际的模型推理，并更新任务状态。考虑到大模型推理的耗时特性，消费端需要支持并发消费和优雅的并发控制。

@Component
@RocketMQMessageListener(
    topic = "llm-task-topic",
    consumerGroup = "llm-task-consumer",
    maxReconsumeTimes = 3
)
@Slf4j
public class LlmTaskConsumer implements RocketMQListener<TaskMessage> {
    private final AsyncTaskRepository repository;
    private final Map<String, ModelExecutor> executors;
    private final TaskCallbackService callbackService;
    private final Semaphore concurrencyLimiter;
    public LlmTaskConsumer(AsyncTaskRepository repository,
                           List<ModelExecutor> executorList,
                           TaskCallbackService callbackService,
                           @Value("${task.consumer.max-concurrent:5}") int maxConcurrent) {
        this.repository = repository;
        this.executors = executorList.stream()
            .collect(Collectors.toMap(ModelExecutor::getModelId, e -> e));
        this.callbackService = callbackService;
        this.concurrencyLimiter = new Semaphore(maxConcurrent);
    }
    @Override
    public void onMessage(TaskMessage message) {
        String taskId = message.getTaskId();
        // 幂等性检查：如果任务已经在处理或已完成，跳过
        AsyncTaskEntity entity = repository.findByTaskId(taskId).orElse(null);
        if (entity == null) return;
        if (entity.getStatus().equals(TaskStatus.COMPLETED.name()) ||
            entity.getStatus().equals(TaskStatus.FAILED.name())) {
            log.info("Task already finished, skip: taskId={}", taskId);
            return;
        }
        // 乐观锁抢占：只有QUEUED状态的任务才能被获取
        int updated = repository.updateStatusToProcessing(
            taskId, TaskStatus.QUEUED.name(), TaskStatus.PROCESSING.name());
        if (updated == 0) {
            log.info("Task already claimed by another consumer: taskId={}", taskId);
            return;
        }
        try {
            concurrencyLimiter.acquire();
            executeTask(entity, message);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            handleTaskFailure(entity, "Task execution interrupted");
        } finally {
            concurrencyLimiter.release();
        }
    }
    private void executeTask(AsyncTaskEntity entity, TaskMessage message) {
        try {
            ModelExecutor executor = executors.get(message.getModelId());
            if (executor == null) {
                handleTaskFailure(entity, "Unknown model: " + message.getModelId());
                return;
            }
            Map<String, Object> params = message.getPayload();
            ExecutionResult result = executor.execute(
                params, progress -> {
                    repository.updateProgress(entity.getTaskId(), progress);
                });
            entity.setStatus(TaskStatus.COMPLETED.name());
            entity.setResultPayload(JsonUtil.toJson(result.getOutput()));
            entity.setCompletedAt(LocalDateTime.now());
            repository.save(entity);
            callbackService.onTaskCompleted(entity);
        } catch (Exception e) {
            log.error("Task execution failed: taskId={}", entity.getTaskId(), e);
            handleTaskFailure(entity, e.getMessage());
        }
    }
    private void handleTaskFailure(AsyncTaskEntity entity, String error) {
        int retryCount = entity.getRetryCount() + 1;
        if (retryCount <= entity.getMaxRetries()) {
            entity.setStatus(TaskStatus.QUEUED.name());
            entity.setRetryCount(retryCount);
            entity.setErrorMessage(error);
            repository.save(entity);
            // 重新投递到队列，延迟重试
            rocketMQTemplate.syncSend("llm-task-topic",
                buildRetryMessage(entity),
                3000,  // timeout
                retryCount);  // delay level
        } else {
            entity.setStatus(TaskStatus.FAILED.name());
            entity.setErrorMessage(error);
            entity.setCompletedAt(LocalDateTime.now());
            repository.save(entity);
            callbackService.onTaskFailed(entity);
        }
    }
}

消费端使用了Semaphore进行并发控制。虽然RocketMQ消费者本身支持并行消费，但大模型推理是GPU密集型任务，GPU资源通常有限。通过Semaphore将并发限制为GPU实例数或略高于实例数，避免过多任务竞争同一GPU导致全部超时。

4.1 Webhook回调设计

回调是异步任务通知客户端的核心机制。设计一个可靠的Webhook回调系统需要处理重试、超时和签名验证。

@Service
@Slf4j
public class TaskCallbackService {
    private final RestTemplate restTemplate;
    private final ObjectMapper objectMapper;
    private final String callbackSecret;
    public TaskCallbackService(
            @Value("${callback.secret}") String callbackSecret) {
        this.restTemplate = new RestTemplate();
        this.objectMapper = new ObjectMapper();
        this.callbackSecret = callbackSecret;
    }
    public void onTaskCompleted(AsyncTaskEntity task) {
        if (task.getCallbackUrl() == null) return;
        executeCallback(task, CallbackEvent.COMPLETED);
    }
    public void onTaskFailed(AsyncTaskEntity task) {
        if (task.getCallbackUrl() == null) return;
        executeCallback(task, CallbackEvent.FAILED);
    }
    private void executeCallback(AsyncTaskEntity task, CallbackEvent event) {
        CallbackPayload payload = CallbackPayload.builder()
            .taskId(task.getTaskId())
            .event(event.name())
            .status(task.getStatus())
            .result(task.getResultPayload())
            .error(task.getErrorMessage())
            .timestamp(Instant.now().toEpochMilli())
            .build();
        // 生成HMAC签名
        String signature = generateSignature(payload);
        HttpHeaders headers = new HttpHeaders();
        headers.setContentType(MediaType.APPLICATION_JSON);
        headers.set("X-Callback-Signature", signature);
        headers.set("X-Callback-Event", event.name());
        HttpEntity<CallbackPayload> request = new HttpEntity<>(payload, headers);
        // 指数退避重试
        for (int attempt = 0; attempt < 5; attempt++) {
            try {
                ResponseEntity<String> response = restTemplate.postForEntity(
                    task.getCallbackUrl(), request, String.class);
                if (response.getStatusCode().is2xxSuccessful()) {
                    log.info("Callback succeeded: taskId={}, attempt={}",
                        task.getTaskId(), attempt + 1);
                    return;
                }
            } catch (Exception e) {
                log.warn("Callback failed: taskId={}, attempt={}",
                    task.getTaskId(), attempt + 1, e);
            }
            long delay = (long) Math.pow(2, attempt) * 1000;
            try { Thread.sleep(delay); } catch (InterruptedException e) {
                Thread.currentThread().interrupt(); break;
            }
        }
        log.error("Callback exhausted retries: taskId={}", task.getTaskId());
        // 记录到失败队列，由补偿任务处理
        saveToCallbackRetryQueue(task);
    }
    private String generateSignature(CallbackPayload payload) {
        try {
            String data = objectMapper.writeValueAsString(payload);
            Mac mac = Mac.getInstance("HmacSHA256");
            SecretKeySpec spec = new SecretKeySpec(
                callbackSecret.getBytes(StandardCharsets.UTF_8), "HmacSHA256");
            mac.init(spec);
            byte[] hash = mac.doFinal(data.getBytes(StandardCharsets.UTF_8));
            return Base64.getEncoder().encodeToString(hash);
        } catch (Exception e) {
            throw new RuntimeException("Signature generation failed", e);
        }
    }
}

签名机制：使用HMAC-SHA256对回调数据进行签名，确保回调请求的来源真实性。客户端收到回调后使用相同的密钥验证签名，防止伪造回调。

4.2 消息推送通知

对于WebSocket连接或Server-Sent Events保持的长连接客户端，可以直接通过推送通道通知任务状态变更。这种实时通知方式比轮询更高效。

@Service
public class TaskNotificationService {
    private final Map<String, SseEmitter> userEmitters = new ConcurrentHashMap<>();
    public SseEmitter subscribe(String userId) {
        SseEmitter emitter = new SseEmitter(3600_000L); // 1小时超时
        userEmitters.put(userId, emitter);
        emitter.onCompletion(() -> userEmitters.remove(userId));
        emitter.onTimeout(() -> userEmitters.remove(userId));
        emitter.onError(e -> userEmitters.remove(userId));
        return emitter;
    }
    public void notifyTaskUpdate(String userId, TaskStatusUpdate update) {
        SseEmitter emitter = userEmitters.get(userId);
        if (emitter != null) {
            try {
                emitter.send(SseEmitter.event()
                    .name("task-update")
                    .data(update));
            } catch (IOException e) {
                userEmitters.remove(userId);
                log.warn("Failed to send SSE notification to user: {}", userId, e);
            }
        }
    }
}

在分布式消息系统中，消息可能被重复投递。网络抖动、消费者重平衡、超时重试等都可能导致同一消息被处理多次。因此，消费端必须实现幂等性保证。

实现幂等性的核心策略有几种。第一种是利用数据库唯一约束。在消费消息前尝试插入一条处理记录，利用数据库的唯一索引来防止重复处理。如果插入成功则继续处理，如果插入冲突（DuplicateKeyException）则说明消息已被处理或正在被处理，直接跳过。

第二种是利用Redis的SETNX命令。以taskId为key在Redis中设置一个短暂的生命周期标记，SETNX成功则表示是首次处理。

第三种是利用任务状态本身。如前文代码所示，使用乐观锁更新状态——只有状态为QUEUED的任务才能被更新为PROCESSING。即使消息重复投递，后续的消息会发现任务状态已非QUEUED而直接跳过。

这三种策略可以组合使用以提供更强的保证。例如，Redis做第一层快速过滤，数据库状态做第二层可靠判断。

@Component
public class TaskIdempotencyGuard {
    private final RedisTemplate<String, String> redisTemplate;
    public boolean tryProcess(String taskId) {
        String key = "task:processing:" + taskId;
        Boolean success = redisTemplate.opsForValue()
            .setIfAbsent(key, "1", Duration.ofMinutes(30));
        return Boolean.TRUE.equals(success);
    }
    public void releaseProcessing(String taskId) {
        redisTemplate.delete("task:processing:" + taskId);
    }
}

在大模型推理场景中，并非所有任务都具有相同的紧迫性。实时对话需要秒级响应，批量文档处理可以容忍分钟级延迟。如果所有任务都进入同一个FIFO队列，紧急任务可能被批量任务阻塞。

多级优先级队列是解决这一问题的有效方法。将任务按优先级划分为高、中、低三个等级，每个等级使用独立的RocketMQ Topic分区。

public enum TaskPriority {
    HIGH(3, "llm-task-high-topic"),
    NORMAL(2, "llm-task-normal-topic"),
    LOW(1, "llm-task-low-topic");
    private final int level;
    private final String topic;
}
// 消费端：优先消费高优先级队列
@Component
public class PriorityTaskDispatcher {
    private final Map<Integer, List<LlmTaskConsumer>> consumerPools;
    public PriorityTaskDispatcher() {
        // 高优先级分配更多消费线程
        consumerPools = Map.of(
            3, createConsumers(6),   // HIGH: 6线程
            2, createConsumers(4),   // NORMAL: 4线程
            1, createConsumers(2)    // LOW: 2线程
        );
    }
}

在消费者线程分配上采用加权策略。高优先级队列分配更多的消费线程，确保紧急任务得到及时响应。低优先级任务在资源紧张时可以降级或延迟处理。

异步任务系统的监控维度比同步系统更为复杂。除了标准的系统指标外，还需要关注任务维度的指标。

关键监控指标包括：各状态任务的数量分布（特别是QUEUED状态的积压量）、任务的平均处理时间和P99延迟、任务的成功率和失败率按任务类型分类、消息队列的消费延迟（从生产到消费的端到端时间）、回调的成功率和重试次数分布。

通过Micrometer + Prometheus + Grafana搭建监控体系：

@Component
public class TaskMetricsCollector {
    private final MeterRegistry registry;
    private final Counter taskSubmitted;
    private final Counter taskCompleted;
    private final Counter taskFailed;
    private final Timer taskExecutionTime;
    private final Gauge queueDepth;
    public TaskMetricsCollector(MeterRegistry registry,
                                AsyncTaskRepository repository) {
        this.registry = registry;
        this.taskSubmitted = Counter.builder("llm.task.submitted")
            .description("Number of tasks submitted")
            .tag("service", "async-task")
            .register(registry);
        this.taskCompleted = Counter.builder("llm.task.completed")
            .register(registry);
        this.taskFailed = Counter.builder("llm.task.failed")
            .register(registry);
        this.taskExecutionTime = Timer.builder("llm.task.execution.time")
            .publishPercentiles(0.5, 0.95, 0.99)
            .register(registry);
        this.queueDepth = Gauge.builder("llm.task.queue.depth",
                () -> repository.countByStatus(TaskStatus.QUEUED.name()))
            .register(registry);
    }
    public void recordTaskSubmitted(String taskType) {
        taskSubmitted.increment();
    }
    public void recordTaskCompleted(String taskType, long durationMs) {
        taskCompleted.increment();
        taskExecutionTime.record(durationMs, TimeUnit.MILLISECONDS);
    }
    public void recordTaskFailed(String taskType) {
        taskFailed.increment();
    }
}

关键告警规则：队列积压超过1000条且持续5分钟以上触发告警；任务失败率超过10%触发告警；消费延迟超过30秒触发告警；回调成功率低于90%触发告警。

异步任务服务是大模型微服务架构中解耦长时推理请求的核心组件，它将"请求-等待"模式转变为"提交-查询/回调"模式，显著提升了系统的吞吐量和用户体验。

本章的核心要点如下。第一，通过消息队列实现任务的生产与消费解耦，使系统具备弹性伸缩和削峰填谷的能力。第二，使用事务消息结合数据库持久化保障任务的可靠投递。第三，通过乐观锁和幂等性设计防止分布式环境下的重复处理。第四，Webhook回调机制结合HMAC签名实现安全可靠的结果通知。第五，多级优先级队列确保紧急任务的SLA不被批量任务影响。第六，完善的监控体系是异步系统运维的基础保障。

在下一章中，我们将讨论流式输出服务的独立化改造，深入探讨如何将大模型的流式生成能力从单体中拆分为可独立部署和扩容的微服务。