构建基于Vue、Valtio与Spring Boot的PWA离线优先同步队列

全栈开发

文章字数: 3.8k

阅读时长: 16 分

标准的PWA Service Worker擅长缓存静态资源与GET请求，但在处理用户离线期间产生的复杂数据变更（创建、更新、删除）时，其能力就显得捉襟见肘。一个常见的错误是，开发者试图在Service Worker中拦截POST/PUT请求并将其暂存，但这很快会演变成一场状态管理的噩梦，尤其是在操作之间存在依赖关系，或需要处理服务端冲突时。问题的核心在于，我们需要一个比简单请求拦截更健壮的机制——一个应用层的、具备持久化与事务性的命令队列。

这次复盘的目标，就是从零构建一套离线优先的数据同步方案。前端使用Vue 3、Valtio进行状态管理，并借助IndexedDB持久化命令队列；后端则由Spring Boot提供一个幂等的、支持批量处理的同步端点。整套机制必须经过Vitest的严格测试，以确保在各种网络异常场景下的数据一致性。

初步构想：从状态变更到命令队列

放弃直接修改本地状态并期望同步成功的乐观模式。我们将采用一种更审慎的架构：任何对数据的修改操作，都不直接作用于“真实”状态，而是被封装成一个“命令”（Command）对象，推入一个持久化的队列中。

// src/services/command.types.ts

// 每个命令必须是独立的、可序列化的操作单元
export interface Command {
  // 客户端生成的唯一ID，用于后端幂等性校验
  id: string; 
  // 操作类型，例如 'CREATE_TASK', 'UPDATE_TASK_STATUS'
  type: string;
  // 操作相关的数据负载
  payload: Record<string, any>;
  // 创建时间戳，用于调试和可能的冲突解决
  timestamp: number;
  // 命令处理过程中的元数据
  metadata: {
    // 尝试同步的次数
    retries: number;
  };
}

UI层看到的数据，将是“基础数据”与“未同步命令队列”共同作用下的一个投影（Projected State）。例如，用户离线创建了一个任务，UI应立即显示这个新任务，尽管它尚未被服务器确认。Valtio的代理模型非常适合实现这种状态的组合与响应。

技术选型决策与理由

Vue.js 3 & Composition API: 提供了构建响应式UI的强大基础。Composition API能让我们将同步逻辑、状态管理逻辑聚合到独立的use函数中，保持组件的整洁。
Valtio: 在这个场景下，Valtio比Vuex或Pinia更具优势。它的极简API和基于Proxy的自动追踪机制，让我们能像操作普通JavaScript对象一样修改状态，而无需定义繁琐的mutations或actions。我们可以轻松地创建一个store，其中包含基础数据数组和命令队列数组，Valtio会自动处理对这两个数组的任何修改并触发UI更新。这种简单性在处理复杂的、临时的UI状态时尤其重要。
Spring Boot: 后端需要一个稳定、类型安全且易于构建RESTful API的框架。Spring Boot的依赖注入、JPA/Hibernate以及强大的Web框架（Spring MVC）使其成为理想选择。我们将重点实现一个用于处理批量命令的幂等Controller。
IndexedDB (via Dexie.js): 为了保证离线操作的持久性，命令队列必须存储在浏览器中。IndexedDB是标准选择，但其原生API相当繁琐。Dexie.js提供了一个优雅的Promise-based的API封装，大大简化了数据库操作。
Vitest: 同步逻辑是应用的核心，也是最容易出错的部分。我们必须对它进行详尽的单元测试。Vitest的快速、ESM-native特性，以及与Vite的无缝集成，使其成为现代前端项目的首选。其内置的mocking和spying能力对于模拟网络状态和依赖项至关重要。

步骤化实现：构建同步引擎

1. 前端状态管理 (Valtio)

首先定义我们的全局状态。我们需要跟踪网络状态、同步状态、任务数据以及命令队列。

// src/store/appStore.ts
import { proxy } from 'valtio';
import { Command } from '@/services/command.types';

