数据模型、分布式任务与可靠性

核心结论：PostgreSQL 保证业务数据可靠落库，Redis 负责短期状态和并发控制，RocketMQ 负责异步任务分发，MinIO 保存文件，Milvus 保存向量索引。任务执行必须以数据库中的持久化状态为准，不能以缓存或消息是否存在为准。

无人机 Copilot 不是一次普通的问答请求。用户说“这块地今天巡一遍消防通道”后，系统要生成任务草稿，等待用户确认，再创建正式任务，检查设备和空域，向无人机平台下发指令，持续接收遥测和告警，最终生成报告。

这条链路跨越多个服务和外部系统。任何一步都可能失败。

例如：

• 草稿已经创建，但前端没有收到响应，用户再次点击。
• 任务已经写入数据库，但消息没有成功发送。
• RocketMQ 重复投递了同一条任务消息。
• 无人机平台已经接收任务，但调用方超时，不知道任务是否下发成功。
• 任务执行过程中服务重启，内存中的状态丢失。
• 任务已经结束，但报告文件还没有上传成功。

这些问题不能靠捕获异常解决，需要从数据模型、状态机、幂等、消息投递和补偿机制一起设计。

本章事实边界

• 已知事实： 现有无人机实战平台具备设备接入、航线规划、任务下发、遥测回传和告警处置能力。新增 Copilot 负责理解用户目标、生成任务草稿、编排业务调用、提供异常处置建议和生成报告。技术栈为 Java 21、Spring Boot 3、Spring AI、Spring AI Alibaba、PostgreSQL、Redis、RocketMQ、Milvus、MinIO、Vue 3、Cesium 和 SSE。
• 目标设计： 本章定义数据表、任务状态机、事务边界、消息可靠性、幂等和故障恢复方案。这些是拟采用的工程设计，不代表现有平台已经实现。
• 验证结果： 暂无，本文只做方案设计。

这一章解决的是 Copilot 的业务执行底座，而不是大模型如何生成答案。

整体可以分成三个部分。

第一部分是业务数据层。 PostgreSQL 保存会话、消息、任务草稿、正式任务、航线、审批和审计记录。任务当前处于什么状态、由谁确认、是否获准执行，都要有持久化记录。

第二部分是可靠性层。 Redis 处理短期缓存、请求去重和分布式协调；RocketMQ 把任务创建、任务执行和报告生成解耦；事务 Outbox 解决数据库提交成功但消息发送失败的问题。

第三部分是文件和知识数据层。 MinIO 保存航线文件、图片、视频片段和任务报告。Milvus 保存设备手册、操作规程、历史案例等知识的向量索引，供 RAG 检索使用。

大模型可以提出执行计划，但不能直接决定任务已创建、已审批或已起飞。这些事实必须由业务服务校验并持久化。

3.1 数据关系

用户
 |
 +--> conversation_session 会话
 |       |
 |       +--> conversation_message 消息
 |
 +--> task_draft 任务草稿
          |
          +--> flight_route 航线
          |
          +--> approval_order 审批单
          |
          +--> drone_task 正式任务
                    |
                    +--> task_event_outbox 待发送事件
                    |
                    +--> task_audit_log 审计日志
                    |
                    +--> MinIO 任务产物
                    |
                    +--> RocketMQ 任务执行事件

这里有三个重要关系。

• 会话和消息是一对多关系。一条消息属于一个会话，用户的原始指令和模型生成内容都可以追溯。
• 草稿和正式任务是不同的业务实体。草稿允许编辑，正式任务进入受控的执行状态机。
• 航线、审批单和正式任务通过明确的外键或业务关联字段连接。审批通过不等于任务已经执行，任务创建也不等于无人机已经起飞。

草稿可能经过多次修改。正式任务创建时，应保存确认版本对应的航线、区域、设备和参数快照，避免草稿后续变化影响已经确认的任务。

3.2 四类存储的职责

哪些数据不能只放 Redis？

正式任务及其状态、用户确认记录、审批结果、审计日志、消息发送记录和需要长期追溯的执行结果，都不能只存 Redis。

