T1.5: 台账异步落库（ArrayBlockingQueue(10000) offer 不阻塞；满200/500ms 单线程批量 saveAll+按日 upsert；队满原子计数+ERROR 不抛；关停 flush 余量；hibernate batch_size=200）
This commit is contained in:
chenxing
@@ -0,0 +1,267 @@
|
||||
package com.yunhai.platform.gateway.log;
|
||||
|
||||
import com.yunhai.platform.gateway.entity.OpenApiLog;
|
||||
import com.yunhai.platform.gateway.entity.OpenApiStatDaily;
|
||||
import com.yunhai.platform.gateway.filter.SanitizeFilter;
|
||||
import com.yunhai.platform.gateway.filter.SignVerifyFilter;
|
||||
import com.yunhai.platform.gateway.repository.OpenApiLogRepository;
|
||||
import com.yunhai.platform.gateway.repository.OpenApiStatDailyRepository;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import javax.annotation.PostConstruct;
|
||||
import javax.annotation.PreDestroy;
|
||||
import javax.servlet.http.HttpServletRequest;
|
||||
import java.time.LocalDateTime;
|
||||
import java.util.ArrayList;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ArrayBlockingQueue;
|
||||
import java.util.concurrent.Executor;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.ScheduledExecutorService;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
import java.util.concurrent.atomic.AtomicLong;
|
||||
|
||||
/**
|
||||
* 台账异步落库（T1.5，有界队列 + 背压）：全量请求观测（GET/401 也记），
|
||||
* 台账故障绝不拖垮请求线程——台账是观测设施，不能反杀业务。
|
||||
* <ul>
|
||||
* <li>{@link ArrayBlockingQueue} 容量 {@value #QUEUE_CAPACITY}，{@link #record} 只 offer 不阻塞；</li>
|
||||
* <li>单刷盘线程：满 {@value #BATCH_SIZE} 条或 {@value #FLUSH_INTERVAL_MS}ms 触发一批
|
||||
* JPA 批量插入 {@code open_api_log}，成功后按日 upsert {@code open_api_stat_daily}；</li>
|
||||
* <li>offer 失败（队列满）→ 原子计数 +1 + ERROR 日志（含累计丢弃数），绝不抛异常；</li>
|
||||
* <li>关停钩子：应用停止时 flush 余量（尽力而为）。</li>
|
||||
* </ul>
|
||||
* 日统计维度 caller：台账 caller 为 NULL（401 无 appKey）时归入 {@value #STAT_CALLER_UNKNOWN}
|
||||
* （stat 表该列 NOT NULL，观测聚合需要一个兜底桶）。
|
||||
*/
|
||||
@Component
|
||||
public class AccessLogService {
|
||||
|
||||
public static final int QUEUE_CAPACITY = 10_000;
|
||||
public static final int BATCH_SIZE = 200;
|
||||
public static final long FLUSH_INTERVAL_MS = 500L;
|
||||
public static final String STAT_CALLER_UNKNOWN = "unknown";
|
||||
/** ANON 判定下台账 caller 口径 */
|
||||
public static final String CALLER_ANONYMOUS = "anonymous";
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(AccessLogService.class);
|
||||
|
||||
private final ArrayBlockingQueue<OpenApiLog> queue;
|
||||
private final AtomicLong dropped = new AtomicLong();
|
||||
private final AtomicBoolean flushPending = new AtomicBoolean();
|
||||
|
||||
@Autowired
|
||||
private OpenApiLogRepository logRepository;
|
||||
|
||||
@Autowired
|
||||
private OpenApiStatDailyRepository statRepository;
|
||||
|
||||
/** 刷盘任务派发器：生产 = 单刷盘线程；测试注入同线程执行器作同步触发点 */
|
||||
private Executor flushDispatcher;
|
||||
private ScheduledExecutorService scheduler;
|
||||
|
||||
public AccessLogService() {
|
||||
this(QUEUE_CAPACITY);
|
||||
}
|
||||
|
||||
/** 测试构造：缩小容量以复现背压（生产路径固定 QUEUE_CAPACITY） */
|
||||
AccessLogService(int capacity) {
|
||||
this.queue = new ArrayBlockingQueue<>(capacity);
|
||||
}
|
||||
|
||||
void setFlushDispatcher(Executor flushDispatcher) {
|
||||
this.flushDispatcher = flushDispatcher;
|
||||
}
|
||||
|
||||
int queueSize() {
|
||||
return queue.size();
|
||||
}
|
||||
|
||||
long droppedCount() {
|
||||
return dropped.get();
|
||||
}
|
||||
|
||||
@PostConstruct
|
||||
void start() {
|
||||
scheduler = Executors.newSingleThreadScheduledExecutor(r -> {
|
||||
Thread t = new Thread(r, "gw-access-log-flush");
|
||||
t.setDaemon(true);
|
||||
return t;
|
||||
});
|
||||
if (flushDispatcher == null) {
|
||||
flushDispatcher = scheduler;
|
||||
}
|
||||
scheduler.scheduleWithFixedDelay(() -> {
|
||||
try {
|
||||
flushOnce();
|
||||
} catch (RuntimeException e) {
|
||||
log.error("[AccessLog] scheduled flush failed", e);
|
||||
}
|
||||
}, FLUSH_INTERVAL_MS, FLUSH_INTERVAL_MS, TimeUnit.MILLISECONDS);
|
||||
}
|
||||
|
||||
@PreDestroy
|
||||
void stop() {
|
||||
if (scheduler != null) {
|
||||
scheduler.shutdown();
|
||||
try {
|
||||
scheduler.awaitTermination(5, TimeUnit.SECONDS);
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
}
|
||||
// 关停钩子：flush 余量（尽力而为）
|
||||
try {
|
||||
while (flushOnce() > 0) {
|
||||
// 排空
|
||||
}
|
||||
} catch (RuntimeException e) {
|
||||
log.error("[AccessLog] shutdown flush failed", e);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 请求线程入口：只入队，绝不同步写库、绝不阻塞。
|
||||
*/
|
||||
public void record(OpenApiLog entry) {
|
||||
if (entry.getTs() == null) {
|
||||
entry.setTs(LocalDateTime.now());
|
||||
}
|
||||
if (!queue.offer(entry)) {
|
||||
long total = dropped.incrementAndGet();
|
||||
log.error("[AccessLog] queue FULL, entry dropped. accumulatedDropped={}", total);
|
||||
return;
|
||||
}
|
||||
if (queue.size() >= BATCH_SIZE) {
|
||||
scheduleFlush();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 按卡口径从请求组装并记录一条台账（T1.6 链上过滤器消费）：
|
||||
* path = 剥段后路径；caller = ANON 记 anonymous、PASS 记验签 appKey、
|
||||
* 否则尽力取 {@code X-App-Key} 头（401 无 appKey 记 NULL）；status = 网关出口状态。
|
||||
*
|
||||
* @param backendCode 提供方业务 code（响应为 JSON 信封时尽力解析，否则 null）
|
||||
*/
|
||||
public void recordRequest(HttpServletRequest req, int status, Integer backendCode, int costMs) {
|
||||
OpenApiLog entry = new OpenApiLog();
|
||||
entry.setTs(LocalDateTime.now());
|
||||
entry.setTargetCode(truncate(targetOf(req), 32));
|
||||
entry.setCallerAppKey(truncate(callerOf(req), 64));
|
||||
entry.setMethod(req.getMethod());
|
||||
Object stripped = req.getAttribute(SanitizeFilter.ATTR_STRIPPED_PATH);
|
||||
entry.setPath(truncate(stripped instanceof String ? (String) stripped : req.getRequestURI(), 256));
|
||||
entry.setStatus(status);
|
||||
entry.setBackendCode(backendCode);
|
||||
entry.setCostMs(costMs);
|
||||
record(entry);
|
||||
}
|
||||
|
||||
private void scheduleFlush() {
|
||||
Executor dispatcher = flushDispatcher;
|
||||
if (dispatcher == null) {
|
||||
return; // 未初始化（单元/异常场景）：500ms 周期兜底，不阻塞请求线程
|
||||
}
|
||||
if (flushPending.compareAndSet(false, true)) {
|
||||
dispatcher.execute(() -> {
|
||||
flushPending.set(false);
|
||||
try {
|
||||
flushOnce();
|
||||
} catch (RuntimeException e) {
|
||||
log.error("[AccessLog] batch flush failed", e);
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
/** 同步触发点（满批/周期/关停共用）：排空至多一批 → 批量插入 → 按日 upsert 统计 */
|
||||
synchronized int flushOnce() {
|
||||
List<OpenApiLog> batch = new ArrayList<>(BATCH_SIZE);
|
||||
queue.drainTo(batch, BATCH_SIZE);
|
||||
if (batch.isEmpty()) {
|
||||
return 0;
|
||||
}
|
||||
logRepository.saveAll(batch);
|
||||
upsertDaily(batch);
|
||||
return batch.size();
|
||||
}
|
||||
|
||||
private void upsertDaily(List<OpenApiLog> batch) {
|
||||
Map<String, OpenApiStatDaily> grouped = new LinkedHashMap<>();
|
||||
for (OpenApiLog e : batch) {
|
||||
String caller = e.getCallerAppKey() == null ? STAT_CALLER_UNKNOWN : e.getCallerAppKey();
|
||||
String key = e.getTs().toLocalDate() + "|" + e.getTargetCode() + "|" + caller;
|
||||
OpenApiStatDaily s = grouped.get(key);
|
||||
if (s == null) {
|
||||
s = new OpenApiStatDaily();
|
||||
s.setStatDate(e.getTs().toLocalDate());
|
||||
s.setTargetCode(e.getTargetCode());
|
||||
s.setCallerAppKey(caller);
|
||||
grouped.put(key, s);
|
||||
}
|
||||
s.setTotal(s.getTotal() + 1);
|
||||
if (isErr(e)) {
|
||||
s.setErrTotal(s.getErrTotal() + 1);
|
||||
}
|
||||
s.setAvgCostMs((int) ((long) s.getAvgCostMs() * (s.getTotal() - 1) + e.getCostMs()) / s.getTotal());
|
||||
}
|
||||
for (OpenApiStatDaily fresh : grouped.values()) {
|
||||
OpenApiStatDaily existing = statRepository.findFirstByStatDateAndTargetCodeAndCallerAppKey(
|
||||
fresh.getStatDate(), fresh.getTargetCode(), fresh.getCallerAppKey());
|
||||
if (existing != null) {
|
||||
int total = existing.getTotal() + fresh.getTotal();
|
||||
existing.setErrTotal(existing.getErrTotal() + fresh.getErrTotal());
|
||||
existing.setAvgCostMs((int) (((long) existing.getAvgCostMs() * existing.getTotal()
|
||||
+ (long) fresh.getAvgCostMs() * fresh.getTotal()) / total));
|
||||
existing.setTotal(total);
|
||||
statRepository.save(existing);
|
||||
} else {
|
||||
statRepository.save(fresh);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private static boolean isErr(OpenApiLog e) {
|
||||
return e.getStatus() >= 400 || (e.getBackendCode() != null && e.getBackendCode() != 0);
|
||||
}
|
||||
|
||||
/** target = 规整化 URI 首段；取不到记 unknown（列 NOT NULL，台账一条都不能丢在写入端） */
|
||||
private static String targetOf(HttpServletRequest req) {
|
||||
Object uri = req.getAttribute(SanitizeFilter.ATTR_SANITIZED_URI);
|
||||
String path = uri instanceof String ? (String) uri : req.getRequestURI();
|
||||
if (path == null || path.length() < 2) {
|
||||
return STAT_CALLER_UNKNOWN;
|
||||
}
|
||||
int second = path.indexOf('/', 1);
|
||||
return second < 0 ? path.substring(1) : path.substring(1, second);
|
||||
}
|
||||
|
||||
private static String callerOf(HttpServletRequest req) {
|
||||
Object verdict = req.getAttribute(SignVerifyFilter.ATTR_VERDICT);
|
||||
if (SignVerifyFilter.VERDICT_ANON.equals(verdict)) {
|
||||
return CALLER_ANONYMOUS;
|
||||
}
|
||||
if (SignVerifyFilter.VERDICT_PASS.equals(verdict)) {
|
||||
Object appKey = req.getAttribute(SignVerifyFilter.ATTR_AUTH_APPKEY);
|
||||
if (appKey instanceof String && !((String) appKey).trim().isEmpty()) {
|
||||
return (String) appKey;
|
||||
}
|
||||
}
|
||||
String header = req.getHeader("X-App-Key");
|
||||
return header == null || header.trim().isEmpty() ? null : header.trim(); // 401 无 appKey = NULL
|
||||
}
|
||||
|
||||
private static String truncate(String s, int max) {
|
||||
if (s == null) {
|
||||
return null;
|
||||
}
|
||||
return s.length() <= max ? s : s.substring(0, max);
|
||||
}
|
||||
}
|
||||
@@ -3,5 +3,11 @@ package com.yunhai.platform.gateway.repository;
|
||||
import com.yunhai.platform.gateway.entity.OpenApiStatDaily;
|
||||
import org.springframework.data.jpa.repository.JpaRepository;
|
||||
|
||||
import java.time.LocalDate;
|
||||
|
||||
public interface OpenApiStatDailyRepository extends JpaRepository<OpenApiStatDaily, Long> {
|
||||
|
||||
/** uk_stat 维度定位（T1.5 按日 upsert）：stat_date + target_code + caller_app_key */
|
||||
OpenApiStatDaily findFirstByStatDateAndTargetCodeAndCallerAppKey(
|
||||
LocalDate statDate, String targetCode, String callerAppKey);
|
||||
}
|
||||
|
||||
@@ -15,6 +15,12 @@ spring:
|
||||
hibernate:
|
||||
# 表结构由 docs/sql/01_schema.sql 交付，应用不自动建表
|
||||
ddl-auto: none
|
||||
properties:
|
||||
hibernate:
|
||||
# 台账批量刷盘（T1.5）：一批 JPA batch insert
|
||||
jdbc:
|
||||
batch_size: 200
|
||||
order_inserts: true
|
||||
redis:
|
||||
# 密钥链（sign:app:{appKey}）走共享 Redis db0——由平台配置层注入
|
||||
host: #
|
||||
|
||||
@@ -0,0 +1,107 @@
|
||||
package com.yunhai.platform.gateway.log;
|
||||
|
||||
import com.yunhai.platform.gateway.entity.OpenApiLog;
|
||||
import com.yunhai.platform.gateway.entity.OpenApiStatDaily;
|
||||
import com.yunhai.platform.gateway.repository.OpenApiLogRepository;
|
||||
import com.yunhai.platform.gateway.repository.OpenApiStatDailyRepository;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.mockito.ArgumentCaptor;
|
||||
import org.springframework.test.util.ReflectionTestUtils;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
import java.util.List;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.assertThatCode;
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.never;
|
||||
import static org.mockito.Mockito.verify;
|
||||
import static org.mockito.Mockito.verifyNoInteractions;
|
||||
|
||||
/**
|
||||
* AccessLogService 三用例（T1.5 TDD）：offer 入队成功路径（请求线程零写库）；
|
||||
* 满 200 条触发批量 flush（同步触发点测，不起真线程等待——注入同线程 dispatcher，
|
||||
* 一批 JPA saveAll + 按日 upsert stat_daily）；队列满 offer 失败 → 丢弃计数 +1 且绝不抛异常。
|
||||
*/
|
||||
class AccessLogServiceTest {
|
||||
|
||||
private AccessLogService service(int capacity, OpenApiLogRepository logRepo,
|
||||
OpenApiStatDailyRepository statRepo) {
|
||||
AccessLogService svc = new AccessLogService(capacity);
|
||||
ReflectionTestUtils.setField(svc, "logRepository", logRepo);
|
||||
ReflectionTestUtils.setField(svc, "statRepository", statRepo);
|
||||
svc.setFlushDispatcher(Runnable::run); // 同步触发点：不真起线程
|
||||
return svc;
|
||||
}
|
||||
|
||||
private OpenApiLog entry(String target, String caller, int status) {
|
||||
OpenApiLog e = new OpenApiLog();
|
||||
e.setTs(LocalDateTime.now());
|
||||
e.setTargetCode(target);
|
||||
e.setCallerAppKey(caller);
|
||||
e.setMethod("GET");
|
||||
e.setPath("/oapi/x");
|
||||
e.setStatus(status);
|
||||
e.setCostMs(10);
|
||||
return e;
|
||||
}
|
||||
|
||||
@Test
|
||||
void offer入队成功且请求线程零写库() {
|
||||
OpenApiLogRepository logRepo = mock(OpenApiLogRepository.class);
|
||||
OpenApiStatDailyRepository statRepo = mock(OpenApiStatDailyRepository.class);
|
||||
AccessLogService svc = service(AccessLogService.QUEUE_CAPACITY, logRepo, statRepo);
|
||||
|
||||
svc.record(entry("usermsg", "100617", 200));
|
||||
|
||||
assertThat(svc.queueSize()).as("入队 1 条").isEqualTo(1);
|
||||
assertThat(svc.droppedCount()).isZero();
|
||||
verifyNoInteractions(logRepo, statRepo); // 红线：不得在请求线程直接写库
|
||||
}
|
||||
|
||||
@Test
|
||||
void 满200条触发批量flush并upsert日统计() {
|
||||
OpenApiLogRepository logRepo = mock(OpenApiLogRepository.class);
|
||||
OpenApiStatDailyRepository statRepo = mock(OpenApiStatDailyRepository.class);
|
||||
AccessLogService svc = service(AccessLogService.QUEUE_CAPACITY, logRepo, statRepo);
|
||||
|
||||
for (int i = 0; i < AccessLogService.BATCH_SIZE; i++) {
|
||||
svc.record(entry("usermsg", "100617", i == 199 ? 500 : 200)); // 末条 HTTP≥400 计 err
|
||||
}
|
||||
|
||||
ArgumentCaptor<List<OpenApiLog>> batchCap = ArgumentCaptor.forClass(List.class);
|
||||
verify(logRepo).saveAll(batchCap.capture());
|
||||
assertThat(batchCap.getValue()).as("一批 200 条批量插入").hasSize(200);
|
||||
assertThat(svc.queueSize()).as("flush 后队列清空").isZero();
|
||||
|
||||
ArgumentCaptor<OpenApiStatDaily> statCap = ArgumentCaptor.forClass(OpenApiStatDaily.class);
|
||||
verify(statRepo).save(statCap.capture());
|
||||
OpenApiStatDaily stat = statCap.getValue();
|
||||
assertThat(stat.getTargetCode()).isEqualTo("usermsg");
|
||||
assertThat(stat.getCallerAppKey()).isEqualTo("100617");
|
||||
assertThat(stat.getTotal()).isEqualTo(200);
|
||||
assertThat(stat.getErrTotal()).as("status≥400 计入 err").isEqualTo(1);
|
||||
assertThat(stat.getAvgCostMs()).isEqualTo(10);
|
||||
}
|
||||
|
||||
@Test
|
||||
void 队列满丢弃计数加一且不抛异常() {
|
||||
OpenApiLogRepository logRepo = mock(OpenApiLogRepository.class);
|
||||
OpenApiStatDailyRepository statRepo = mock(OpenApiStatDailyRepository.class);
|
||||
AccessLogService svc = service(2, logRepo, statRepo); // 测试容量：小队列复现背压
|
||||
|
||||
svc.record(entry("usermsg", "a", 200));
|
||||
svc.record(entry("usermsg", "a", 200));
|
||||
assertThat(svc.queueSize()).isEqualTo(2);
|
||||
|
||||
assertThatCode(() -> {
|
||||
svc.record(entry("usermsg", "a", 200)); // 第 3 条 = 队列满
|
||||
svc.record(entry("usermsg", "a", 200));
|
||||
}).as("绝不阻塞/抛出，反杀业务不可接受").doesNotThrowAnyException();
|
||||
|
||||
assertThat(svc.droppedCount()).as("累计丢弃数").isEqualTo(2);
|
||||
assertThat(svc.queueSize()).isEqualTo(2);
|
||||
verify(logRepo, never()).saveAll(any());
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user