export interface Task {
  id: string; // 后端ID或临时前端ID
  title: string;
  completed: boolean;
  // 标记该任务是否仅存在于本地（由一个未同步的CREATE命令生成）
  isLocal?: boolean; 
}

interface AppState {
  tasks: Task[];
  commandQueue: Command[];
  isOnline: boolean;
  // 同步过程的状态机：'IDLE', 'SYNCING', 'ERROR'
  syncState: 'IDLE' | 'SYNCING' | 'ERROR'; 
}

// Valtio的魅力在于，这就是全部定义。
// 直接修改这个proxy对象，任何使用它的Vue组件都会自动更新。
export const store = proxy<AppState>({
  tasks: [],
  commandQueue: [],
  isOnline: navigator.onLine,
  syncState: 'IDLE',
});

// 监听网络状态变化
window.addEventListener('online', () => store.isOnline = true);
window.addEventListener('offline', () => store.isOnline = false);

2. 命令队列的持久化 (Dexie.js)

接下来，创建数据库服务来管理IndexedDB。

// src/services/db.ts
import Dexie, { Table } from 'dexie';
import { Command } from './command.types';

export class CommandDB extends Dexie {
  commands!: Table<Command>;

  constructor() {
    super('MyAppDB');
    this.version(1).stores({
      // 'id'是主键
      commands: 'id, timestamp', 
    });
  }
}

export const db = new CommandDB();

3. 核心同步服务 (SyncService)

这是整个机制的大脑。它负责监听网络变化、从IndexedDB加载命令、与后端通信，并更新本地状态。

// src/services/SyncService.ts
import { ref, watch } from 'vue';
import { store } from '@/store/appStore';
import { db } from './db';
import { Command } from './command.types';
import { v4 as uuidv4 } from 'uuid';

// 一个常见的错误是在组件内部实现同步逻辑。
// 必须将其抽离到一个独立的、与UI无关的服务中。
class SyncService {
  private isSyncing = ref(false);

  constructor() {
    this.initialize();
  }

  private async initialize() {
    // 应用启动时，从IndexedDB加载未处理的命令到Valtio store
    await this.loadQueueFromDB();

    // 使用Vue的watch来响应式地触发同步
    watch(
      () => store.isOnline,
      (isOnline, wasOffline) => {
        // 关键逻辑：仅在从离线恢复到在线时触发
        if (isOnline && wasOffline === false) {
          console.log('Connection restored. Starting synchronization...');
          this.triggerSync();
        }
      },
      { immediate: true } // 立即执行一次，处理启动时就在线的情况
    );
  }

  private async loadQueueFromDB() {
    try {
      const commands = await db.commands.orderBy('timestamp').toArray();
      store.commandQueue = commands;
    } catch (error) {
      console.error('Failed to load command queue from IndexedDB:', error);
      store.syncState = 'ERROR';
    }
  }

  // 这是暴露给应用层（例如Vue组件）的唯一方法
  public async addCommand(type: string, payload: Record<string, any>): Promise<void> {
    const command: Command = {
      id: uuidv4(),
      type,
      payload,
      timestamp: Date.now(),
      metadata: { retries: 0 },
    };

    // 1. 持久化到IndexedDB
    try {
      await db.commands.add(command);
      // 2. 更新内存中的状态 (Valtio)
      store.commandQueue.push(command);
      // 3. 立即尝试同步（如果在线）
      this.triggerSync();
    } catch (error) {
        console.error('Failed to add command to queue:', error);
    }
  }