Redis 可能因淘汰、重启、故障转移或过期策略丢失数据。即使开启持久化，也不应将它作为任务状态和审批记录的唯一事实来源。

Milvus 也不能代替 PostgreSQL。向量相似度检索结果不是正式业务记录，向量索引可以重建，审批结果和任务状态不能靠重新检索恢复。

4.1 从草稿确认到任务执行

1. 用户确认草稿前端提交 draftId、版本号和幂等键
2. 校验草稿检查归属、版本、设备、区域和参数
3. 校验审批检查空域、权限及必需审批是否通过
4. 创建正式任务PostgreSQL 事务内写入任务、审计和 Outbox
5. 提交事务任务与待发送事件同时持久化
6. 发布任务事件Outbox 发布器将事件发送到 RocketMQ
7. 消费任务事件消费者按事件ID和任务状态幂等处理
8. 执行前复核检查设备在线、空域有效性和执行条件
9. 调用无人机平台使用平台支持的幂等或任务查询机制下发
10. 回写执行结果更新任务状态、外部任务ID和审计记录
11. 持续接收事件遥测、告警和完成事件驱动状态更新
12. 生成任务报告生成文件并上传 MinIO，记录报告元数据

4.2 同步和异步的分界

同步创建草稿，异步执行任务。

用户提交目标后，系统需要尽快返回一个可以查看、修改的草稿。此时同步执行的工作包括：

1. 校验用户身份和输入格式。
2. 调用模型解析目标，提取巡检区域、时间要求和任务意图。
3. 查询设备、空域等必要信息，生成航线和任务建议。
4. 保存草稿、航线版本及相关消息，返回 draftId。

这里的同步是接口调用方式，不意味着所有操作必须放在一个数据库事务里。模型调用、地图查询和航线规划可能耗时较长，不能把数据库事务从头到尾一直打开。

建议先完成必要的只读查询和模型处理，再用短事务保存草稿及其关联记录。若草稿需要分阶段生成，可以持久化草稿状态，例如 GENERATING、READY 和 FAILED，由前端轮询或通过 SSE 接收更新。

用户确认后，接口同步完成权限、版本、审批条件检查和正式任务创建，并返回任务 ID。真正的设备下发、长时间执行、告警处理和报告生成由异步流程完成。

需要注意：创建任务成功不代表下发成功，更不代表无人机已经起飞。

4.3 消息队列怎么选

本项目已有 RocketMQ，建议继续使用它承载任务事件，不必为了引入新技术而再增加一种消息队列。

• 任务创建、任务执行、报告生成等流程使用持久化业务事件。
• 任务状态更新通过明确的事件类型传递，例如 TaskCreated、TaskDispatchResult、TaskCompleted。
• 消息携带 eventId、taskId、eventType、occurredAt 和必要的版本信息。
• 消费端不能假设消息只会送达一次，应按至少一次投递的思路设计幂等。

RocketMQ 负责传递事件，不负责替代业务状态机，也不能单独保证数据库和无人机平台之间的事务一致性。

5.1 故障场景与处理方式

5.2 事务 Outbox

问题在于，数据库事务和 RocketMQ 发送通常不属于同一个本地事务。

假设程序先提交任务，再发送消息。如果数据库提交成功、消息发送失败，任务就可能永远无人执行。反过来，如果先发消息再提交事务，消费者可能收到一个数据库里还不存在的任务。

解决办法是使用事务 Outbox。

在同一个 PostgreSQL 本地事务中完成：

1. 插入 drone_task。
2. 插入 task_audit_log。
3. 插入 task_event_outbox。

三者一起提交或一起回滚。

事务提交后，独立的发布器扫描未发送的 Outbox 记录，发送到 RocketMQ。收到 Broker 的发送成功确认后，再将记录标记为已发送。

如果发布器在发送成功后、更新 Outbox 状态前崩溃，重启后可能再次发送。因此 Outbox 解决的是“业务数据和待发送事件不会分离”的问题，并不保证消息只发送一次。消费者仍然必须幂等。

5.3 状态变更如何保证原子性

