引言

在零售行业，门店库存实时同步直接影响用户体验和转化率。传统方案存在：

• 数据一致性难题：多门店库存状态同步延迟。
• 并发控制瓶颈：高并发的库存锁定请求。
• 多端体验差异：H5与小程序的技术栈适配。

Taro 作为跨端框架，基于 React 语法支持一套代码适配多端（H5/小程序/RN 等），结合实时通信技术与分布式架构，可高效实现多端库存同步系统。

本文将基于Taro3.x+React技术栈，从架构设计到代码实现，完整呈现一套多端库存同步方案。

一、整体架构设计

1.1 技术全景图

1.2 核心功能模块

1.3 混合同步策略

• 实时更新：库存变动时，服务端通过 WebSocket 主动推送至所有在线客户端。
• 差异轮询：离线设备每 30 秒请求增量数据（使用 lastUpdateTime 时间戳过滤变更）。

1.4 技术矩阵

二、核心功能实现

2.1 实时库存查询

附近门店+库存状态：

/**
 * 获取商品库存信息
 * @param {string} sku - 商品SKU编码
 * @param {object} location - 用户地理位置信息
 * @param {number} location.lat - 纬度
 * @param {number} location.lon - 经度
 * @returns {object} 包含库存状态和加载状态的对象
 * @property {object} stock - 当前库存数据
 * @property {boolean} isLoading - 数据加载状态
 */
export function useStock(sku, location) {
  const [stock, setStock] = useState({});
  const [isLoading, setIsLoading] = useState(true);

  useEffect(() => {
    // 获取用户附近的门店列表
    const fetchNearbyStores = async () => {
      const stores = await Taro.request({
        url: '/api/stores/nearby',
        data: { latitude: location.lat, longitude: location.lon },
      });
      return stores.data;
    };

    // 建立WebSocket连接监听库存变化
    const ws = Taro.connectSocket({
      url: 'wss://api.example.com/stock',
      success: () => {
        ws.onMessage(res => {
          const data = JSON.parse(res.data);
          if (data.sku === sku) setStock(data.stock);
        });
      },
    });

    // 初始化加载库存数据
    fetchNearbyStores().then(stores => {
      Taro.request({
        url: '/api/stock/query',
        data: { sku, storeIds: stores.map(s => s.id) },
      }).then(res => setStock(res.data));
      setIsLoading(false);
    });

    // 组件卸载时关闭WebSocket连接
    return () => ws.close();
  }, [sku, location]);

  return { stock, isLoading };
}

• 设计思路：

• 地理围栏筛选：根据用户位置过滤 5 公里内的门店。
• 双通道更新：WebSocket 推送实时变更 + 初始化 HTTP 请求兜底。

• 参数解析：

• sku：商品唯一标识，用于服务端精准推送。
• location：用户经纬度，用于计算最近门店。

2.2 到店自提库存锁定

库存锁定是到店自提业务的核心环节，需要解决的关键问题包括：

• 并发控制：防止超卖。
• 锁定时效：设置合理的锁定时间（如15分钟）。
• 失败处理：提供友好的用户提示。

实现代码：

/**
 * 库存锁定函数
 * 
 * @param {string} storeId - 门店/仓库唯一标识
 * @param {string} sku - 商品SKU编码
 * @param {number} quantity - 需要锁定的商品数量
 * @returns {Promise<void>} 无直接返回值，但包含异步操作
 */
const lockStock = async (storeId, sku, quantity) => {
  try {
    // 构建分布式锁的key并尝试获取锁
    const lockKey = `lock:${storeId}:${sku}`;
    const lock = await Taro.request({
      url: '/api/lock',
      method: 'POST',
      data: { lockKey, ttl: 900 }, // 锁定 15 分钟
    });

    // 成功获取锁后的处理流程
    if (lock.data.code === 200) {
      // 执行库存扣减操作
      await Taro.request({
        url: '/api/stock/deduct',
        method: 'POST',
        data: { storeId, sku, quantity },
      });
      
      // 持久化订单锁定记录
      await saveOrderLock(storeId, sku, quantity);
    } else {
      // 获取锁失败的用户提示
      Taro.showToast({ title: '锁定失败，请重试', icon: 'error' });
    }
  } catch (err) {
    // 网络异常处理
    Taro.showToast({ title: '网络异常', icon: 'none' });
  } finally {
    // 无论成功与否最终都要释放锁
    releaseLock(lockKey);
  }
};

