PSI进销存系统离线模式与数据同步

发布时间：2025-06-22 | 分类：技术实现

引言

在实际业务场景中，网络不稳定或完全没有网络的情况经常出现。PSI 进销存系统支持离线模式，允许用户在离线状态下继续进行业务操作，待网络恢复后自动同步数据。本文将介绍离线模式的设计与实现。

离线模式架构

离线数据处理架构：

本地数据存储

使用 IndexedDB 实现本地数据存储：

// 本地数据库管理器
class LocalDatabase {
  constructor(dbName, version) {
    this.dbName = dbName;
    this.version = version;
    this.db = null;
  }

  // 初始化数据库
  async initialize() {
    return new Promise((resolve, reject) => {
      const request = indexedDB.open(this.dbName, this.version);

      request.onerror = () => reject(request.error);
      request.onsuccess = () => {
        this.db = request.result;
        resolve(this.db);
      };

      request.onupgradeneeded = (event) => {
        const db = event.target.result;

        // 商品表
        if (!db.objectStoreNames.contains('products')) {
          const productStore = db.createObjectStore('products', { keyPath: 'id' });
          productStore.createIndex('category', 'category', { unique: false });
          productStore.createIndex('updatedAt', 'updatedAt', { unique: false });
        }

        // 库存表
        if (!db.objectStoreNames.contains('inventory')) {
          const inventoryStore = db.createObjectStore('inventory', { keyPath: 'productId' });
          inventoryStore.createIndex('warehouse', 'warehouse', { unique: false });
        }

        // 客户表
        if (!db.objectStoreNames.contains('customers')) {
          const customerStore = db.createObjectStore('customers', { keyPath: 'id' });
          customerStore.createIndex('type', 'type', { unique: false });
        }

        // 订单表
        if (!db.objectStoreNames.contains('orders')) {
          const orderStore = db.createObjectStore('orders', { keyPath: 'id' });
          orderStore.createIndex('status', 'status', { unique: false });
          orderStore.createIndex('createdAt', 'createdAt', { unique: false });
          orderStore.createIndex('syncStatus', 'syncStatus', { unique: false });
        }

        // 操作队列表
        if (!db.objectStoreNames.contains('operationQueue')) {
          const queueStore = db.createObjectStore('operationQueue', { keyPath: 'id', autoIncrement: true });
          queueStore.createIndex('status', 'status', { unique: false });
          queueStore.createIndex('createdAt', 'createdAt', { unique: false });
        }

        // 变更日志表
        if (!db.objectStoreNames.contains('changeLog')) {
          const logStore = db.createObjectStore('changeLog', { keyPath: 'id', autoIncrement: true });
          logStore.createIndex('entityType', 'entityType', { unique: false });
          logStore.createIndex('entityId', 'entityId', { unique: false });
          logStore.createIndex('timestamp', 'timestamp', { unique: false });
        }
      };
    });
  }

  // 添加记录
  async add(storeName, data) {
    return new Promise((resolve, reject) => {
      const transaction = this.db.transaction(storeName, 'readwrite');
      const store = transaction.objectStore(storeName);
      const request = store.add(data);

      request.onsuccess = () => resolve(request.result);
      request.onerror = () => reject(request.error);
    });
  }

  // 更新记录
  async put(storeName, data) {
    return new Promise((resolve, reject) => {
      const transaction = this.db.transaction(storeName, 'readwrite');
      const store = transaction.objectStore(storeName);
      const request = store.put(data);

      request.onsuccess = () => resolve(request.result);
      request.onerror = () => reject(request.error);
    });
  }

  // 查询记录
  async get(storeName, key) {
    return new Promise((resolve, reject) => {
      const transaction = this.db.transaction(storeName, 'readonly');
      const store = transaction.objectStore(storeName);
      const request = store.get(key);

      request.onsuccess = () => resolve(request.result);
      request.onerror = () => reject(request.error);
    });
  }

  // 查询所有记录
  async getAll(storeName, indexName, queryValue) {
    return new Promise((resolve, reject) => {
      const transaction = this.db.transaction(storeName, 'readonly');
      const store = transaction.objectStore(storeName);

      let request;
      if (indexName && queryValue !== undefined) {
        const index = store.index(indexName);
        request = index.getAll(queryValue);
      } else {
        request = store.getAll();
      }

      request.onsuccess = () => resolve(request.result);
      request.onerror = () => reject(request.error);
    });
  }

