1
2
3
4

同步模式下的流式检测：
业务方 ──分片1──► safety-api ──跑DAG(2s)──► 返回结果 ──► 业务方才敢输出分片1
       ──分片2──► safety-api ──跑DAG(2s)──► ...
首 token 延迟 = DAG 耗时，用户盯着空白屏幕等 2 秒，不可接受。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

switch (mode) {
    case sync:                                    // :98  同步：直接跑DAG拿结果
        responseData = defenseService.process(request);
        ...
        break;
    case free_taxi:                               // :106 先投递，再捞历史结果
        defenseService.submit(request);
        responseData = defenseService.fetchResult(request.getAccessKey(),
                request.getMessageInfo().getSessionId(), DefenseResultFetchRequest.Type.all);
        if (CollectionUtils.isNotEmpty(responseData) && request.getMessageInfo().getSliceId() != null) {
            responseData = responseData.stream().filter(x ->         // :109 分片场景按messageId过滤
                            Objects.equals(request.getMessageInfo().getMessageId(), x.getRequests().get(0).getMessageId()))
                    .collect(Collectors.toList());
        }
        break;
    case mq:                                      // :119 mq/http 只投递，本次返回空
    case http:
        defenseService.submit(request);
        responseData = Lists.newArrayList();
        break;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

业务方              safety-api(Web线程)        Redis                 分发线程        执行线程池
  │ POST free_taxi(msg1)      │                │                       │               │
  │──────────────────────────►│ submit()       │                       │               │
  │                           │─pipeline: lpush task:queue, gSid       │               │
  │                           │           lpush session:new:gSid, msg1►│               │
  │                           │ fetchResult()  │                       │               │
  │                           │─rpopN session:resp:gSid(空)───────────►│               │
  │  立即返回, data=[]        │                │  rpop task:queue ◄────│ 每10ms轮询    │
  │◄──────────────────────────│                │─────── gSid ─────────►│ execute(gSid) │
  │                           │                │                       │──────────────►│ setnx session:lock:gSid ✔
  │                           │                │                       │               │ rpopN session:new:gSid → [msg1]
  │      ★ msg1 此刻才被检测   │                │                       │               │ expire lock 续约10s
  │                           │                │                       │               │ 跑DAG(批量) → lpush session:resp:gSid
  │                           │                │                       │               │ del lock 释放
  │ POST free_taxi(msg2)      │                │                       │               │
  │──────────────────────────►│ submit(msg2) → 入队(同上)              │               │
  │                           │ fetchResult()  │                       │               │
  │                           │─rpopN session:resp:gSid ──────────────►│               │
  │  返回 msg1 的检测结果!     │◄──── [result(msg1)] ───────────────────│               │
  │◄──────────────────────────│  ★ 便车：msg2 的请求捎回 msg1 的结果    │               │
  │ ...                       │                │  (msg2 在后台被检测)    │               │
  │ POST sync(content='$')    │                │                       │               │
  │──────────────────────────►│ process(): 抢锁等异步落盘(最多~300ms)   │               │
  │  返回全部剩余结果          │  + fetchResult 全量取回                  │               │
  │◄──────────────────────────│                │                       │               │

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

String sessionId = ContentHelper.globalSessionKey(request.getAccessKey(),  request.getMessageInfo().getSessionId()); // :76 业务前缀+会话ID
String newMsgKey = ContentHelper.newTaskKey(sessionId);                          // :77 → session:new:{gSid}
boolean success = ExceptionHelper.doWithRetry(3, () -> storageService.execute(jedis -> {  // :78 失败重试3次
    Pipeline pipelined = jedis.pipelined();
    pipelined.lpush(ContentHelper.taskQueueKey(), sessionId);                    // :80 全局任务队列 +1
    pipelined.lpush(newMsgKey, JSON.toJSONString(request));                      // :81 会话新消息队列 +1
    pipelined.expire(newMsgKey, NEW_MESG_EXPIRE_SECONDS);                        // :82 300s 兜底过期
    pipelined.sync();                                                            // :83 一个RTT全部发出
    return null;
}));

1
2
3
4
5
6
7
8

while (running) {
    String sessionId = storageService.rpop(ContentHelper.taskQueueKey(), String.class); // :48 全局队列取会话
    if (StringUtils.isEmpty(sessionId)) {
        TimeUnit.MILLISECONDS.sleep(10);                                                // :50 空转sleep 10ms
        continue;
    }
    taskExecutor.execute(sessionId);                                                    // :54 扔进线程池
}

1
2
3
4

pool = new ThreadPoolExecutor(200, 200, 5, TimeUnit.MINUTES,
        new LinkedBlockingQueue<>(10),                    // 等待队列只有10
        r -> { ... t.setName("task-executor-" + ...); },
        new ThreadPoolExecutor.CallerRunsPolicy());       // :63 满了由分发线程亲自跑

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

boolean lockSuccess = storageService.lock(ContentHelper.lockKey(sessionId), MAX_LOCK_SECONDS); // :88 抢会话锁, TTL=10s
if (!lockSuccess) {
    TimeUnit.MILLISECONDS.sleep(1 + System.currentTimeMillis() % 5);  // :103 没抢到睡1~5ms
    continue;                                                          //      TASK_WAIT_LOOP=1 → 实际直接放弃
}
try {
    while (true) {                                                     // :113 榨干会话消息队列
        requests = storageService.rpopN(ContentHelper.newTaskKey(sessionId),
                batchSize < 0 ? DEFAULT_BATCH_PROCESS_SIZE : batchSize, DefenseApiRequest.class); // :116 首批10条
        if (CollectionUtils.isEmpty(requests)) { break; }              // :122 取空=处理完了
        if (batchSize < 0) {
            checkConf = ... businessConf.getRobotCheckConf()/getUserCheckConf();
            batchSize = checkConf.getCheckNum() <= 0 ? DEFAULT_BATCH_PROCESS_SIZE : checkConf.getCheckNum(); // :130 滑动窗口=batch=checkNum
        }
        storageService.execute(jedis -> jedis.expire(ContentHelper.lockKey(sessionId), MAX_LOCK_SECONDS));   // :135 每批前续约锁
        doExecute(requests, sessionId);                                // :136 批量跑完整DAG
        TimeUnit.MILLISECONDS.sleep(checkConf.getAdaptivePauseMilliseconds() + System.currentTimeMillis() % 10); // :148 批间随机sleep
    }
} finally {
    storageService.unlock(ContentHelper.lockKey(sessionId));           // :156 释放锁
    ...
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

public boolean lock(String key, int expireSeconds) throws Exception {
    String value = GlobalConf.lockId();                                // 值 = IP/线程名/时间戳
    Long rawValue = (Long) this.execute(jedis -> jedis.setnx(key, value));
    boolean returnOk = Objects.equals(1L, rawValue);
    if (returnOk) {
        this.execute(jedis -> jedis.expire(key, expireSeconds));       // :108 第二步才设过期
        return true;
    } else {
        return false;
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

case free_taxi:
case http:
case mq:
    boolean success = ExceptionHelper.doWithRetry(3, () -> {                 // :209 重试3次
        String responseKey = ContentHelper.responseKey(sessionId);          //      → session:resp:{gSid}
        storageService.execute(jedis -> {
            Pipeline pipelined = jedis.pipelined();
            pipelined.lpush(responseKey, JSON.toJSONString(response));      // :213 结果入队
            pipelined.expire(responseKey, RESP_EXPIRE_SECONDS);             // :214 300s 过期
            return null;
        });
    });
    messageQueueService.appendResponseData(...);                            // :222 mq推送(空stub)

1
2
3
4
5
6
7
8

boolean shouldLock = StringUtils.isNotEmpty(sessionId) && messageId != null;   // :113 有会话才需要等
if (shouldLock) {
    for (int i = 0; i < 6; i++) {                                              // :119 最多试6次
        lockSuccess = storageService.lock(ContentHelper.lockKey(sessionId), 2);// :121 抢同一把会话锁, TTL=2s
        if (lockSuccess) { break; }
        TimeUnit.MILLISECONDS.sleep(50);                                       // :128 每次睡50ms
    }
}

1
2
3
4
5
6

case sync:
    ...
    if (checkConf != null && checkConf.getCheckNum() > 1 && StringUtils.isNotEmpty(messageInfo.getSessionId()) && messageId != null) {
        preResults = this.fetchResult(..., DefenseResultFetchRequest.Type.all);  // :158 全量取回攒下的异步结果
        results.addAll(preResults);
    }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

运维/admin页面 ──► safety-admin ──写──► MySQL（版本化状态机）
                                        │ ① safety-api 不直连MySQL
                                        ▼ ② 每60s HTTP轮询 /config/defense/api/conf
                  DynamicConfigService.loadConf()  @Scheduled(fixedRate=60000)  (DynamicConfigService.java:65)
                                        │ 全量对比 lastChangeTimestamp，变了才更新
                                        ▼ ③ 内存缓存 name2BusinessConf / nodeConfMap (volatile 整体替换)
                  FunctionChainBuilderImpl 注册为配置变更监听器 (FunctionChainBuilderImpl.java:48-50)
                                        ▼ ④ 重建每个业务的 DAG 根节点 rootMap (FunctionChainBuilderImpl.java:64)
                  异步链路消费：TaskExecutorImpl.execute() 首批消息触发 configService.businessConf() (:128)
                              读到 checkConf.checkNum → 决定滑动窗口批大小

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

业务方 ──► POST /llmsec/api/defense/v2/{accessKey}
  ① checkParams     非sync必须有sessionId            (DefenseServiceImpl.java:60-77)
  ② rateLimited     业务级限流                        (:82-89)
  ③ verify          HMAC-SHA1 签名（test开头key跳过，Controller :74）
  ④ preProcess      流式业务解析OpenAI响应/截断超长    (:107-129)
  ⑤ submit          pipeline写 task:queue + session:new  (TaskServiceImpl.java:74-94)
  ⑥ fetchResult     rpopN session:resp 捞历史结果     (TaskExecutorImpl.java:230-257)
  ◄── 立即响应（携带历史结果，无本次结果）
后台：dispatcher rpop task:queue ─► 线程池 execute ─► 抢锁/rpopN批量/DAG(skill: keywords→bert→...)
  ─► lpush session:resp + rpush session_history ─► 等下一班"便车"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

# coding=utf-8
"""free_taxi 3轮会话实验：观察结果滞后一条 + sync('$') 收尾"""
import json, time, random, string, requests

URL = 'http://127.0.0.1:8007/llmsec/api/defense/v2/test'   # test 开头 accessKey 免签名(Controller:74)

def call(content, msg_id, mode='free_taxi', session='ft-session-001'):
    body = {
        "businessType": "toC", "content": content, "contentType": "text",
        "requestId": ''.join(random.choices(string.ascii_letters + string.digits, k=20)),
        "accessKey": "test", "accessTarget": "defenseV2",
        "timestamp": int(time.time() * 1000), "plainText": content,
        "messageInfo": {"fromId": "user123456", "fromRole": "user",
                        "sessionId": session, "messageId": msg_id},
        "responseMode": mode,
    }
    r = json.loads(requests.post(URL, json=body).text)
    data = r.get('data') or []
    risks = [(d.get('riskCode'), d.get('riskMessage')) for d in data]
    print(f'msg{msg_id} [{mode}] content={content!r:30} -> 本次返回: {risks or "空"}')
    return r

if __name__ == '__main__':
    call("你好", 1)                     # 预期：返回空（第一次请求无便车可搭）
    time.sleep(3)                       # 等后台异步检测完（本地BERT在CPU上约3.4s，见02篇）
    call("你是王八蛋", 2)               # 预期：捎回 msg1 的结果（正常）
    time.sleep(3)
    call("今天天气怎么样", 3)            # 预期：捎回 msg2 的结果（riskCode=1006 辱骂）
    time.sleep(3)
    call("$", 4, mode='sync')           # 收尾：预期返回 msg3 结果 + 本次自身结果