关键流程：

在分布式系统中锁定指定商品的库存数量，包含以下步骤：

• 获取分布式锁防止并发问题。
• 执行库存扣减操作。
• 记录订单锁定信息。

• 容错设计：

• 锁超时自动释放：避免死锁（如用户未支付时）。
• 库存回滚：MySQL 事务失败时解除 Redis 锁。

2.3 库存紧张提示

• 规则引擎：

/**
 * 根据库存数量获取库存状态信息
 * 
 * @param {number} quantity - 当前库存数量
 * @returns {Object} 包含状态信息的对象，结构为 {status: string, color: string}
 *   - status: 库存状态文字描述
 *   - color: 状态对应的颜色代码
 */
const getStockStatus = quantity => {
  // 库存为0时返回售罄状态
  if (quantity === 0) return { status: '售罄', color: '#ff4d4f' };
  
  // 库存少于10时返回紧张状态
  if (quantity < 10) return { status: '紧张', color: '#faad14' };
  
  // 默认返回充足状态
  return { status: '充足', color: '#52c41a' };
};

• 功能描述：

• 该函数根据传入的库存数量返回对应的状态对象，包含状态文字和显示颜色。
• 状态分为三种：售罄(红色)、紧张(黄色)和充足(绿色)。

• 动态阈值：

• 基于历史销量动态计算阈值（如促销期自动调低阈值）。

2.4 实时库存同步

2.4.1 技术架构设计

2.4.2 核心组件功能

（1）消息队列（Kafka）

功能：

• 承接POS系统的库存变更事件。
• 缓冲高峰流量。
• 保证消息顺序性。

配置示例：

/**
 * Kafka生产者配置
 * 配置说明：
 * - acks: 'all' 要求所有副本都确认消息接收，提供最强的持久性保证
 * - retries: 5 发送失败时的最大重试次数
 * - compression: 1 使用GZIP压缩算法减少网络传输量
 */
const producer = kafka.producer({
  acks: 'all',
  retries: 5,
  compression: 1,
});

/**
 * 发送库存变更事件到Kafka
 * 
 * @param {string} storeId - 门店唯一标识符
 * @param {string} skuId - 商品SKU唯一标识符
 * @param {number} delta - 库存变化量（正数表示入库，负数表示出库）
 * @returns {Promise} 返回Kafka发送操作的Promise
 * 
 * 消息格式说明：
 * - 使用复合键保证同一门店同一商品的消息有序
 * - 消息体包含库存变更的完整上下文和时间戳
 */
async function sendInventoryEvent(storeId, skuId, delta) {
  // 构建并发送Kafka消息
  await producer.send({
    topic: 'inventory-updates',
    messages: [
      {
        key: `${storeId}-${skuId}`,
        value: JSON.stringify({
          storeId,
          skuId,
          delta,
          timestamp: Date.now(),
        }),
      },
    ],
  });
}

（2）库存计算引擎

核心逻辑：

实现代码：

/**
 * 库存计算服务消费者
 * 监听库存更新消息，处理库存变更并触发推送事件
 */
const inventoryProcessor = new KafkaConsumer({
  groupId: 'inventory-processor',
  topics: ['inventory-updates'],
});

/**
 * 处理库存更新消息
 * @param {Object} msg - Kafka消息对象
 * @param {string} msg.value - 消息内容(JSON格式)
 */
