当前位置：首页 > news >正文

Redis中Geospatial 实际应用指南

news 2025/10/14 6:47:28

Redis 的 Geospatial（地理空间）功能在实际项目中非常实用，它基于 Sorted Set 数据结构，通过 GeoHash 算法将经纬度编码为分数，实现了高效的地理位置存储和查询。

1. Geospatial 核心命令

# 添加地理位置
GEOADD key longitude latitude member# 获取地理位置
GEOPOS key member# 计算距离
GEODIST key member1 member2 [unit]# 附近搜索
GEORADIUS key longitude latitude radius unit [WITHDIST] [WITHCOORD] [COUNT count]
GEORADIUSBYMEMBER key member radius unit [WITHDIST] [WITHCOORD] [COUNT count]# GeoHash（用于外部地图服务）
GEOHASH key member

2. 实际应用场景

2.1 附近的人/商家搜索

import redis
import jsonclass NearbyService:def __init__(self):self.redis_client = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)def add_user_location(self, user_id, lng, lat):"""添加用户位置"""self.redis_client.geoadd('user:locations',lng, lat, f"user:{user_id}")def find_nearby_users(self, center_lng, center_lat, radius_km=5, limit=10):"""查找附近用户"""return self.redis_client.georadius('user:locations',center_lng, center_lat, radius_km, 'km',withdist=True,  # 返回距离withcoord=True, # 返回坐标count=limit,    # 限制数量sort='ASC'      # 按距离排序)def calculate_distance(self, user1_id, user2_id):"""计算两个用户间的距离"""return self.redis_client.geodist('user:locations',f"user:{user1_id}",f"user:{user2_id}",unit='km')# 使用示例
service = NearbyService()# 添加测试数据
service.add_user_location("1001", 116.3974, 39.9093)  # 北京
service.add_user_location("1002", 116.4074, 39.9193)  # 北京附近
service.add_user_location("1003", 121.4737, 31.2304)  # 上海# 查找天安门附近的用户
nearby_users = service.find_nearby_users(116.3974, 39.9093, 10)
print("附近用户:", nearby_users)

2.2 外卖/快递配送系统

class DeliverySystem:def __init__(self):self.redis_client = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)def add_restaurant(self, restaurant_id, lng, lat, name):"""添加餐厅位置"""self.redis_client.geoadd('restaurants', lng, lat, restaurant_id)self.redis_client.hset(f'restaurant:{restaurant_id}', 'name', name)def add_rider(self, rider_id, lng, lat):"""添加骑手位置"""self.redis_client.geoadd('riders', lng, lat, f"rider:{rider_id}")def find_nearest_riders(self, restaurant_id, radius_km=3, count=5):"""为餐厅寻找最近骑手"""# 获取餐厅位置pos = self.redis_client.geopos('restaurants', restaurant_id)if not pos or not pos[0]:return []lng, lat = pos[0]# 查找附近骑手riders = self.redis_client.georadius('riders', lng, lat, radius_km, 'km',withdist=True,count=count,sort='ASC')return ridersdef track_delivery_range(self, order_id, target_lng, target_lat, max_radius_km):"""监控配送范围"""key = f"delivery:range:{order_id}"# 设置目标位置和半径self.redis_client.geoadd(key, target_lng, target_lat, "target")self.redis_client.expire(key, 3600)  # 1小时过期return key# 使用示例
delivery = DeliverySystem()# 添加餐厅
delivery.add_restaurant("r001", 116.3974, 39.9093, "北京烤鸭店")# 添加骑手
delivery.add_rider("d001", 116.4000, 39.9100)
delivery.add_rider("d002", 116.3900, 39.9000)# 寻找最近骑手
nearest_riders = delivery.find_nearest_riders("r001")
print("最近骑手:", nearest_riders)

2.3 车辆监控与电子围栏

class VehicleTrackingSystem:def __init__(self):self.redis_client = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)def update_vehicle_location(self, vehicle_id, lng, lat):"""更新车辆位置"""self.redis_client.geoadd('vehicles:current', lng, lat, vehicle_id)# 同时保存到历史记录（使用时间戳）timestamp_key = f"vehicles:history:{vehicle_id}"self.redis_client.geoadd(timestamp_key, lng, lat, str(int(time.time())))# 保留最近100个位置点self.redis_client.zremrangebyrank(timestamp_key, 0, -100)def create_geofence(self, fence_id, points):"""创建电子围栏"""# points: [(lng1, lat1), (lng2, lat2), ...]fence_key = f"geofence:{fence_id}"# 添加围栏边界点for i, (lng, lat) in enumerate(points):self.redis_client.geoadd(fence_key, lng, lat, f"point_{i}")return fence_keydef check_vehicles_in_fence(self, fence_id, radius_km=0.1):"""检查围栏内车辆"""fence_key = f"geofence:{fence_id}"# 获取围栏中心点（使用第一个点作为参考）center = self.redis_client.geopos(fence_key, "point_0")if not center or not center[0]:return []lng, lat = center[0]# 查找围栏内的车辆vehicles = self.redis_client.georadius('vehicles:current',lng, lat, radius_km, 'km',withdist=True,withcoord=True)return vehiclesdef get_vehicle_trail(self, vehicle_id, start_time=0, end_time=None):"""获取车辆轨迹"""if end_time is None:end_time = int(time.time())key = f"vehicles:history:{vehicle_id}"positions = self.redis_client.zrangebyscore(key, start_time, end_time, withscores=True)return positions# 使用示例
tracking = VehicleTrackingSystem()# 更新车辆位置
tracking.update_vehicle_location("car001", 116.3974, 39.9093)# 创建禁行区域围栏
fence_points = [(116.3964, 39.9083),(116.3984, 39.9083),(116.3984, 39.9103),(116.3964, 39.9103)
]
tracking.create_geofence("restricted_area", fence_points)# 检查禁行区域内车辆
violations = tracking.check_vehicles_in_fence("restricted_area")
print("禁行区域内车辆:", violations)

3. 性能优化建议

3.1 数据分片策略

def get_shard_key(base_key, lng, lat, shard_size=10):"""根据地理位置分片"""# 将经纬度映射到分片lng_shard = int((lng + 180) / 360 * shard_size)lat_shard = int((lat + 90) / 180 * shard_size)return f"{base_key}:{lng_shard}:{lat_shard}"# 使用分片存储
shard_key = get_shard_key("user:locations", lng, lat)
redis_client.geoadd(shard_key, lng, lat, user_id)

3.2 缓存策略

def get_nearby_users_cached(center_lng, center_lat, radius_km=5):"""带缓存的附近用户查询"""cache_key = f"nearby:{round(center_lng, 2)}:{round(center_lat, 2)}:{radius_km}"# 尝试从缓存获取cached = redis_client.get(cache_key)if cached:return json.loads(cached)# 查询数据库result = find_nearby_users(center_lng, center_lat, radius_km)# 缓存结果（30秒过期）redis_client.setex(cache_key, 30, json.dumps(result))return result