任务状态不能只靠 Java 对象中的字段更新。每次重要状态变化都要在 PostgreSQL 中完成条件更新，并同步写入审计记录和后续 Outbox 事件。

例如，任务从 CREATED 进入 QUEUED：

BEGIN;

UPDATE drone_task
SET status = 'QUEUED',
    version = version + 1,
    updated_at = now()
WHERE id = :task_id
  AND status = 'CREATED'
  AND version = :expected_version;

-- 必须检查 UPDATE 影响行数是否为 1。
-- 同一事务内写审计记录及 Outbox 事件。

COMMIT;

如果更新行数为零，说明状态或版本已经变化，当前操作不能继续按成功处理。

这里使用的是乐观锁。version 可以防止两个消费者同时基于旧状态覆盖新状态。对于必须互斥的短临界区，也可以使用 SELECT ... FOR UPDATE，但不能持锁等待模型响应、消息投递或无人机执行。

5.4 分布式事务的边界

不能为了追求“所有服务同时成功”就把整个飞行任务包进一个长事务。无人机可能已经接收任务甚至起飞，数据库回滚无法让它回到起飞前的物理状态。

5.5 超时、重试、死信与补偿

• 超时： 给数据库访问、模型调用、设备查询和外部 API 设置独立超时。超时只代表调用方没有及时得到结果，不代表远端操作一定失败。
• 重试： 对明确可重试的网络错误、限流和临时服务不可用，使用有限次数、指数退避和随机抖动。参数应根据外部平台限制配置，不预设未经验证的固定次数。
• 死信： 消息超过重试策略后进入死信处理流程，保留原始 eventId、taskId、错误信息和重试记录，供人工排查或经过审核后重新投递。
• 补偿： 根据实际执行结果修复业务状态。例如任务创建成功但下发失败，可以转为 DISPATCH_FAILED，经过检查后重新下发；报告上传失败则重新生成或上传。补偿不是把所有数据回滚到最初状态。
• 不确定结果： 下发超时后先查询无人机平台是否已有对应任务。若外部平台既不支持幂等键，也无法查询请求结果，就不能安全地自动重发起飞类指令，应进入人工核查或受控恢复流程。

6.1 核心表设计

表之间通过外键或明确的业务关联字段建立关系。对于正式任务，还应保存确认时使用的草稿版本、航线版本、设备标识和关键执行参数快照。

审批单可以在任务创建时建立，也可以在审批前建立草稿关联的审批记录。具体采用哪种模式取决于现有审批流程，但审批状态必须明确关联到最终执行的任务版本，不能仅根据一个通用的“审批通过”标记放行。

6.2 任务表 DDL

以下是目标设计中的 PostgreSQL 表结构示例：

CREATE TABLE drone_task (
    id                  UUID PRIMARY KEY,
    draft_id            UUID NOT NULL,
    draft_version       INTEGER NOT NULL,

    user_id             VARCHAR(128) NOT NULL,
    device_id           VARCHAR(128) NOT NULL,
    route_id            UUID NOT NULL,

    status              VARCHAR(32) NOT NULL,
    version             BIGINT NOT NULL DEFAULT 0,

    idempotency_key     VARCHAR(160) NOT NULL,
    external_task_id    VARCHAR(160),

    dispatch_attempts   INTEGER NOT NULL DEFAULT 0,
    last_error_code     VARCHAR(80),
    last_error_message  TEXT,

    created_at          TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at          TIMESTAMPTZ NOT NULL DEFAULT now(),

    CONSTRAINT uq_task_idempotency
        UNIQUE (user_id, idempotency_key),

    CONSTRAINT ck_task_status CHECK (
        status IN (
            'CREATED',
            'QUEUED',
            'VALIDATING',
            'DISPATCHING',
            'EXECUTING',
            'COMPLETED',
            'DISPATCH_FAILED',
            'FAILED',
            'CANCEL_PENDING',
            'CANCELLED',
            'MANUAL_REVIEW'
        )
    )
);

CREATE INDEX idx_drone_task_status_created
    ON drone_task (status, created_at);

CREATE INDEX idx_drone_task_device_status
    ON drone_task (device_id, status);