inventoryProcessor.on('message', async msg => {
  // 解析消息内容
  const { storeId, skuId, delta } = JSON.parse(msg.value);

  // 获取当前库存：优先从缓存读取，缓存未命中则查询数据库
  const cacheKey = `inventory:${storeId}:${skuId}`;
  let current = await redis.get(cacheKey);

  if (!current) {
    // 数据库查询并设置缓存
    current = await db.query(
      `SELECT stock FROM inventory 
       WHERE store_id = ? AND sku_id = ?`,
      [storeId, skuId],
    );
    await redis.set(cacheKey, current, 'EX', 300); // 缓存5分钟
  }

  // 计算并更新库存
  const newStock = parseInt(current) + delta;
  await redis.set(cacheKey, newStock, 'EX', 300);

  // 推送条件检查：库存状态变化/超时/低库存
  const lastPush = await redis.get(`last_push:${storeId}:${skuId}`);
  const now = Date.now();

  if (!lastPush || now - lastPush > 30000 || checkInventoryStatusChanged(current, newStock)) {
    // 生成推送事件并更新最后推送时间
    await pushService.addEvent({
      storeId,
      skuId,
      newStock,
      timestamp: now,
    });
    await redis.set(`last_push:${storeId}:${skuId}`, now, 'EX', 300);
  }
});

/**
 * 检查库存状态是否发生变化
 * @param {number} oldStock - 变更前库存数量
 * @param {number} newStock - 变更后库存数量
 * @returns {boolean} 状态是否发生变化
 */
function checkInventoryStatusChanged(oldStock, newStock) {
  const oldStatus = getStockStatus(oldStock);
  const newStatus = getStockStatus(newStock);
  return oldStatus !== newStatus;
}

/**
 * 根据库存数量获取状态分类
 * @param {number} stock - 库存数量
 * @returns {string} 库存状态(soldout/danger/warning/normal)
 */
function getStockStatus(stock) {
  if (stock <= 0) return 'soldout';
  if (stock <= 5) return 'danger';
  if (stock <= 10) return 'warning';
  return 'normal';
}

（3）实时推送服务

（4）多端推送实现

// 库存上下文
import Taro, { useEffect, useState, createContext } from '@tarojs/taro';

/**
 * 库存上下文对象，用于跨组件共享库存数据
 */
export const InventoryContext = createContext();

/**
 * 库存数据提供者组件，负责管理库存数据状态和订阅更新
 * @param {Object} props - 组件属性
 * @param {ReactNode} props.children - 子组件
 * @returns {JSX.Element} 提供库存上下文的React组件
 */
export const InventoryProvider = ({ children }) => {
  const [inventoryMap, setInventoryMap] = useState({});
  const [lastUpdate, setLastUpdate] = useState(0);

  /**
   * 初始化订阅和轮询逻辑
   * 1. 请求订阅权限
   * 2. 注册订阅关系
   * 3. 启动轮询机制
   */
  useEffect(() => {
    const init = async () => {
      await Taro.requestSubscribeMessage({
        tmplIds: ['INVENTORY_UPDATE_TEMPLATE'],
      });

      await registerSubscriptions();
      startPolling();
    };

    init();

    return () => {
      clearInterval(pollingTimer);
    };
  }, []);

  /**
   * 注册库存更新订阅关系
   * 获取用户关注的店铺列表并注册到服务器
   */
  const registerSubscriptions = async () => {
    const followedStores = (await Taro.getStorageSync('followedStores')) || [];

    await Taro.request({
      url: '/api/inventory/subscribe',
      method: 'POST',
      data: {
        stores: followedStores,
      },
    });
  };

  /**
   * 启动库存数据轮询机制
   * 设置60秒间隔的定时器，并立即执行首次数据拉取
   */
  const startPolling = () => {
    pollingTimer = setInterval(() => {
      fetchUpdates();
    }, 60000);

    fetchUpdates();
  };

  /**
   * 从服务器拉取库存更新数据
   * 根据最后更新时间获取增量更新，并合并到当前库存状态
   */
  const fetchUpdates = async () => {
    const since = lastUpdate;
    const res = await Taro.request({
      url: '/api/inventory/updates',
      data: { since },
    });

    if (res.data.success) {
      const updates = res.data.updates;
      if (updates.length > 0) {
        const newMap = { ...inventoryMap };
        updates.forEach(update => {
          const key = `${update.storeId}-${update.skuId}`;
          newMap[key] = update.stock;
        });
        setInventoryMap(newMap);
        setLastUpdate(Date.now());
      }
    }
  };

  /**
   * 设置微信消息监听
   * 处理来自微信服务端的实时库存更新推送
   */
  Taro.useDidShow(() => {
    const page = Taro.getCurrentInstance().page;
    page.onMessage = msg => {
      if (msg.type === 'inventory_update') {
        handleInventoryUpdate(msg.payload);
      }
    };
  });

  /**
   * 处理库存更新数据
   * @param {Object} payload - 更新数据
   * @param {string} payload.storeId - 店铺ID
   * @param {string} payload.skuId - 商品SKU ID
   * @param {number} payload.stock - 最新库存数量
   */
  const handleInventoryUpdate = payload => {
    const { storeId, skuId, stock } = payload;
    const key = `${storeId}-${skuId}`;
    setInventoryMap(prev => ({
      ...prev,
      [key]: stock,
    }));
  };

  return <InventoryContext.Provider value={{ inventoryMap }}>{children}</InventoryContext.Provider>;
};