  public async triggerSync() {
    if (!store.isOnline || this.isSyncing.value || store.commandQueue.length === 0) {
      return;
    }

    this.isSyncing.value = true;
    store.syncState = 'SYNCING';

    // 复制队列以防在同步过程中有新命令加入
    const commandsToSync = [...store.commandQueue];

    try {
      // 在真实项目中，这里应该有更复杂的批量和错误处理逻辑
      // 例如，一次只发送N个命令，或者对特定错误进行重试
      const response = await fetch('/api/sync/batch', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify(commandsToSync),
      });

      if (!response.ok) {
        throw new Error(`Sync failed with status: ${response.status}`);
      }
      
      const results = await response.json();
      
      // 处理结果，例如后端可能返回一些命令失败
      // 这里简化为全部成功
      await db.commands.bulkDelete(commandsToSync.map(c => c.id));
      store.commandQueue = store.commandQueue.filter(
        c => !commandsToSync.some(synced => synced.id === c.id)
      );

      console.log('Sync successful.');
      store.syncState = 'IDLE';
      // 如果同步后队列仍有内容，则再次触发
      if (store.commandQueue.length > 0) {
        this.triggerSync();
      }

    } catch (error) {
      console.error('Synchronization failed:', error);
      store.syncState = 'ERROR';
      // 此处可以加入重试逻辑，例如增加命令的retries计数
    } finally {
      this.isSyncing.value = false;
    }
  }
}

// 单例模式
export const syncService = new SyncService();

4. 后端幂等端点 (Spring Boot)

后端的关键是实现一个接受命令列表并保证幂等性的端点。幂等性通过检查命令的客户端ID是否已被处理过来实现。

// CommandDto.java
// 使用Record简化DTO定义
public record CommandDto(
    @NotNull String id,
    @NotBlank String type,
    @NotNull Map<String, Object> payload,
    long timestamp
) {}

// SyncRequest.java
public record SyncRequest(
    @NotEmpty List<CommandDto> commands
) {}

// ProcessedCommand.java (JPA Entity)
@Entity
public class ProcessedCommand {
    @Id
    private String commandId;
    private Instant processedAt;
    // ... constructors, getters, setters
}


// SyncController.java
@RestController
@RequestMapping("/api/sync")
public class SyncController {

    private final SyncService syncService;

    public SyncController(SyncService syncService) {
        this.syncService = syncService;
    }

    @PostMapping("/batch")
    public ResponseEntity<Map<String, String>> handleBatchSync(@RequestBody @Valid SyncRequest request) {
        // 在真实项目中，这里的处理应该是异步的，并返回一个任务ID
        // 为了简化，我们这里同步处理
        syncService.processCommands(request.commands());

        // 理想情况下，响应应包含每个命令的处理结果
        return ResponseEntity.ok(Map.of("status", "processed"));
    }
}


// SyncService.java
@Service
@Slf4j
public class SyncService {

    private final ProcessedCommandRepository processedCommandRepository;
    private final TaskService taskService; // 假设有处理任务的业务服务

    public SyncService(ProcessedCommandRepository processedCommandRepository, TaskService taskService) {
        this.processedCommandRepository = processedCommandRepository;
        this.taskService = taskService;
    }

    // 关键：@Transactional保证批处理的原子性
    @Transactional
    public void processCommands(List<CommandDto> commands) {
        // 按时间戳排序，确保操作顺序
        commands.sort(Comparator.comparingLong(CommandDto::timestamp));

        List<String> commandIds = commands.stream().map(CommandDto::id).collect(Collectors.toList());
        
        // 幂等性检查：批量查询已存在的ID
        Set<String> alreadyProcessed = processedCommandRepository.findAllById(commandIds)
                .stream()
                .map(ProcessedCommand::getCommandId)
                .collect(Collectors.toSet());

        List<ProcessedCommand> newProcessedCommands = new ArrayList<>();

        for (CommandDto command : commands) {
            if (alreadyProcessed.contains(command.id())) {
                log.warn("Skipping already processed command: {}", command.id());
                continue;
            }

            // 真正的业务逻辑分发
            // 这里的坑在于：如果一个命令失败，整个事务会回滚。
            // 需要根据业务需求决定是全部回滚还是部分成功。
            // 使用try-catch可以实现部分成功，但会破坏事务原子性。
            try {
                switch (command.type()) {
                    case "CREATE_TASK":
                        taskService.createTask(command.payload());
                        break;
                    case "UPDATE_TASK_STATUS":
                        taskService.updateTaskStatus(command.payload());
                        break;
                    // ... more command types
                    default:
                        log.error("Unknown command type: {}", command.type());
                        // 抛出异常以回滚事务
                        throw new IllegalArgumentException("Unknown command type");
                }
                
                newProcessedCommands.add(new ProcessedCommand(command.id(), Instant.now()));

            } catch (Exception e) {
                log.error("Failed to process command {}: {}", command.id(), e.getMessage());
                // 决定是否中止整个批处理
                throw new RuntimeException("Failed to process command batch", e);
            }
        }
        
        // 批量保存已处理的命令ID
        processedCommandRepository.saveAll(newProcessedCommands);
    }
}

