智能调度系统的数据结构：运力、订单与仓库的三维匹配算法支撑-尧图网络科技

智能调度系统的数据结构：运力、订单与仓库的三维匹配算法支撑

一、当调度员比算法更"智能"：多目标优化的数据困境

某即时配送平台的调度算法上线后，配送员的投诉量上升了30%。算法追求的是"全局最优"——总配送距离最短、总超时率最低。但配送员看到的是：被分配到离家15公里外的订单片区、午高峰连续接了5个反方向的订单。

调度问题的本质不是"最短路径"，而是**运力（谁有空送）× 订单（什么货送到哪）× 仓库（从哪个仓出货）**的三维匹配。每增加一个约束条件（时效、成本、配送员偏好），计算复杂度指数级增加。而数据层需要支撑的查询是："在上海市浦东新区的所有空闲配送员中，谁距离仓库B最近、且车辆类型能装下订单#12345的货物？"

二、多约束调度的数据模型

三、Redis GEO + 空间索引的运力匹配

class DispatchEngine: def __init__(self, redis_client, mysql_pool): self.redis = redis_client self.mysql = mysql_pool def find_best_courier(self, order_id: str, warehouse_id: str) -> dict: """为订单找到最优配送员""" # Step 1: 获取仓库和订单信息 warehouse = self._get_warehouse(warehouse_id) order = self._get_order(order_id) if not warehouse or not order: raise DispatchException("仓库或订单信息不存在") # Step 2: 查询仓库周边5km内的空闲配送员 try: nearby_couriers = self.redis.georadius( "courier:locations", warehouse['longitude'], warehouse['latitude'], 5, # 5km半径 unit='km', withcoord=True, withdist=True, sort='ASC' ) except RedisError as e: raise DispatchException("运力位置查询失败", e) if not nearby_couriers: return {'status': 'no_courier', 'suggestion': '扩大搜索半径'} # Step 3: 过滤和评分 candidates = [] for courier_data in nearby_couriers: courier_id = courier_data[0].decode() distance_km = courier_data[1] # 过滤：必须空闲且有对应车型 status = self.redis.hgetall(f"courier:status:{courier_id}") if not status or status.get(b'status') != b'FREE': continue if order.get('vehicle_type') and \ status.get(b'vehicle_type') != order['vehicle_type'].encode(): continue score = self._calculate_score( courier_id, distance_km, order, warehouse ) candidates.append({ 'courier_id': courier_id, 'distance_km': round(distance_km, 2), 'score': round(score, 4) }) # Step 4: 返回最优 candidates.sort(key=lambda x: x['score'], reverse=True) if not candidates: return {'status': 'no_match', 'message': '无符合条件的配送员'} return { 'status': 'success', 'best': candidates[0], 'alternatives': candidates[1:4], 'total_candidates': len(candidates) } def _calculate_score(self, courier_id, distance_km, order, warehouse): """多维度评分""" score = 0.0 # 距离分（距离越近越好） if distance_km <= 1: score += 0.4 elif distance_km <= 3: score += 0.3 elif distance_km <= 5: score += 0.1 # 装载率分（车上已有同方向订单可以拼单） existing_orders = self._get_courier_orders(courier_id) route_compatibility = self._calculate_route_overlap( existing_orders, order ) score += route_compatibility * 0.3 # 配送员偏好区匹配 courier_preferred_area = self._get_courier_preference(courier_id) if self._is_in_area(order['delivery_lat'], order['delivery_lng'], courier_preferred_area): score += 0.2 # 历史配送经验（配送员是否熟悉该区域） experience_score = self._get_experience_score( courier_id, order['delivery_area'] ) score += experience_score * 0.1 return score def assign_order(self, order_id: str, courier_id: str): """执行订单分配（Redis CAS操作）""" lock_key = f"courier:lock:{courier_id}" # 使用SETNX实现乐观锁 try: acquired = self.redis.set( lock_key, order_id, nx=True, ex=30 ) if not acquired: raise DispatchException("配送员已被其他订单锁定") # 更新配送员状态 self.redis.hset( f"courier:status:{courier_id}", mapping={ 'status': 'BUSY', 'current_order': order_id, 'assigned_at': datetime.now().isoformat() } ) # 从空闲集合中移除 self.redis.geoadd( "courier:locations", (0, 0, courier_id) # 用0坐标移除（hack） ) # 写入MySQL（异步） self._async_save_assignment(order_id, courier_id) return {'status': 'assigned', 'courier_id': courier_id} except RedisError as e: raise DispatchException(f"分配失败: {e}") finally: # 如果分配失败，释放锁 if 'acquired' in locals() and acquired: self.redis.delete(lock_key)