返利机器人的商品数据同步方案：API拉取与增量更新的技术实现-平芜编程栈

返利机器人的商品数据同步方案：API拉取与增量更新的技术实现

大家好，我是微赚淘客系统3.0 的研发者省赚客！

在返利机器人场景中，商品数据的实时性与准确性直接影响用户转化率。为保障商品库始终与电商平台（如淘宝联盟、京东联盟）保持同步，微赚淘客系统3.0 采用“全量快照 + 增量拉取 + 本地缓存”三层架构，确保高并发下低延迟响应。

一、商品数据模型设计

本地商品表product_item包含核心字段：

item_id（平台商品ID，主键）
title,price,coupon_amount,commission_rate
update_time（来自平台的最后更新时间戳）
sync_version（本地同步版本号）

该模型支持通过update_time判断是否需增量更新。

二、全量初始化同步

首次接入或数据异常时，执行全量拉取。以淘宝联盟为例，使用其taobao.tbk.item.get接口分页获取：

packagejuwatech.cn.sync.service;importjuwatech.cn.sync.mapper.ProductMapper;importjuwatech.cn.sync.client.TaoBaoApiClient;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.stereotype.Service;importjava.util.List;@ServicepublicclassFullSyncService{@AutowiredprivateTaoBaoApiClienttaoBaoClient;@AutowiredprivateProductMapperproductMapper;publicvoidfullSync(){intpage=1;finalintpageSize=100;booleanhasMore=true;while(hasMore){varresponse=taoBaoClient.getItems(page,pageSize);List<ProductItem>items=response.getData();if(items.isEmpty()){hasMore=false;}else{// 批量插入或覆盖（ON DUPLICATE KEY UPDATE）productMapper.batchUpsert(items);page++;// 避免触发限流Thread.sleep(200);}}}}

其中batchUpsert使用 MySQL 的INSERT ... ON DUPLICATE KEY UPDATE语句：

<insertid="batchUpsert"parameterType="java.util.List">INSERT INTO product_item (item_id, title, price, coupon_amount, commission_rate, update_time, sync_version) VALUES<foreachcollection="list"item="item"separator=",">(#{item.itemId}, #{item.title}, #{item.price}, #{item.couponAmount}, #{item.commissionRate}, #{item.updateTime}, #{item.syncVersion})</foreach>ON DUPLICATE KEY UPDATE title = VALUES(title), price = VALUES(price), coupon_amount = VALUES(coupon_amount), commission_rate = VALUES(commission_rate), update_time = VALUES(update_time), sync_version = sync_version + 1</insert>

三、增量更新机制

每日定时任务拉取过去24小时内变更的商品：

packagejuwatech.cn.sync.task;importjuwatech.cn.sync.service.IncrementalSyncService;importorg.springframework.scheduling.annotation.Scheduled;importorg.springframework.stereotype.Component;importjava.time.LocalDateTime;importjava.time.ZoneOffset;@ComponentpublicclassIncrementalSyncTask{@AutowiredprivateIncrementalSyncServiceincrementalSyncService;@Scheduled(cron="0 0 3 * * ?")// 每天凌晨3点publicvoidrunIncrementalSync(){longstartTs=LocalDateTime.now().minusHours(24).toInstant(ZoneOffset.of("+8")).toEpochMilli();longendTs=System.currentTimeMillis();incrementalSyncService.syncByTimeRange(startTs,endTs);}}

增量服务实现：

packagejuwatech.cn.sync.service;importjuwatech.cn.sync.mapper.ProductMapper;importjuwatech.cn.sync.client.TaoBaoApiClient;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.stereotype.Service;importjava.util.List;@ServicepublicclassIncrementalSyncService{@AutowiredprivateTaoBaoApiClienttaoBaoClient;@AutowiredprivateProductMapperproductMapper;publicvoidsyncByTimeRange(longstartTime,longendTime){intpage=1;finalintpageSize=100;booleanhasMore=true;while(hasMore){varresponse=taoBaoClient.getItemsUpdatedBetween(startTime,endTime,page,pageSize);List<ProductItem>items=response.getData();if(items.isEmpty()){hasMore=false;}else{// 仅当平台 update_time > 本地记录时才更新for(ProductItemitem:items){varlocal=productMapper.selectById(item.getItemId());if(local==null||item.getUpdateTime().isAfter(local.getUpdateTime())){productMapper.upsert(item);}}page++;try{Thread.sleep(100);}catch(InterruptedExceptione){/* ignore */}}}}}

四、本地缓存加速查询

为提升机器人响应速度，商品数据加载至 Redis 缓存，设置 TTL 为 6 小时：

packagejuwatech.cn.product.cache;importjuwatech.cn.sync.mapper.ProductMapper;importorg.springframework.data.redis.core.StringRedisTemplate;importorg.springframework.stereotype.Component;importcom.fasterxml.jackson.databind.ObjectMapper;importjavax.annotation.Resource;importjava.util.concurrent.TimeUnit;@ComponentpublicclassProductCache{@ResourceprivateStringRedisTemplateredisTemplate;@ResourceprivateProductMapperproductMapper;privatefinalObjectMapperobjectMapper=newObjectMapper();publicProductItemgetProduct(StringitemId){Stringkey="product:"+itemId;Stringjson=redisTemplate.opsForValue().get(key);if(json!=null){try{returnobjectMapper.readValue(json,ProductItem.class);}catch(Exceptionignored){}}// 回源数据库ProductItemitem=productMapper.selectById(itemId);if(item!=null){try{redisTemplate.opsForValue().set(key,objectMapper.writeValueAsString(item),6,TimeUnit.HOURS);}catch(Exceptionignored){}}returnitem;}}

五、失败重试与监控告警

所有同步任务均集成 Spring Retry 与日志追踪：

@Retryable(value={Exception.class},maxAttempts=3,backoff=@Backoff(delay=2000))publicvoidsafeSync(StringitemId){// 调用远程API}@Recoverpublicvoidrecover(Exceptionex,StringitemId){alertService.notify("商品同步失败","itemId="+itemId+", error="+ex.getMessage());}