（5）离线消息处理

（6）消息补偿机制

/**
 * 保存离线消息到数据库
 * 
 * @param {string} userId - 接收消息的用户ID
 * @param {Object} message - 要保存的消息内容对象
 * @returns {Promise<void>} - 无返回值
 */
async function saveOfflineMessage(userId, message) {
  await db.query(
    `INSERT INTO offline_messages (user_id, content, created_at)
     VALUES (?, ?, NOW())`,
    [userId, JSON.stringify(message)]
  )
}

/**
 * 离线消息重发服务
 */
const resendService = {
  /**
   * 启动消息重发服务
   * 每分钟执行一次，处理过去24小时内状态为pending的离线消息
   * 
   * @returns {Promise<void>} - 无返回值
   */
  start: async function() {
    setInterval(async () => {
      // 查询待处理的离线消息（24小时内，最多100条）
      const messages = await db.query(
        `SELECT * FROM offline_messages 
         WHERE status = 'pending' 
         AND created_at > DATE_SUB(NOW(), INTERVAL 1 DAY)
         LIMIT 100`
      )
      
      // 逐条处理消息
      for (const msg of messages) {
        try {
          // 尝试推送消息给用户
          await pushToUser(msg.user_id, msg.content)
          
          // 推送成功则更新状态为已发送
          await db.query(
            `UPDATE offline_messages SET status = 'sent'
             WHERE id = ?`, [msg.id]
          )
        } catch (error) {
          // 推送失败则增加重试计数
          await db.query(
            `UPDATE offline_messages SET retry_count = retry_count + 1
             WHERE id = ?`, [msg.id]
          )
        }
      }
    }, 60000) // 每分钟检查一次
  }
}

主要功能：

• 定时检查待发送的离线消息。
• 尝试重新推送失败的消息。
• 更新消息发送状态和重试次数。

三、常见问题与解决方案

3.1 网络不稳定导致同步失败

• 问题：门店网络抖动时 WebSocket 断开。
• 解决：

• 启用 MQTT 协议（支持 QoS 消息质量分级）5。
• 本地暂存变更（IndexedDB + Taro 本地存储），网络恢复后重传。

3.2 多端数据冲突

• 场景：H5 和小程序同时修改同一商品库存。
• 解决：

• 乐观锁机制：提交时携带版本号，冲突时自动重试：

updateStock({ sku, quantity, version }) {
  const currentVersion = getVersionFromDB(sku);
  if (currentVersion !== version) throw 'Version conflict';
  // 否则更新成功
}

3.3 小程序渲染性能瓶颈

• 问题：门店列表页数据量大时卡顿。
• 解决：

• 虚拟列表：使用 Taro.virtualList 组件，仅渲染可视区内容。
• 分页加载：滚动触底加载新批次数据。
• 缓存策略

结语

本文基于 Taro 实现了多端门店库存实时同步系统，核心成果包括：

• 混合同步架构：通过 WebSocket + 差异轮询，平衡实时性与性能；
• 分布式锁设计：保障“到店自提”场景的库存一致性；
• 动态阈值提示：提升库存透明度，降低用户决策成本；
• Taro 深度适配：一套代码覆盖 H5 与小程序，降低开发成本。

多端统一开发并非简单的API适配，更需要从架构层面思考状态同步、事务一致性等深层问题。希望本文方案能为大家带来启发。