  // 删除记录
  async delete(storeName, key) {
    return new Promise((resolve, reject) => {
      const transaction = this.db.transaction(storeName, 'readwrite');
      const store = transaction.objectStore(storeName);
      const request = store.delete(key);

      request.onsuccess = () => resolve();
      request.onerror = () => reject(request.error);
    });
  }

  // 清除所有数据
  async clear(storeName) {
    return new Promise((resolve, reject) => {
      const transaction = this.db.transaction(storeName, 'readwrite');
      const store = transaction.objectStore(storeName);
      const request = store.clear();

      request.onsuccess = () => resolve();
      request.onerror = () => reject(request.error);
    });
  }
}

// 本地数据库单例
const localDB = new LocalDatabase('PSI_Offline_DB', 1);
await localDB.initialize();

离线操作队列

缓存离线操作请求：

// 离线操作管理器
class OfflineOperationManager {
  constructor(localDB, syncService) {
    this.localDB = localDB;
    this.syncService = syncService;
    this.isOnline = navigator.onLine;

    this.init();
  }

  init() {
    // 监听网络状态变化
    window.addEventListener('online', () => this.onNetworkOnline());
    window.addEventListener('offline', () => this.onNetworkOffline());

    // 定期尝试同步
    setInterval(() => {
      if (this.isOnline) {
        this.processQueue();
      }
    }, 30000); // 每30秒尝试同步
  }

  // 网络恢复
  onNetworkOnline() {
    console.log('Network online, starting sync...');
    this.isOnline = true;
    this.processQueue();
  }

  // 网络断开
  onNetworkOffline() {
    console.log('Network offline, switching to offline mode');
    this.isOnline = false;
    this.showOfflineNotification();
  }

  // 记录操作到队列
  async recordOperation(operation) {
    const queueItem = {
      id: this.generateId(),
      type: operation.type, // CREATE, UPDATE, DELETE
      entityType: operation.entityType, // order, product, customer
      entityId: operation.entityId,
      data: operation.data,
      timestamp: Date.now(),
      status: 'pending', // pending, syncing, completed, failed
      retryCount: 0,
      error: null
    };

    await this.localDB.add('operationQueue', queueItem);

    // 如果在线，立即尝试同步
    if (this.isOnline) {
      await this.processQueue();
    }

    return queueItem;
  }

  // 处理队列
  async processQueue() {
    const pendingOps = await this.localDB.getAll('operationQueue', 'status', 'pending');

    if (pendingOps.length === 0) {
      return;
    }

    console.log(`Processing ${pendingOps.length} pending operations`);

    for (const op of pendingOps) {
      // 标记为同步中
      op.status = 'syncing';
      await this.localDB.put('operationQueue', op);

      try {
        // 执行同步
        await this.syncService.syncOperation(op);

        // 标记为完成
        op.status = 'completed';
        op.completedAt = Date.now();
        await this.localDB.put('operationQueue', op);

        // 清理变更日志
        await this.localDB.delete('changeLog', op.entityId);
      } catch (error) {
        // 处理失败
        op.status = 'failed';
        op.error = error.message;
        op.retryCount++;
        await this.localDB.put('operationQueue', op);

        // 如果重试次数过多，记录警告
        if (op.retryCount >= 5) {
          console.error(`Operation ${op.id} failed after 5 retries`, error);
        }
      }
    }
  }

  // 创建订单（在线/离线通用）
  async createOrder(orderData) {
    const order = {
      id: this.generateId(),
      ...orderData,
      status: 'pending',
      syncStatus: this.isOnline ? 'synced' : 'pending',
      createdAt: Date.now(),
      updatedAt: Date.now(),
      clientId: this.generateId() // 客户端生成的临时ID
    };

    // 保存到本地
    await this.localDB.add('orders', order);

    if (!this.isOnline) {
      // 离线模式，记录操作
      await this.recordOperation({
        type: 'CREATE',
        entityType: 'order',
        entityId: order.id,
        data: order
      });
    }

    return order;
  }

  // 更新订单
  async updateOrder(orderId, updates) {
    const order = await this.localDB.get('orders', orderId);
    if (!order) {
      throw new Error('Order not found');
    }

    const updatedOrder = {
      ...order,
      ...updates,
      updatedAt: Date.now()
    };

    await this.localDB.put('orders', updatedOrder);

    if (!this.isOnline) {
      await this.recordOperation({
        type: 'UPDATE',
        entityType: 'order',
        entityId: orderId,
        data: updatedOrder
      });
    }

    return updatedOrder;
  }

