feat:优化签到排队等待

This commit is contained in:
2026-11-05 23:19:06 +08:00
parent 406ffe71af
commit 8a4b1c7ae8
4 changed files with 497 additions and 3 deletions
@@ -2,6 +2,7 @@ package com.budwk.app.zhgh.map;
import cn.dev33.satoken.annotation.SaCheckLogin;
import cn.dev33.satoken.annotation.SaCheckPermission;
import cn.dev33.satoken.annotation.SaMode;
import cn.hutool.core.util.StrUtil;
import cn.hutool.http.HttpUtil;
import cn.hutool.json.JSONUtil;
@@ -15,9 +16,13 @@ import io.swagger.annotations.ApiOperation;
import org.nutz.ioc.loader.annotation.Inject;
import org.nutz.ioc.loader.annotation.IocBean;
import org.nutz.mvc.annotation.At;
import org.nutz.mvc.annotation.GET;
import org.nutz.mvc.annotation.Ok;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import javax.validation.Valid;
import java.io.IOException;
/**
* @ClassName TMapController
@@ -35,6 +40,9 @@ public class TMapController {
@Inject
private SysConfigService configService;
@Inject
private TencentLocationService tencentLocationService;
@At
@SaCheckLogin
@ApiOperation("关键词输入提示")
@@ -49,4 +57,36 @@ public class TMapController {
String response = HttpUtil.get(url);
return Result.success(JSONUtil.parse(response));
}
/**
* 代理腾讯定位组件的坐标转换请求,异步排队领取共享调用名额。
*
* @param request 腾讯组件的单点坐标及 JSONP 回调请求
* @param response 腾讯协议的异步响应
* @throws IOException 即时错误响应写入失败时抛出
*/
@At("/ws/coord/v1/translate")
@GET
@Ok("void")
@SaCheckLogin
@SaCheckPermission(value = {"trainSignUp.mine", "h5.trainSignUp.mine"}, mode = SaMode.OR)
public void translate(HttpServletRequest request, HttpServletResponse response) throws IOException {
tencentLocationService.proxy(true, KEY, request, response);
}
/**
* 代理腾讯定位组件的逆地址解析请求,保留腾讯返回状态及请求编号。
*
* @param request 腾讯组件的 GCJ-02 坐标及 JSONP 回调请求
* @param response 腾讯协议的异步响应
* @throws IOException 即时错误响应写入失败时抛出
*/
@At("/ws/geocoder/v1/")
@GET
@Ok("void")
@SaCheckLogin
@SaCheckPermission(value = {"trainSignUp.mine", "h5.trainSignUp.mine"}, mode = SaMode.OR)
public void geocoder(HttpServletRequest request, HttpServletResponse response) throws IOException {
tencentLocationService.proxy(false, KEY, request, response);
}
}
@@ -0,0 +1,76 @@
package com.budwk.app.zhgh.map;
import org.nutz.boot.starter.WebFilterFace;
import org.nutz.ioc.loader.annotation.IocBean;
import javax.servlet.DispatcherType;
import javax.servlet.Filter;
import javax.servlet.FilterRegistration;
import javax.servlet.ServletContext;
import java.util.EnumSet;
import java.util.Map;
/**
* 为定位代理经过的过滤器开启 Servlet 异步支持,保留原有过滤器顺序及鉴权流程。
* NutzBoot 的 WebFilterFace 没有异步配置项,需在过滤器注册完成后补充该能力。
*/
@IocBean
public class TencentLocationAsyncSupport implements WebFilterFace {
/** @return 定位异步支持的注册名称 */
@Override
public String getName() {
return "tencentLocationAsyncSupport";
}
/** @return 定位代理所在的业务路径 */
@Override
public String getPathSpec() {
return "/platform/TMap/ws/*";
}
/** @return 仅处理原始请求,不重新分派鉴权链 */
@Override
public EnumSet<DispatcherType> getDispatches() {
return EnumSet.of(DispatcherType.REQUEST);
}
/** @return null,仅补充注册配置,不增加或替换业务过滤器 */
@Override
public Filter getFilter() {
return null;
}
/** @return 无额外过滤器参数 */
@Override
public Map<String, String> getInitParameters() {
return Map.of();
}
/** @return 最后执行,确保 Nutz 和登录过滤器均已完成注册 */
@Override
public int getOrder() {
return Integer.MAX_VALUE;
}
/**
* 在容器启动阶段补齐定位代理过滤器链的异步标记;其他入口仍按原有同步流程执行。
*
* @param context 当前应用 Servlet 上下文
*/
@Override
public void setServletContext(ServletContext context) {
String[] paths = {"/platform/TMap/ws/coord/v1/translate", "/platform/TMap/ws/geocoder/v1/"};
for (FilterRegistration registration : context.getFilterRegistrations().values()) {
boolean matches = registration.getServletNameMappings().contains("*");
for (String pattern : registration.getUrlPatternMappings()) {
for (String path : paths) {
matches |= pattern.equals(path) || (pattern.endsWith("/*")
&& path.startsWith(pattern.substring(0, pattern.length() - 1)));
}
}
if (matches && registration instanceof FilterRegistration.Dynamic) {
((FilterRegistration.Dynamic) registration).setAsyncSupported(true);
}
}
}
}
@@ -0,0 +1,357 @@
package com.budwk.app.zhgh.map;
import cn.hutool.crypto.digest.DigestUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.nutz.integration.jedis.RedisService;
import org.nutz.ioc.loader.annotation.Inject;
import org.nutz.ioc.loader.annotation.IocBean;
import javax.servlet.AsyncContext;
import javax.servlet.AsyncEvent;
import javax.servlet.AsyncListener;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.net.URI;
import java.net.URLEncoder;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
/**
* 腾讯定位组件的业务代理:按 Key 和接口共享滑动窗口名额,异步等待并调用腾讯服务。
* 定位期间不访问业务数据库,只有页面取得有效位置后才进入现有签到事务。
*/
@Slf4j
@IocBean(create = "init", depose = "close")
public class TencentLocationService {
private static final int MAX_PENDING = 2000;
private static final long REQUEST_TIMEOUT_MS = 20000;
// Redis 时间统一多实例计时;完成请求不归还名额,避免将每秒限额误当作在途并发数。
// Redis 3.0 禁止 TIME 后在同一 Lua 内写入,因此先读取服务端时间,再原子维护窗口。
private static final String ACQUIRE = "local now=tonumber(ARGV[2]); "
+ "local last=redis.call('ZREVRANGE',KEYS[1],0,0,'WITHSCORES'); "
+ "if #last>0 then now=math.max(now,tonumber(last[2])); end; "
+ "redis.call('ZREMRANGEBYSCORE',KEYS[1],'-inf',now-1000); "
+ "if redis.call('ZCARD',KEYS[1])<80 then "
+ "redis.call('ZADD',KEYS[1],now,ARGV[1]); redis.call('PEXPIRE',KEYS[1],2000); return 0; end; "
+ "local first=redis.call('ZRANGE',KEYS[1],0,0,'WITHSCORES'); "
+ "return math.max(1,1000-now+tonumber(first[2]));";
@Inject
private RedisService redisService;
private final Map<String, Lane> lanes = new ConcurrentHashMap<>();
private final Set<Job> pending = ConcurrentHashMap.newKeySet();
private final Semaphore capacity = new Semaphore(MAX_PENDING);
private final Object lifecycle = new Object();
private volatile boolean closed;
private ScheduledExecutorService dispatcher;
private ExecutorService httpExecutor;
private HttpClient client;
/** 初始化共享 HTTP 客户端和队列调度器,不为每个等待请求创建线程。 */
public void init() {
httpExecutor = Executors.newFixedThreadPool(8, runnable -> {
Thread thread = new Thread(runnable, "tencent-location-http");
thread.setDaemon(true);
return thread;
});
client = HttpClient.newBuilder().executor(httpExecutor)
.connectTimeout(Duration.ofSeconds(3)).build();
dispatcher = Executors.newSingleThreadScheduledExecutor(runnable -> {
Thread thread = new Thread(runnable, "tencent-location-dispatch");
thread.setDaemon(true);
return thread;
});
dispatcher.scheduleWithFixedDelay(this::drain, 0, 10, TimeUnit.MILLISECONDS);
}
/**
* 接收腾讯组件的 JSONP 请求,校验后异步排队;队满或超时返回组件可识别的失败结果。
*
* @param translate true 为坐标转换,false 为逆地址解析,仅允许这两个固定接口
* @param key 服务端配置的腾讯 Key,不接受客户端替换
* @param request 已通过登录及签到权限检查的请求
* @param response 输出腾讯原始 JSON/JSONP 协议,不包装业务 Result
* @throws IOException 无法写入即时错误响应时抛出
*/
public void proxy(boolean translate, String key, HttpServletRequest request,
HttpServletResponse response) throws IOException {
String callback = request.getParameter("callback");
if (callback != null && !callback.matches("[A-Za-z_$][A-Za-z0-9_$]{0,127}")) {
response.setStatus(HttpServletResponse.SC_BAD_REQUEST);
write(response, null, error(400, "定位回调参数不合法"));
return;
}
String coordinate = request.getParameter(translate ? "locations" : "location");
String type = request.getParameter("type");
if (!validCoordinate(coordinate) || (translate && (type == null || !type.matches("[1-6]")))) {
write(response, callback, error(400, "定位坐标或坐标类型不合法"));
return;
}
String endpoint = translate ? "/ws/coord/v1/translate" : "/ws/geocoder/v1/";
// 只转发 SDK 实际使用的单点坐标;固定主机、服务端 Key 和 JSON 输出,禁止开放代理。
String query = "?output=json&key=" + URLEncoder.encode(key, StandardCharsets.UTF_8)
+ (translate ? "&type=" + type + "&locations=" : "&location=")
+ URLEncoder.encode(coordinate, StandardCharsets.UTF_8);
URI uri = URI.create("https://apis.map.qq.com" + endpoint + query);
String rateKey = "tencent:location:rate:" + DigestUtil.sha256Hex(key) + ":" + endpoint;
synchronized (lifecycle) {
if (closed || !request.isAsyncSupported()) {
write(response, callback, error(503, "定位代理暂不可用"));
return;
}
if (!capacity.tryAcquire()) {
write(response, callback, error(429, "定位请求排队已满,请稍后重试"));
return;
}
AsyncContext context;
try {
context = request.startAsync(request, response);
} catch (RuntimeException e) {
capacity.release();
write(response, callback, error(503, "定位异步请求无法启动"));
return;
}
Lane lane = lanes.computeIfAbsent(rateKey, Lane::new);
Job job = new Job(context, lane, endpoint, uri, callback);
pending.add(job);
try {
context.setTimeout(REQUEST_TIMEOUT_MS);
context.addListener(new AsyncListener() {
/**
* 容器提前完成时释放队列名额,禁止继续调用上游。
* @param event 容器完成事件
*/
@Override
public void onComplete(AsyncEvent event) {
finish(job, error(499, "定位请求已结束"), false);
}
/**
* 超时同时取消未执行任务及在途 HTTP 请求。
* @param event 容器超时事件
*/
@Override
public void onTimeout(AsyncEvent event) {
finish(job, error(504, "定位代理等待超时,请重新定位"), true);
}
/**
* 客户端连接错误后不继续消耗腾讯调用额度。
* @param event 客户端连接错误事件
*/
@Override
public void onError(AsyncEvent event) {
finish(job, error(499, "定位连接已断开"), false);
}
/**
* 本代理直接完成响应,不执行二次异步分派。
* @param event 异步分派事件
*/
@Override
public void onStartAsync(AsyncEvent event) {
}
});
if (!lane.queue.offer(job)) {
finish(job, error(429, "定位请求排队已满,请稍后重试"), true);
}
} catch (RuntimeException e) {
finish(job, error(503, "定位请求排队失败"), true);
}
}
}
private boolean validCoordinate(String value) {
if (value == null || value.length() > 80) return false;
String[] parts = value.split(",", -1);
if (parts.length != 2) return false;
try {
double lat = Double.parseDouble(parts[0]);
double lng = Double.parseDouble(parts[1]);
return Double.isFinite(lat) && Double.isFinite(lng) && Math.abs(lat) <= 90 && Math.abs(lng) <= 180;
} catch (NumberFormatException e) {
return false;
}
}
private void drain() {
if (closed) return;
for (Lane lane : lanes.values()) {
if (System.nanoTime() < lane.nextAttempt) continue;
// 每个接口只由队头申请名额,避免数百个任务同时轮询 Redis。
for (int i = 0; i < 80 && !closed; i++) {
Job job = lane.queue.peek();
if (job == null) break;
if (job.done) {
lane.queue.remove(job);
continue;
}
if (elapsedMillis(job.createdAt) >= REQUEST_TIMEOUT_MS) {
finish(job, error(504, "定位代理等待超时,请重新定位"), true);
continue;
}
try {
List<String> time = redisService.time();
long now = Long.parseLong(time.get(0)) * 1000 + Long.parseLong(time.get(1)) / 1000;
long wait = ((Number) redisService.eval(ACQUIRE, List.of(lane.rateKey),
List.of(job.id, Long.toString(now)))).longValue();
if (wait > 0) {
lane.nextAttempt = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(wait);
break;
}
lane.queue.remove(job);
send(job);
} catch (Exception e) {
// Redis 故障不能退化为直连,否则多个实例会共同击穿腾讯每秒额度。
log.warn("腾讯定位调度失败 requestId={} endpoint={} errorType={}",
job.id, job.endpoint, e.getClass().getSimpleName());
finish(job, error(503, "定位调度暂不可用,请稍后重试"), true);
break;
}
}
}
}
private void send(Job job) {
synchronized (job) {
if (job.done) return;
long remaining = REQUEST_TIMEOUT_MS - elapsedMillis(job.createdAt);
if (closed || remaining <= 0) {
finish(job, error(504, "定位代理等待超时,请重新定位"), true);
return;
}
job.startedAt = System.nanoTime();
HttpRequest request = HttpRequest.newBuilder(job.uri)
.timeout(Duration.ofMillis(Math.min(5000, remaining))).GET().build();
job.future = client.sendAsync(request, HttpResponse.BodyHandlers.ofString(StandardCharsets.UTF_8));
job.future.whenComplete((response, failure) -> {
if (failure != null) {
finish(job, error(502, "腾讯定位接口调用失败或超时"), true);
return;
}
try {
if (response.statusCode() != 200) {
finish(job, error(502, "腾讯定位接口响应异常"), true);
return;
}
JSONObject body = JSONUtil.parseObj(response.body());
if (body.getInt("status") == null) throw new IllegalArgumentException("Missing status");
finish(job, body, true);
} catch (RuntimeException e) {
finish(job, error(502, "腾讯定位接口返回格式异常"), true);
}
});
}
}
private void finish(Job job, JSONObject body, boolean reply) {
CompletableFuture<?> future;
synchronized (job) {
if (job.done) return;
job.done = true;
future = job.future;
}
if (future != null && !future.isDone()) future.cancel(true);
job.lane.queue.remove(job);
try {
long queueMs = job.startedAt == 0 ? elapsedMillis(job.createdAt)
: TimeUnit.NANOSECONDS.toMillis(job.startedAt - job.createdAt);
// 不记录 Key、请求 URL 或经纬度;腾讯 request_id 可与浏览器定位结果关联。
log.info("腾讯定位代理 requestId={} endpoint={} queueMs={} apiMs={} totalMs={} status={} tencentRequestId={}",
job.id, job.endpoint, queueMs, job.startedAt == 0 ? 0 : elapsedMillis(job.startedAt),
elapsedMillis(job.createdAt), body.getInt("status"), body.getStr("request_id", ""));
if (reply) write((HttpServletResponse) job.context.getResponse(), job.callback, body);
} catch (IOException | IllegalStateException e) {
log.debug("腾讯定位响应连接已结束 requestId={}", job.id);
} finally {
pending.remove(job);
capacity.release();
try {
job.context.complete();
} catch (IllegalStateException ignored) {
// 容器的超时或连接关闭回调可能已完成同一请求。
}
}
}
private static long elapsedMillis(long start) {
return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);
}
private static JSONObject error(int status, String message) {
return new JSONObject().set("status", status).set("message", message);
}
private static void write(HttpServletResponse response, String callback, JSONObject body) throws IOException {
response.setCharacterEncoding("UTF-8");
response.setContentType(callback == null ? "application/json;charset=UTF-8" : "application/javascript;charset=UTF-8");
response.setHeader("Cache-Control", "no-store");
response.setHeader("X-Content-Type-Options", "nosniff");
String json = JSONUtil.toJsonStr(body);
byte[] bytes = (callback == null ? json : callback + "(" + json + ");").getBytes(StandardCharsets.UTF_8);
response.setContentLength(bytes.length);
response.getOutputStream().write(bytes);
}
/** 应用关闭时取消排队及在途请求,释放异步响应和线程资源。 */
public void close() {
Job[] jobs;
synchronized (lifecycle) {
closed = true;
if (dispatcher != null) dispatcher.shutdownNow();
jobs = pending.toArray(new Job[0]);
}
for (Job job : jobs) finish(job, error(503, "定位服务正在关闭"), true);
if (httpExecutor != null) httpExecutor.shutdownNow();
}
private static class Lane {
private final String rateKey;
private final ArrayBlockingQueue<Job> queue = new ArrayBlockingQueue<>(MAX_PENDING);
private long nextAttempt;
private Lane(String rateKey) {
this.rateKey = rateKey;
}
}
private static class Job {
private final String id = UUID.randomUUID().toString();
private final long createdAt = System.nanoTime();
private final AsyncContext context;
private final Lane lane;
private final String endpoint;
private final URI uri;
private final String callback;
private volatile boolean done;
private long startedAt;
private CompletableFuture<HttpResponse<String>> future;
private Job(AsyncContext context, Lane lane, String endpoint, URI uri, String callback) {
this.context = context;
this.lane = lane;
this.endpoint = endpoint;
this.uri = uri;
this.callback = callback;
}
}
}
@@ -594,7 +594,9 @@ layout("/layouts/platform_h5.html"){
})
scanRuntime.circle = new TMap.MultiCircle({map: scanRuntime.map,
geometries: [{id: "range", center: center, radius: this.signRadius}]})
scanRuntime.locator = new LBS.WebComponent.Geolocation({key: key, referer: "zhgh_zjipc-trainSignUp"})
// 腾讯组件仍负责真实定位,两步 WebService 请求经业务代理统一排队,避免直连突破共享额度。
scanRuntime.locator = new LBS.WebComponent.Geolocation({key: key, referer: "zhgh_zjipc-trainSignUp",
domain: window.location.origin + "/platform/TMap"})
this.$set(this, "mapReady", true)
return this.refreshPosition()
}).catch(error => {
@@ -647,6 +649,7 @@ layout("/layouts/platform_h5.html"){
this.$set(this, "locationMessage", "正在获取您的当前位置")
scanRuntime.markers.remove(["current"])
const requestId = ++scanRuntime.requestId
const locationStarted = performance.now()
return new Promise((resolve, reject) => {
let completed = false
let timer
@@ -671,6 +674,10 @@ layout("/layouts/platform_h5.html"){
this.$set(this, "distance", distance)
this.$set(this, "locationMessage", distance <= this.signRadius ? "已进入签到范围" : "尚未进入签到范围")
this.fitMap(coords)
// 定位阶段包含设备取位、坐标转换和逆地址解析,不记录本人经纬度。
console.info("签到阶段耗时", {stage: "location", requestId: requestId,
elapsedMs: Math.round(performance.now() - locationStarted), success: true,
tencentRequestId: result.gcoder_request_id})
resolve(coords)
} catch (err) { reject(err) }
}
@@ -678,7 +685,8 @@ layout("/layouts/platform_h5.html"){
if (window.isSecureContext === false) {
finish(new Error("当前页面不是安全环境,请使用HTTPS地址定位")); return
}
timer = window.setTimeout(() => finish(new Error("定位超时,请检查网络和定位权限后重试")), 12000)
// 为共享额度排队预留时间;外层兜底晚于 SDK 截止,保留其具体失败原因。
timer = window.setTimeout(() => finish(new Error("定位超时,请检查网络和定位权限后重试")), 30000)
try {
scanRuntime.locator.getLocation(result => finish(null, result), detail => {
const status = Number(detail && detail.status)
@@ -688,9 +696,11 @@ layout("/layouts/platform_h5.html"){
// 只记录错误诊断,不记录本人经纬度。
console.warn("腾讯签到定位失败", {status: status, message: detail && detail.message, logs: detail && detail.logs})
finish(new Error(message))
}, {timeout: 10000, highAccuracy: true, maximumAge: 0})
}, {timeout: 25000, highAccuracy: true, maximumAge: 0})
} catch (error) { finish(new Error("腾讯定位调用失败,请重新加载地图")) }
}).catch(error => {
console.warn("签到阶段耗时", {stage: "location", requestId: requestId,
elapsedMs: Math.round(performance.now() - locationStarted), success: false})
if (!scanRuntime.destroyed) {
this.$set(this, "locationFailed", true)
this.$set(this, "locationMessage", error.message)
@@ -762,7 +772,18 @@ layout("/layouts/platform_h5.html"){
return this.getUserLocation().then((coords) => {
if (scanRuntime.destroyed) return null
if (this.distance > this.signRadius) throw new Error("尚未进入签到范围,请到达地点后重试")
const signStarted = performance.now()
const requestId = scanRuntime.requestId
return this.$axios.post("/platform/trainSignUp/mine/drivingScan", {courseId: this.courseId, lat: coords.lat, lng: coords.lng})
.then(resp => {
console.info("签到阶段耗时", {stage: "sign", requestId: requestId,
elapsedMs: Math.round(performance.now() - signStarted), code: resp.code})
return resp
}, error => {
console.warn("签到阶段耗时", {stage: "sign", requestId: requestId,
elapsedMs: Math.round(performance.now() - signStarted), success: false})
throw error
})
}).then((resp) => {
if (!resp || scanRuntime.destroyed) return
if (!this.extensionEnabled) { this.$set(this, "legacyResult", resp); return }