package com.xiaopeng.operation.utils;
|
|
import com.xiaopeng.api.feignClient.vehicleConvoy.XpVehicleClient;
|
import com.xiaopeng.api.model.operation.XpElectronicFenceArea;
|
import com.xiaopeng.api.model.vehicleConvoy.XpVehicle;
|
import com.xiaopeng.common.core.domain.R;
|
import com.xiaopeng.operation.utils.map.GeodesyUtil;
|
import org.springframework.data.redis.core.*;
|
import org.springframework.data.redis.core.script.DefaultRedisScript;
|
import org.springframework.stereotype.Component;
|
|
import java.util.*;
|
import java.util.concurrent.TimeUnit;
|
|
/**
|
* 基于RedisTemplate的排队系统工具类
|
* 实现流程图中的排队校验逻辑
|
*/
|
public class QueueManager {
|
|
|
|
|
private final RedisTemplate<String, Object> redisTemplate;
|
private final XpVehicleClient xpVehicleClient;
|
private final GeodesyUtil geodesyUtil;
|
// Redis Key前缀定义
|
private static final String QUEUE_PREFIX = "queue:";
|
private static final String ORDER_PREFIX = "order:";
|
private static final String WAITING_USERS_PREFIX = "waiting_users:";
|
private static final String FREE_VEHICLES_PREFIX = "free_vehicles:";
|
private static final String USER_QUEUE_POSITION_PREFIX = "user_queue:";
|
private static final String TIMER_PREFIX = "timer:";
|
|
// 配置参数
|
private static final int DEFAULT_TIMEOUT_MINUTES = 30;
|
private static final int DEFAULT_CANCEL_TIMEOUT_SECONDS = 1800; // 30分钟
|
|
// 获取不同类型的操作对象
|
private ValueOperations<String, Object> valueOps;
|
private SetOperations<String, Object> setOps;
|
private ZSetOperations<String, Object> zSetOps;
|
private HashOperations<String, String, Object> hashOps;
|
|
public QueueManager(RedisTemplate<String, Object> redisTemplate, XpVehicleClient xpVehicleClient, GeodesyUtil geodesyUtil) {
|
this.redisTemplate = redisTemplate;
|
this.valueOps = redisTemplate.opsForValue();
|
this.setOps = redisTemplate.opsForSet();
|
this.zSetOps = redisTemplate.opsForZSet();
|
this.hashOps = redisTemplate.opsForHash();
|
this.xpVehicleClient = xpVehicleClient;
|
this.geodesyUtil = geodesyUtil;
|
}
|
|
/**
|
* 初始化排队校验流程
|
* @param areaId 区域ID
|
* @param userId 用户ID
|
* @param orderId 订单ID
|
* @return 排队结果对象
|
*/
|
public QueueResult startQueueProcess(double lon,double lat,double distance,String areaId, String userId, String orderId,Long startPointId) {
|
// 步骤1: 计算范围内空闲车辆数
|
int freeVehiclesCount = countFreeVehicles(areaId,lon,lat,distance);
|
|
if(freeVehiclesCount==0){
|
// 进入排队
|
generateQueueNumber(startPointId.toString(), userId, orderId);
|
|
|
}
|
|
// 步骤2: 计算范围内呼叫人数
|
int waitingUsersCount = countWaitingUsers(areaId);
|
|
// 步骤3: 排队校验判断
|
boolean needQueue = waitingUsersCount > freeVehiclesCount;
|
|
QueueResult result = new QueueResult();
|
result.setNeedQueue(needQueue);
|
result.setQueuePosition(0);
|
result.setOrderId(orderId);
|
result.setUserId(userId);
|
|
if (needQueue) {
|
// 需要排队 - 不设置超时自动取消
|
// 步骤4: 生成排队序号
|
int queueNumber = generateQueueNumber(areaId, userId, orderId);
|
result.setQueuePosition(queueNumber);
|
|
// 步骤5: 队列管理，排序更新
|
updateQueuePosition(areaId, userId, queueNumber);
|
|
// 保存订单信息（不设置超时）
|
saveOrderInfo(orderId, userId, areaId, queueNumber, -1);
|
} else {
|
// 不需要排队 - 普通倒计时30分钟
|
// 保存订单信息（设置30分钟超时）
|
saveOrderInfo(orderId, userId, areaId, 0, DEFAULT_CANCEL_TIMEOUT_SECONDS);
|
|
// 启动30分钟倒计时
|
startCountdownTimer(orderId, DEFAULT_TIMEOUT_MINUTES);
|
}
|
|
return result;
|
}
|
|
/**
|
* 计算范围内空闲车辆数
|
*/
|
private int countFreeVehicles(String areaId,double lon,double lat,double distance) {
|
int count=0;
|
// 获取电子围栏的车辆
|
R<List<XpVehicle>> vehicles = xpVehicleClient.listByFenceId(Long.valueOf(areaId));
|
List<XpVehicle> data = vehicles.getData();
|
for(XpVehicle xpVehicle : data){
|
Object value = redisTemplate.opsForValue().get("DRIVER" + String.valueOf(xpVehicle.getId()));
|
if(null != value){
|
Map<String, Double> distance1 = geodesyUtil.getDistance(lon + "," + lat, value.toString());
|
double d = Double.valueOf(distance1.get("WGS84")).doubleValue();
|
if(d < (distance * 1000)){
|
count++;
|
}
|
}
|
}
|
return count;
|
}
|
|
/**
|
* 计算范围内呼叫人数
|
*/
|
private int countWaitingUsers(String areaId) {
|
String key = WAITING_USERS_PREFIX + areaId;
|
Long size = zSetOps.size(key);
|
return size != null ? size.intValue() : 0;
|
}
|
|
/**
|
* 生成排队序号
|
*/
|
private int generateQueueNumber(String areaId, String userId, String orderId) {
|
String queueKey = QUEUE_PREFIX + areaId;
|
|
// 判断是否白名单 // 重新排序
|
|
// 使用Redis原子操作获取下一个排队序号
|
// 计算分数
|
Long queueNumber = valueOps.increment(queueKey);
|
|
// 将用户加入等待队列，分数为排队序号
|
String waitingKey = WAITING_USERS_PREFIX + areaId;
|
zSetOps.add(waitingKey, userId, queueNumber.doubleValue());
|
|
// 存储用户排队位置映射
|
String userQueueKey = USER_QUEUE_POSITION_PREFIX + userId;
|
Map<String, Object> userQueueInfo = new HashMap<>();
|
userQueueInfo.put("areaId", areaId);
|
userQueueInfo.put("orderId", orderId);
|
userQueueInfo.put("queueNumber", String.valueOf(queueNumber));
|
userQueueInfo.put("joinTime", String.valueOf(System.currentTimeMillis()));
|
hashOps.putAll(userQueueKey, userQueueInfo);
|
|
return queueNumber.intValue();
|
}
|
|
/**
|
* 更新队列排序
|
*/
|
private void updateQueuePosition(String areaId, String userId, int queueNumber) {
|
// 这里可以实现更复杂的队列排序逻辑
|
// 例如：根据优先级、VIP等级等重新排序
|
|
// 当前实现：按加入顺序排序，queueNumber即为位置
|
// 可以扩展为使用zset的分数进行动态排序
|
}
|
|
/**
|
* 保存订单信息
|
*/
|
private void saveOrderInfo(String orderId, String userId,
|
String areaId, int queueNumber, long timeoutSeconds) {
|
String orderKey = ORDER_PREFIX + orderId;
|
|
Map<String, Object> orderInfo = new HashMap<>();
|
orderInfo.put("userId", userId);
|
orderInfo.put("areaId", areaId);
|
orderInfo.put("queueNumber", String.valueOf(queueNumber));
|
orderInfo.put("status", queueNumber > 0 ? "waiting" : "processing");
|
orderInfo.put("createTime", String.valueOf(System.currentTimeMillis()));
|
|
hashOps.putAll(orderKey, orderInfo);
|
|
// 设置过期时间（如果需要）
|
if (timeoutSeconds > 0) {
|
redisTemplate.expire(orderKey, timeoutSeconds, TimeUnit.SECONDS);
|
}
|
}
|
|
/**
|
* 启动倒计时定时器
|
*/
|
private void startCountdownTimer(String orderId, int minutes) {
|
// 设置一个过期key作为倒计时标记
|
String timerKey = TIMER_PREFIX + orderId;
|
valueOps.set(timerKey, "countdown", minutes, TimeUnit.MINUTES);
|
}
|
|
/**
|
* 用户取消订单
|
*/
|
public boolean cancelOrder(String orderId, String userId) {
|
String orderKey = ORDER_PREFIX + orderId;
|
|
// 检查订单是否存在且属于该用户
|
Object storedUserId = hashOps.get(orderKey, "userId");
|
if (storedUserId == null || !storedUserId.equals(userId)) {
|
return false;
|
}
|
|
// 获取订单信息
|
Object areaIdObj = hashOps.get(orderKey, "areaId");
|
Object queueNumberObj = hashOps.get(orderKey, "queueNumber");
|
|
// 更新订单状态为取消
|
hashOps.put(orderKey, "status", "cancelled");
|
hashOps.put(orderKey, "cancelTime", String.valueOf(System.currentTimeMillis()));
|
|
// 如果用户在排队中，从等待队列移除
|
if (areaIdObj != null && queueNumberObj != null) {
|
String areaId = (String) areaIdObj;
|
int queueNumber = Integer.parseInt((String) queueNumberObj);
|
|
if (queueNumber > 0) {
|
String waitingKey = WAITING_USERS_PREFIX + areaId;
|
zSetOps.remove(waitingKey, userId);
|
|
// 清理用户排队信息
|
String userQueueKey = USER_QUEUE_POSITION_PREFIX + userId;
|
redisTemplate.delete(userQueueKey);
|
}
|
}
|
|
// 清理定时器
|
redisTemplate.delete(TIMER_PREFIX + orderId);
|
|
return true;
|
}
|
|
/**
|
* 获取用户排队位置
|
*/
|
public int getQueuePosition(String areaId, String userId) {
|
String waitingKey = WAITING_USERS_PREFIX + areaId;
|
Long rank = zSetOps.rank(waitingKey, userId);
|
|
if (rank != null) {
|
// zrank返回的是从0开始的索引，排队位置需要+1
|
return rank.intValue() + 1;
|
}
|
return -1; // 不在队列中
|
}
|
|
/**
|
* 车辆空闲时，分配订单给排队的用户
|
*/
|
public String allocateOrderToWaitingUser(String areaId, String vehicleId) {
|
String waitingKey = WAITING_USERS_PREFIX + areaId;
|
|
// 获取队列中的第一个用户（分数最小的）
|
Set<Object> firstUsers = zSetOps.range(waitingKey, 0, 0);
|
if (firstUsers == null || firstUsers.isEmpty()) {
|
return null;
|
}
|
|
String userId = (String) firstUsers.iterator().next();
|
|
// 使用Lua脚本保证原子性操作
|
String luaScript =
|
"local waitingKey = KEYS[1]\n" +
|
"local userId = ARGV[1]\n" +
|
"local vehicleId = ARGV[2]\n" +
|
"local userQueueKey = 'user_queue:' .. userId\n" +
|
"local orderId = redis.call('hget', userQueueKey, 'orderId')\n" +
|
"\n" +
|
"if orderId then\n" +
|
" redis.call('zrem', waitingKey, userId)\n" +
|
" \n" +
|
" local orderKey = 'order:' .. orderId\n" +
|
" redis.call('hset', orderKey, 'status', 'allocated')\n" +
|
" redis.call('hset', orderKey, 'vehicleId', vehicleId)\n" +
|
" redis.call('hset', orderKey, 'allocateTime', ARGV[3])\n" +
|
" \n" +
|
" redis.call('del', userQueueKey)\n" +
|
"end\n" +
|
"\n" +
|
"return orderId or ''";
|
|
DefaultRedisScript<String> script = new DefaultRedisScript<>();
|
script.setScriptText(luaScript);
|
script.setResultType(String.class);
|
|
List<String> keys = Collections.singletonList(waitingKey);
|
Object[] args = {userId, vehicleId, String.valueOf(System.currentTimeMillis())};
|
|
String orderId = redisTemplate.execute(script, keys, args);
|
|
return orderId != null && !orderId.isEmpty() ? orderId : null;
|
}
|
|
/**
|
* 添加空闲车辆
|
*/
|
public void addFreeVehicle(String areaId, String vehicleId) {
|
String key = FREE_VEHICLES_PREFIX + areaId;
|
setOps.add(key, vehicleId);
|
}
|
|
/**
|
* 移除空闲车辆
|
*/
|
public void removeFreeVehicle(String areaId, String vehicleId) {
|
String key = FREE_VEHICLES_PREFIX + areaId;
|
setOps.remove(key, vehicleId);
|
}
|
|
/**
|
* 获取等待队列详情
|
*/
|
public List<Map<String, Object>> getWaitingQueueDetails(String areaId) {
|
String waitingKey = WAITING_USERS_PREFIX + areaId;
|
|
// 获取所有等待用户及其分数（排队序号）
|
Set<ZSetOperations.TypedTuple<Object>> waitingUsers =
|
zSetOps.rangeWithScores(waitingKey, 0, -1);
|
|
if (waitingUsers == null || waitingUsers.isEmpty()) {
|
return Collections.emptyList();
|
}
|
|
List<Map<String, Object>> result = new ArrayList<>();
|
int position = 1;
|
|
for (ZSetOperations.TypedTuple<Object> tuple : waitingUsers) {
|
String userId = (String) tuple.getValue();
|
Double score = tuple.getScore();
|
|
Map<String, Object> userInfo = new HashMap<>();
|
userInfo.put("userId", userId);
|
userInfo.put("queuePosition", position);
|
userInfo.put("queueNumber", score != null ? score.intValue() : 0);
|
|
// 获取用户详细信息
|
String userQueueKey = USER_QUEUE_POSITION_PREFIX + userId;
|
Map<String, Object> queueInfo = hashOps.entries(userQueueKey);
|
if (queueInfo != null) {
|
userInfo.putAll(queueInfo);
|
}
|
|
result.add(userInfo);
|
position++;
|
}
|
|
return result;
|
}
|
|
/**
|
* 获取订单状态
|
*/
|
public Map<String, Object> getOrderStatus(String orderId) {
|
String orderKey = ORDER_PREFIX + orderId;
|
Map<String, Object> orderInfo = hashOps.entries(orderKey);
|
|
if (orderInfo != null && !orderInfo.isEmpty()) {
|
// 如果订单在排队中，计算当前排队位置
|
Object queueNumberObj = orderInfo.get("queueNumber");
|
Object areaIdObj = orderInfo.get("areaId");
|
Object userIdObj = orderInfo.get("userId");
|
|
if (queueNumberObj != null && areaIdObj != null && userIdObj != null) {
|
int queueNumber = Integer.parseInt((String) queueNumberObj);
|
if (queueNumber > 0) {
|
String areaId = (String) areaIdObj;
|
String userId = (String) userIdObj;
|
int currentPosition = getQueuePosition(areaId, userId);
|
orderInfo.put("currentQueuePosition", currentPosition);
|
}
|
}
|
}
|
|
return orderInfo;
|
}
|
|
/**
|
* 清理过期数据（后台定时任务调用）
|
*/
|
public void cleanupExpiredData() {
|
// 清理超时未处理的订单（30分钟倒计时）
|
// 这里可以使用Redis的keys扫描或设置定时任务
|
}
|
|
/**
|
* 批量获取排队信息
|
*/
|
public List<QueueResult> batchGetQueueInfo(List<String> orderIds) {
|
List<QueueResult> results = new ArrayList<>();
|
|
redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
|
for (String orderId : orderIds) {
|
String orderKey = ORDER_PREFIX + orderId;
|
|
Map<byte[], byte[]> orderBytes = connection.hGetAll(orderKey.getBytes());
|
if (orderBytes != null && !orderBytes.isEmpty()) {
|
Map<String, String> orderInfo = new HashMap<>();
|
for (Map.Entry<byte[], byte[]> entry : orderBytes.entrySet()) {
|
orderInfo.put(
|
new String(entry.getKey()),
|
new String(entry.getValue())
|
);
|
}
|
|
QueueResult result = new QueueResult();
|
result.setOrderId(orderId);
|
result.setUserId(orderInfo.get("userId"));
|
result.setQueuePosition(Integer.parseInt(orderInfo.getOrDefault("queueNumber", "0")));
|
result.setNeedQueue(result.getQueuePosition() > 0);
|
|
results.add(result);
|
}
|
}
|
return null;
|
});
|
|
return results;
|
}
|
|
/**
|
* 排队结果对象
|
*/
|
public static class QueueResult {
|
private boolean needQueue;
|
private int queuePosition;
|
private String orderId;
|
private String userId;
|
private long createTime;
|
private String status;
|
|
// getters and setters
|
public boolean isNeedQueue() {
|
return needQueue;
|
}
|
|
public void setNeedQueue(boolean needQueue) {
|
this.needQueue = needQueue;
|
}
|
|
public int getQueuePosition() {
|
return queuePosition;
|
}
|
|
public void setQueuePosition(int queuePosition) {
|
this.queuePosition = queuePosition;
|
}
|
|
public String getOrderId() {
|
return orderId;
|
}
|
|
public void setOrderId(String orderId) {
|
this.orderId = orderId;
|
}
|
|
public String getUserId() {
|
return userId;
|
}
|
|
public void setUserId(String userId) {
|
this.userId = userId;
|
}
|
|
public long getCreateTime() {
|
return createTime;
|
}
|
|
public void setCreateTime(long createTime) {
|
this.createTime = createTime;
|
}
|
|
public String getStatus() {
|
return status;
|
}
|
|
public void setStatus(String status) {
|
this.status = status;
|
}
|
|
@Override
|
public String toString() {
|
return "QueueResult{" +
|
"needQueue=" + needQueue +
|
", queuePosition=" + queuePosition +
|
", orderId='" + orderId + '\'' +
|
", userId='" + userId + '\'' +
|
", createTime=" + createTime +
|
", status='" + status + '\'' +
|
'}';
|
}
|
}
|
}
|