这个后端实现了一个基础但健壮的模式：通过在事务中先检查再执行，最后记录ID的方式，确保了即使客户端重复发送同一批命令，业务逻辑也只会执行一次。

5. 单元测试 (Vitest)

现在来测试最关键的前端SyncService。我们需要模拟fetch和Dexie。

// src/services/SyncService.test.ts
import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest';
import { store } from '@/store/appStore';
import { db } from './db';
import { syncService } from './SyncService'; // 导入单例

// Mock Dexie
vi.mock('./db', () => {
  const mockDb = {
    commands: {
      add: vi.fn(),
      bulkDelete: vi.fn(),
      orderBy: vi.fn().mockReturnThis(),
      toArray: vi.fn(),
    },
  };
  return { db: mockDb };
});

// Mock fetch
const mockFetch = vi.fn();
global.fetch = mockFetch;

describe('SyncService', () => {
  beforeEach(() => {
    // 重置所有mock和store状态
    vi.clearAllMocks();
    store.isOnline = true;
    store.commandQueue = [];
    store.syncState = 'IDLE';
  });
  
  it('should not sync when offline', async () => {
    store.isOnline = false;
    store.commandQueue.push({ id: '1', type: 'TEST', payload: {}, timestamp: Date.now(), metadata: { retries: 0 } });

    await syncService.triggerSync();
    
    expect(mockFetch).not.toHaveBeenCalled();
    expect(store.syncState).toBe('IDLE');
  });

  it('should trigger sync when coming online', async () => {
    // 模拟初始离线状态，并有命令在队列中
    store.isOnline = false;
    store.commandQueue.push({ id: '1', type: 'TEST', payload: {}, timestamp: Date.now(), metadata: { retries: 0 } });

    // 模拟网络恢复
    store.isOnline = true;

    // Vitest的watch触发可能需要一个tick
    await new Promise(resolve => setTimeout(resolve, 0));

    // 这里我们需要重新审视SyncService的实现，让watch的触发更易于测试
    // 一种改进是暴露一个方法来手动触发watch的回调
    // 但为了保持原样，我们假设watch已触发了triggerSync
    
    // 我们直接调用triggerSync来测试核心逻辑
    mockFetch.mockResolvedValueOnce({ ok: true, json: () => Promise.resolve({ status: 'ok' }) });
    await syncService.triggerSync();

    expect(mockFetch).toHaveBeenCalledTimes(1);
    expect(db.commands.bulkDelete).toHaveBeenCalledWith(['1']);
    expect(store.commandQueue).toHaveLength(0);
    expect(store.syncState).toBe('IDLE');
  });
  
  it('should handle fetch failure gracefully', async () => {
    const command = { id: '1', type: 'TEST', payload: {}, timestamp: Date.now(), metadata: { retries: 0 } };
    store.commandQueue.push(command);
    
    mockFetch.mockRejectedValueOnce(new Error('Network Error'));
    
    await syncService.triggerSync();
    
    expect(mockFetch).toHaveBeenCalledTimes(1);
    // 失败后，命令不应被删除
    expect(db.commands.bulkDelete).not.toHaveBeenCalled();
    expect(store.commandQueue).toHaveLength(1);
    expect(store.syncState).toBe('ERROR');
  });
  
  it('should add a command and trigger sync immediately if online', async () => {
    store.isOnline = true;
    
    mockFetch.mockResolvedValueOnce({ ok: true, json: () => Promise.resolve({ status: 'ok' }) });
    
    await syncService.addCommand('CREATE_TASK', { title: 'New Task' });
    
    expect(db.commands.add).toHaveBeenCalledTimes(1);
    expect(store.commandQueue).toHaveLength(1); // 此时还未同步完成
    
    // 等待异步的sync完成
    await new Promise(resolve => setTimeout(resolve, 0));

    expect(mockFetch).toHaveBeenCalledTimes(1);
    expect(db.commands.bulkDelete).toHaveBeenCalledTimes(1);
    expect(store.commandQueue).toHaveLength(0);
  });
});