这张表保存任务当前状态。状态变更历史另存于 task_audit_log，不要只保留最后一次错误。

external_task_id 保存无人机平台返回的任务标识。它与本系统的 id 不同，便于通过外部接口查询真实执行状态。

6.3 幂等键设计

幂等键需要按业务操作区分，不能所有接口共用一个键。

前端生成的请求键必须在同一次用户操作的重试中保持不变。不能每次点击都随机生成新键，否则数据库无法判断它们是否属于同一个请求。

用户再次确认同一个草稿版本时，服务端先查询已有操作结果。若任务已经创建，就返回原任务，而不是再创建一条。

6.4 消费端如何做到幂等

消费者不能采用“收到消息就直接执行”的简单逻辑。

建议将以下操作放在同一个 PostgreSQL 事务内：

1. 尝试插入 processed_event。
2. 如果唯一约束冲突，说明该消费者已经处理过该事件，直接结束。
3. 根据任务当前状态和 version 检查事件是否合法。
4. 更新任务状态，写入审计记录和后续 Outbox 事件。
5. 提交事务后再确认消息消费成功。

如果消费者需要调用无人机平台，不能把外部调用和本地数据库事务假设为原子操作。可以先持久化 DISPATCHING 状态及操作记录，再调用外部平台，最后根据明确的结果回写状态。

如果进程在外部下发成功后崩溃，恢复程序必须查询外部任务或利用平台幂等键识别已完成的操作。不能只看本地 processed_event 就认定设备端动作一定完成。

6.5 Java 服务接口

public interface TaskApplicationService {

    TaskCreatedResult confirmDraft(
        ConfirmDraftCommand command
    );

    void dispatchTask(UUID taskId);

    void reconcileTask(UUID taskId);
}

public record ConfirmDraftCommand(
    UUID draftId,
    int draftVersion,
    String idempotencyKey,
    String userId
) {}

public record TaskCreatedResult(
    UUID taskId,
    String status
) {}

public interface DronePlatformClient {

    DispatchResult dispatch(
        DispatchCommand command
    );

    ExternalTaskStatus queryTask(
        String externalTaskId
    );
}

confirmDraft 负责校验、创建任务和写入 Outbox。dispatchTask 由异步消费者或任务执行器触发。reconcileTask 用于对账和恢复不确定状态。

实际实现中，应将数据库事务放在独立的事务服务方法中，例如 TaskTransactionService.createTaskAndOutbox()，避免同一个 Spring Bean 内部自调用导致 @Transactional 代理失效。

还要注意，@Transactional 只能管理配置的事务资源，不能自动把 PostgreSQL、RocketMQ、MinIO 和无人机平台变成一个原子事务。

7.1 任务状态机

任务状态必须有明确的合法流转，不能允许任意字符串之间互相跳转。

CREATED
   |
   v
QUEUED --> VALIDATING
                |
                v
           DISPATCHING
            /       \
           v         v
       EXECUTING  DISPATCH_FAILED
           |          |
           v          +--> MANUAL_REVIEW
       COMPLETED

这是简化示意。正式实现还要覆盖校验失败、执行失败、取消申请、取消成功和人工核查等分支。

需要明确几个边界：

• CREATED：正式任务已在数据库创建，但不代表已开始执行。
• QUEUED：任务已进入待执行流程。
• DISPATCHING：系统正在向外部平台提交任务。
• EXECUTING：已有可靠证据表明外部任务正在执行。
• COMPLETED：收到可信的完成结果并完成必要校验。
• DISPATCH_FAILED：下发明确失败，可以根据原因判断是否允许重试。
• MANUAL_REVIEW：执行结果不确定，需要对账或人工处理。

状态更新必须检查前置状态和版本号。迟到的遥测或重复的完成事件不能把 COMPLETED 改回 EXECUTING。

7.2 超时恢复和补偿

后台应设置定时对账任务，扫描长时间停留在 QUEUED、DISPATCHING 或 EXECUTING 的记录。

对账逻辑需要结合：

• 数据库当前状态和最后更新时间。
• RocketMQ 的投递和消费记录。
• 无人机平台的外部任务状态。
• 最近一次遥测和告警时间。
• 已记录的下发请求及响应。