  // 删除订单
  async deleteOrder(orderId) {
    await this.localDB.delete('orders', orderId);

    if (!this.isOnline) {
      await this.recordOperation({
        type: 'DELETE',
        entityType: 'order',
        entityId: orderId,
        data: { id: orderId }
      });
    }
  }

  // 查询待同步数量
  async getPendingCount() {
    const pending = await this.localDB.getAll('operationQueue', 'status', 'pending');
    const syncing = await this.localDB.getAll('operationQueue', 'status', 'syncing');
    return pending.length + syncing.length;
  }
}

数据同步服务

实现增量数据同步：

// 数据同步服务
class SyncService {
  constructor(apiBaseUrl, localDB) {
    this.apiBaseUrl = apiBaseUrl;
    this.localDB = localDB;
    this.lastSyncTime = localStorage.getItem('lastSyncTime') || 0;
  }

  // 同步单个操作
  async syncOperation(operation) {
    const endpoint = `${this.apiBaseUrl}/${operation.entityType}`;

    switch (operation.type) {
      case 'CREATE':
        return this.syncCreate(endpoint, operation);
      case 'UPDATE':
        return this.syncUpdate(endpoint, operation);
      case 'DELETE':
        return this.syncDelete(endpoint, operation);
      default:
        throw new Error(`Unknown operation type: ${operation.type}`);
    }
  }

  // 同步创建操作
  async syncCreate(endpoint, operation) {
    const response = await fetch(endpoint, {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify(operation.data)
    });

    if (!response.ok) {
      throw new Error(`Sync failed: ${response.status}`);
    }

    const serverData = await response.json();

    // 更新本地记录的服务器ID（用于后续关联）
    const localEntity = await this.localDB.get(operation.entityType + 's', operation.entityId);
    if (localEntity) {
      localEntity.serverId = serverData.id;
      localEntity.syncStatus = 'synced';
      localEntity.syncError = null;
      await this.localDB.put(operation.entityType + 's', localEntity);
    }

    return serverData;
  }

  // 同步更新操作
  async syncUpdate(endpoint, operation) {
    const response = await fetch(`${endpoint}/${operation.entityId}`, {
      method: 'PUT',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify(operation.data)
    });

    if (!response.ok) {
      throw new Error(`Sync failed: ${response.status}`);
    }

    // 更新本地状态
    const localEntity = await this.localDB.get(operation.entityType + 's', operation.entityId);
    if (localEntity) {
      localEntity.syncStatus = 'synced';
      localEntity.syncError = null;
      localEntity.updatedAt = Date.now();
      await this.localDB.put(operation.entityType + 's', localEntity);
    }
  }

  // 同步删除操作
  async syncDelete(endpoint, operation) {
    const response = await fetch(`${endpoint}/${operation.entityId}`, {
      method: 'DELETE'
    });

    // 忽略404错误（可能已被其他设备删除）
    if (!response.ok && response.status !== 404) {
      throw new Error(`Sync failed: ${response.status}`);
    }
  }

  // 增量同步获取最新数据
  async pullChanges() {
    const response = await fetch(
      `${this.apiBaseUrl}/sync?since=${this.lastSyncTime}`,
      { method: 'GET' }
    );

    if (!response.ok) {
      throw new Error(`Pull failed: ${response.status}`);
    }

    const changes = await response.json();

    // 应用变更到本地
    for (const change of changes) {
      await this.applyChange(change);
    }

    // 更新最后同步时间
    this.lastSyncTime = Date.now();
    localStorage.setItem('lastSyncTime', this.lastSyncTime);

    return changes.length;
  }

  // 应用服务器变更
  async applyChange(change) {
    const storeName = change.entityType + 's';
    const entity = change.data;

    switch (change.operation) {
      case 'CREATE':
      case 'UPDATE':
        entity.syncStatus = 'synced';
        await this.localDB.put(storeName, entity);
        break;
      case 'DELETE':
        await this.localDB.delete(storeName, entity.id);
        break;
    }
  }

  // 完整同步
  async fullSync() {
    // 1. 先推送上未完成的操作
    await this.pushPendingChanges();

    // 2. 拉取服务器最新数据
    const changedCount = await this.pullChanges();

    console.log(`Full sync completed, ${changedCount} changes pulled`);

    return changedCount;
  }