这些测试覆盖了核心场景：离线/在线状态切换、同步成功与失败。在真实项目中，还需要测试更多边缘情况，如多个命令的批量处理、特定命令失败后的队列状态等。

最终成果的整合与展现

sequenceDiagram
    participant User
    participant VueComponent
    participant ValtioStore
    participant SyncService
    participant IndexedDB
    participant SpringBootAPI

    User->>VueComponent: Clicks 'Add Task' (Offline)
    VueComponent->>SyncService: addCommand('CREATE_TASK', {title: '...'})
    SyncService->>IndexedDB: Persists new command
    SyncService->>ValtioStore: Pushes command to queue
    ValtioStore-->>VueComponent: UI updates to show new task (optimistically)
    
    User->>VueComponent: Goes Online
    Note over SyncService: Network listener detects 'online' event
    SyncService->>SyncService: triggerSync()
    SyncService->>ValtioStore: Reads command queue
    SyncService->>SpringBootAPI: POST /api/sync/batch with commands
    SpringBootAPI->>SpringBootAPI: Processes commands (idempotent check)
    SpringBootAPI-->>SyncService: Returns success response
    SyncService->>IndexedDB: Deletes processed commands
    SyncService->>ValtioStore: Removes commands from queue
    ValtioStore-->>VueComponent: UI state confirmed (e.g., remove 'syncing' indicator)

这个流程图清晰地展示了从用户操作到数据最终一致的全过程。用户在离线时能无缝操作，所有变更都被可靠地记录下来。一旦网络恢复，同步服务会自动在后台完成工作，对用户体验的影响降到最低。

遗留问题与未来迭代

当前这套方案虽然健壮，但仍有其局限性。

冲突解决机制过于简单：目前的实现是“客户端优先”，后端盲目执行命令。如果用户在设备A离线修改了任务，而设备B在线修改了同一个任务，当设备A上线同步时，可能会覆盖设备B的修改。更复杂的场景需要引入版本号（ETags）、时间戳或甚至是CRDTs（无冲突复制数据类型）来进行更智能的冲突合并。
命令依赖性：该架构假设所有命令都是独立的。但如果用户离线先创建一个任务，然后立即更新这个任务的状态，这两个命令之间就存在依赖。后端必须按顺序处理它们。当前的排序是基于客户端时间戳，这在多设备场景下并不可靠。一个改进是引入向量时钟或让客户端在命令中明确声明其依赖的前一个命令ID。
错误处理与重试策略：简单的“失败即停止”策略不够 resilient。对于可恢复的错误（如服务器临时不可用503），应实现带指数退避的重试逻辑。对于永久性错误（如业务校验失败400），应将命令移至一个“死信队列”，并通知用户手动干预。
大规模队列性能：如果用户长时间离线并产生大量命令，一次性将整个队列发送到后端可能会导致请求体过大或处理时间过长。应实现批处理（Pagination）机制，每次只同步队列中的前N个命令。

Vue.js PWA Spring Boot Valtio Vitest

构建基于Cloud Functions的Trino异步查询网关并集成Relay与Headless UI

2023-10-27 后端架构

Google Cloud Functions Relay Trino Headless UI Serverless

构建从Solid.js到containerd的跨异步边界分布式追踪系统

2023-10-27 可观测性

Solid.js Kafka containerd Zipkin