例如，任务长时间停留在 DISPATCHING，不能直接判定下发失败。应先查询外部平台。如果平台显示任务已经创建，就补齐外部任务 ID 并推进状态；如果平台明确未创建，才考虑重新提交；如果无法确认，就转入人工核查。

取消任务也一样。数据库里改成 CANCELLED 不等于无人机真的停止飞行。必须根据外部平台的取消结果或设备状态确认取消成功。

7.3 审计与安全

审计记录应覆盖用户确认、审批通过或拒绝、任务创建、任务下发、取消、失败、人工介入和最终完成等关键动作。

至少记录操作人、操作类型、关联任务、前后状态、时间、请求关联 ID 和结果摘要。审计日志应限制修改权限，不能让普通业务接口随意删除。

无人机执行还需要额外的安全门禁：

• 用户只能操作有权限访问的设备和区域。
• 执行前重新检查空域、设备状态和有效审批，不能完全信任草稿生成时的检查结果。
• 大模型输出的设备 ID、坐标和动作参数必须由服务端校验。
• 涉及起飞、返航、降落等高风险动作时，必须符合既定的人工确认和安全策略。
• 遥测和告警可能重复、乱序或延迟，需要结合事件时间、任务 ID 和序列信息判断。
• SSE 断开只影响实时通知，不应影响任务执行。前端重连后应通过任务查询接口获取持久化状态。

本章方案不能只靠单元测试证明可靠。需要覆盖数据库、消息队列、外部平台模拟器和服务重启等场景。

8.1 数据一致性测试

• 创建任务时人为触发数据库异常，确认任务、审计和 Outbox 同时回滚。
• 任务提交成功后关闭 RocketMQ，确认 Outbox 记录仍在，恢复后可以补发。
• 模拟消息发送成功后发布器崩溃，确认重复投递不会重复创建任务。
• 并发提交相同幂等键，确认只创建一条正式任务。
• 两个消费者同时更新同一任务，确认只有合法的状态转换能够成功。

8.2 执行恢复测试

• 模拟无人机平台已接收任务，但 HTTP 响应超时。
• 模拟服务在下发成功后、写回 external_task_id 前重启。
• 模拟遥测乱序、告警重复和任务完成事件延迟。
• 模拟取消请求超时，确认系统不会提前认定任务已取消。
• 模拟报告上传失败，确认任务结果和报告生成状态可以分别恢复。

8.3 验证标准

以上是目标设计的验收标准，暂无实际测试结果，不能宣称已通过验证。后续应使用集成测试、故障注入和外部平台模拟器执行验证，再记录实际结果和发现的问题。

问答一：为什么任务状态放 PostgreSQL，而不是 Redis？

答： 因为任务状态是业务事实，需要长期保存、追溯和恢复。Redis 更适合缓存、限流和短期协调，不能作为任务和审批记录的唯一存储。即使 Redis 开启持久化，也不应该把核心业务状态完全依赖在缓存上。任务状态以 PostgreSQL 为准，Redis 丢失后可以重新建立缓存。

问答二：任务已经保存到数据库，但消息没有发送成功，怎么办？

答： 我会使用事务 Outbox。创建任务时，在同一个 PostgreSQL 事务里写入任务、审计记录和待发送事件。事务提交后，由独立发布器把事件发送到 RocketMQ。发送失败就继续扫描和重试。因为发送成功后仍可能发生重复投递，所以消费端还要通过事件 ID、任务状态和版本号实现幂等。Outbox 解决数据库与待发送事件的一致性，不代表整个分布式流程只执行一次。

问答三：无人机下发请求超时，为什么不能直接重试？

答： 因为超时只能说明调用方没有及时收到结果，不能证明设备端没有接收任务。如果平台已经创建任务，直接重试可能造成重复任务，甚至重复执行高风险动作。我会先查询外部任务状态，或者使用平台支持的幂等键确认原请求结果。只有明确未执行、满足安全条件时才重试。结果无法确认时，进入对账或人工核查，而不是盲目重新起飞。