  // 推送待处理变更
  async pushPendingChanges() {
    const pending = await this.localDB.getAll('operationQueue', 'status', 'pending');

    for (const op of pending) {
      try {
        await this.syncOperation(op);

        op.status = 'completed';
        op.completedAt = Date.now();
        await this.localDB.put('operationQueue', op);
      } catch (error) {
        op.error = error.message;
        await this.localDB.put('operationQueue', op);
      }
    }
  }
}

冲突解决策略

处理数据同步冲突：

// 冲突检测与解决
class ConflictResolver {
  constructor(localDB) {
    this.localDB = localDB;
  }

  // 检测冲突
  async detectConflict(localEntity, serverEntity) {
    if (!serverEntity) return null;

    const localTime = localEntity.updatedAt;
    const serverTime = serverEntity.updatedAt;

    // 如果本地修改时间晚于服务器，且本地有未同步的修改
    if (localTime > serverTime && localEntity.syncStatus === 'pending') {
      return {
        type: 'CONFLICT',
        localEntity,
        serverEntity,
        localTime,
        serverTime
      };
    }

    return null;
  }

  // 解决策略
  resolveConflict(conflict, strategy = 'local_wins') {
    switch (strategy) {
      case 'local_wins':
        return conflict.localEntity;
      case 'server_wins':
        return conflict.serverEntity;
      case 'manual':
        // 返回冲突信息，等待用户决策
        return { needsResolution: true, conflict };
      case 'merge':
        return this.mergeEntities(conflict.localEntity, conflict.serverEntity);
      default:
        return conflict.localEntity;
    }
  }

  // 合并实体
  mergeEntities(local, server) {
    const merged = { ...server };

    // 本地修改的字段优先
    for (const key in local) {
      if (local[key] !== server[key]) {
        merged[key] = local[key];
      }
    }

    return merged;
  }

  // 自动解决冲突（基于规则）
  async autoResolve(conflict) {
    const entity = conflict.localEntity;

    // 规则1：订单状态以服务器为准（避免重复处理）
    if (entity.entityType === 'order') {
      return conflict.serverEntity;
    }

    // 规则2：库存数据以服务器为准（避免超卖）
    if (entity.entityType === 'inventory') {
      // 记录本地修改用于后续分析
      await this.logConflict(conflict);
      return conflict.serverEntity;
    }

    // 规则3：客户信息使用本地版本（可能已人工修改）
    if (entity.entityType === 'customer') {
      return conflict.localEntity;
    }

    // 默认使用本地版本
    return conflict.localEntity;
  }

  // 记录冲突日志
  async logConflict(conflict) {
    await this.localDB.add('changeLog', {
      entityType: conflict.localEntity.entityType,
      entityId: conflict.localEntity.id,
      localData: conflict.localEntity,
      serverData: conflict.serverEntity,
      timestamp: Date.now(),
      resolved: false,
      resolution: null
    });
  }

  // 获取所有未解决的冲突
  async getUnresolvedConflicts() {
    return this.localDB.getAll('changeLog', 'resolved', false);
  }

  // 手动解决冲突
  async resolveManually(conflictId, resolution) {
    const conflict = await this.localDB.get('changeLog', conflictId);
    conflict.resolved = true;
    conflict.resolution = resolution;
    conflict.resolvedAt = Date.now();

    await this.localDB.put('changeLog', conflict);

    // 应用解决后的数据
    if (resolution === 'use_local') {
      await this.localDB.put(conflict.entityType + 's', conflict.localData);
    } else if (resolution === 'use_server') {
      await this.localDB.put(conflict.entityType + 's', conflict.serverData);
    }

    return conflict;
  }
}

最佳实践建议

• 数据分区：根据业务需求合理划分同步范围
• 冲突日志：记录所有冲突便于后续审计
• 版本控制：实现乐观锁防止更新丢失
• 压缩传输：大数据量采用增量同步减少带宽
• 离线提示：清晰告知用户当前网络状态

总结

离线模式让 PSI 进销存系统更加实用：

• 随时可用：网络不稳定也能正常业务操作
• 数据安全：本地存储保证数据不丢失
• 自动同步：网络恢复后自动同步无需人工干预
• 冲突处理：多种策略解决数据冲突

通过离线模式设计，提升用户体验和工作效率。