仓储机器人的实时数据架构：WMS与机器人调度系统的低延迟同步

一、当机器人"抢单"时把货架撞翻了

某智能仓库部署了100台AGV搬运机器人。WMS系统下达指令"去A-12货架取货"，机器人调度系统(RCS)分配到机器人R-07。但在R-07移动到A-12的3秒内，另一台R-15因为通信延迟没有及时收到"A-12已被锁定"的广播消息，也驶向A-12——两机在过道交汇处紧急制动，货架上的货物散落一地。

这就是仓储机器人场景的核心挑战：WMS（仓库管理系统）和RCS（机器人控制系统）之间的状态同步延迟必须控制在100ms以内——超过这个阈值，机器人就会基于过时的信息做出冲突决策。

二、WMS-RCS的实时通信架构

三、实时任务调度实现

import redis
import time
import json
from threading import Thread

class WMSRobotBridge:
    def __init__(self, redis_client):
        self.redis = redis_client
        self.pubsub = redis_client.pubsub()
        
        # 订阅机器人状态更新
        self.pubsub.psubscribe("robot:state:*")
        self.listener_thread = Thread(target=self._listen_robot_states)
        self.listener_thread.daemon = True
        self.listener_thread.start()
    
    def dispatch_task(self, task: dict) -> bool:
        """WMS下发搬运任务"""
        task_id = task['task_id']
        shelf_id = task['shelf_id']
        target_location = task['target_location']
        
        try:
            # Step 1: 获取货架锁（SET NX，100ms超时）
            lock_key = f"shelf:lock:{shelf_id}"
            acquired = self.redis.set(
                lock_key, task_id, nx=True, px=100
            )
            
            if not acquired:
                raise ShelfLockedException(f"货架 {shelf_id} 已被其他任务锁定")
            
            # Step 2: 发布任务到RCS
            task_payload = json.dumps({
                'task_id': task_id,
                'action': 'm_move',
                'shelf_id': shelf_id,
                'from_location': self._get_shelf_location(shelf_id),
                'to_location': target_location,
                'priority': task.get('priority', 1),
                'timestamp': int(time.time() * 1000)
            })
            
            # 使用Redis Pub/Sub发布任务
            receivers = self.redis.publish('rcs:task:new', task_payload)
            
            if receivers == 0:
                # 没有RCS消费者在线
                self.redis.delete(lock_key)
                raise RCSError("无RCS节点在线")
            
            # Step 3: 将任务加入待确认队列
            self.redis.zadd(
                'rcs:pending_tasks',
                {task_id: int(time.time())}
            )
            
            # 等待RCS确认（100ms超时）
            confirmed = self._wait_confirmation(task_id, timeout_ms=100)
            
            if not confirmed:
                self.redis.delete(lock_key)
                raise RCSTimeoutException("RCS确认超时")
            
            return True
            
        except (ShelfLockedException, RCSError, RCSTimeoutException) as e:
            raise e
        except Exception as e:
            raise WMSException(f"任务下发失败: {e}")
    
    def _wait_confirmation(self, task_id: str, 
                            timeout_ms: int = 100) -> bool:
        """等待RCS确认（使用BLPOP阻塞式队列）"""
        queue_key = f"rcs:confirm:{task_id}"
        
        try:
            result = self.redis.blpop(queue_key, 
                                       timeout=timeout_ms / 1000.0)
            if result:
                _, data = result
                confirm = json.loads(data)
                return confirm.get('status') == 'ok'
        except redis.TimeoutError:
            pass
        
        return False
    
    def _listen_robot_states(self):
        """监听机器人状态更新（后台线程）"""
        for message in self.pubsub.listen():
            if message['type'] != 'pmessage':
                continue
            
            try:
                channel = message['channel'].decode()
                data = json.loads(message['data'])
                robot_id = channel.split(':')[-1]
                
                # 更新机器人位置到Redis GEO
                self.redis.geoadd(
                    'robot:positions',
                    (float(data['x']), float(data['y']), robot_id)
                )
                
                # 更新状态Hash
                self.redis.hset(
                    f'robot:detail:{robot_id}',
                    mapping={
                        'status': data['status'],
                        'battery': str(data['battery']),
                        'current_task': data.get('task_id', ''),
                        'last_update': str(time.time())
                    }
                )
                
            except Exception as e:
                # 单条消息解析失败不影响后续处理
                print(f"Robot state parse error: {e}")

RCS侧的冲突检测：

class RCSTrafficControl:
    def __init__(self, redis_client):
        self.redis = redis_client
        self.map = self._load_warehouse_map()
    
    def check_path_collision(self, robot_id: str, 
                              path: list) -> bool:
        """检测规划的路径是否与其他机器人的路径冲突"""
        collision_window_ms = 5000  # 5秒内的路径段
        
        now = int(time.time() * 1000)
        
        for segment in path:
            # 对每个路段加时间窗口锁
            lock_key = (
                f"path:lock:{segment['x']}:{segment['y']}:"
                f"{segment['t_start']}:{segment['t_end']}"
            )
            
            # 尝试获取锁
            acquired = self.redis.set(
                lock_key, robot_id, 
                nx=True,  # 仅当key不存在时设置
                px=collision_window_ms
            )
            
            if not acquired:
                # 路段被占用，返回冲突
                occupying_robot = self.redis.get(lock_key)
                return False
        
        return True
    
    def plan_alternative_path(self, robot_id: str,
                               start: tuple, end: tuple,
                               occupied_segments: list) -> list:
        """避开冲突路段重新规划路径"""
        # A*路径规划，避开occupied_segments标记的路段
        # 此处简化实现
        path = self._astar_search(start, end, occupied_segments)
        if path is None:
            raise NoPathException("无法找到无冲突路径")
        return path

四、WMS-RCS实时同步的三个关键SLA

SLA一：任务下发延迟 < 100ms。从WMS API调用到RCS收到任务并确认，整个链路的耗时。超时的常见原因：Redis的BGSAVE阻塞、网络丢包重传、RCS节点的GC停顿。

SLA二：货架锁的粒度与时长。锁太细（锁单个货格）→锁数量爆炸；锁太粗（锁整排货架）→并发度低。经验值是锁单个货架面（约2-3个货格），锁有效期100ms（任务的Round-Trip时间）。

SLA三：机器人心跳间隔。机器人需要每50ms上报一次位置和状态。如果100ms没有心跳，RCS应假定该机器人已失联，立即向相邻机器人广播"避开该区域"的紧急指令。

五、总结

WMS与机器人调度系统的通信延迟决定了仓库的"物流节奏"——100ms以内的延迟，机器人的运行效率是人工作业的3-5倍；超过200ms，机器人开始互相等待和避让，效率下降到不如人工。

Redis在中间承担了"共享内存+实时广播"的角色：任务队列（ZSET按优先级排序）、状态同步（Hash结构）、空间索引（GEO命令）、路径锁（SET NX带TTL）。MySQL只做最终的任务持久化和审计日志。

本文属于「行业场景与项目复盘」系列，深入分析仓储机器人场景下WMS与RCS的低延迟实时数据同步架构。