进审系统解决业务流量突增案例分享

仇智慧2024年09月12日约 12 分钟

本文字数：3765字

预计阅读时间：25分钟

01
引言

在现代信息化社会中，各种业务系统的数据信息通过审核系统进行预处理，并最终进入人工审核环节，以确保数据的合法性和安全性。然而，当某一业务线数据流量异常突增时，系统可能面临资源耗尽、处理滞后甚至崩溃的风险。本文将详细介绍如何通过优化审核系统中的处理链路，有效应对业务数据流量异常突增的问题。

02
审核系统概述

审核系统是一个集成多种业务系统的数据处理平台。不同业务系统的数据在进入审核系统后，经过一系列预处理步骤，最终保存到数据库中供人工审核。审核系统的处理链路由多个处理器依次执行，这些处理器包括联系方式过滤器、高亮处理器、MD5检测处理器、关键词检测处理器、免审检测处理器、用户标签检测处理器、AI检测处理器、通道变化处理器、基因检测处理器、产品策略处理器、发送消息处理器和保存数据库处理器。

03
现有处理链路及责任链模式

责任链模式是一种行为设计模式，它允许多个处理器依次处理请求，每个处理器都有机会处理请求或将其传递给下一个处理器。在我们的审核系统中，所有的处理器按照固定顺序执行，构成了一条处理链路。

代码示例：

public abstract class Handler {
    protected Handler next;    public void setNext(Handler next) {
        this.next = next;
    }
    public abstract void handle(Request request);
}
public class ContactFilterHandler extends Handler {
    @Override
    public void handle(Request request) {
        // 处理联系方式过滤
        if (next != null) {
            next.handle(request);
        }
    }
}
// 其他处理器类似定义

04
处理链路的挑战

尽管责任链模式为我们提供了清晰的处理逻辑，但当业务数据量异常增大时，系统资源可能被耗尽，导致处理链路中的处理器无法及时处理数据，进而影响整个审核系统的性能和稳定性。

05
第一次优化：异常检测器与处理链分离

设计思想

为了应对业务数据流量的异常增大，我们引入了一个异常检测器装置，利用滑动窗口原理，每10分钟统计当前窗口期的进审数据量，当超过设定的阈值时，认为该业务线数据异常，否则认为正常。并根据检测结果将数据分流至不同的处理链路。

一. 异常检测器

实现细节

窗口大小：设定一个固定的时间窗口，例如10分钟；
滑动频率：每分钟更新一次窗口，即窗口滑动一步；
数据统计：在窗口内统计数据流量，计算总量；
异常检测：将窗口内的总数据量与设定的阈值进行比较，判断是否异常。

代码示例

import java.util.LinkedList;
import java.util.Queue;public class SlidingWindowDetector {
    private final int windowSize; // 窗口大小（单位：分钟）
    private final int slideStep; // 滑动步长（单位：分钟）
    private final int threshold; // 异常流量阈值
    private final Queue<Integer> dataPoints; // 存储窗口内的数据点
    private int currentTotal; // 当前窗口内的总数据量
    public SlidingWindowDetector(int windowSize, int slideStep, int threshold) {
        this.windowSize = windowSize;
        this.slideStep = slideStep;
        this.threshold = threshold;
        this.dataPoints = new LinkedList<>();
        this.currentTotal = 0;
    }
    public void addDataPoint(int dataPoint) {
        dataPoints.add(dataPoint);
        currentTotal += dataPoint;
        if (dataPoints.size() > windowSize) {
            currentTotal -= dataPoints.poll();
        }
    }
    public boolean isTrafficAbnormal() {
        return currentTotal > threshold;
    }
    public void slideWindow() {
        for (int i = 0; i < slideStep; i++) {
            if (!dataPoints.isEmpty()) {
                currentTotal -= dataPoints.poll();
            }
        }
    }
    public static void main(String[] args) {
        int windowSize = 10; // 窗口大小10分钟
        int slideStep = 1; // 每分钟滑动一次
        int threshold = 1000; // 设定的异常流量阈值
        SlidingWindowDetector detector = new SlidingWindowDetector(windowSize, slideStep, threshold);
        // 模拟数据流入
        for (int i = 0; i < 20; i++) {
            int dataPoint = (int) (Math.random() * 200); // 模拟每分钟的数据量
            detector.addDataPoint(dataPoint);
            System.out.println("Minute " + (i + 1) + ": Data Point = " + dataPoint + ", Current Total = " + detector.currentTotal);
            if (detector.isTrafficAbnormal()) {
                System.out.println("Traffic abnormal detected at minute " + (i + 1));
            }
            // 模拟每分钟滑动窗口
            detector.slideWindow();
        }
    }
}

详细说明

构造函数：初始化滑动窗口的大小、滑动步长和异常流量阈值；
addDataPoint：添加新的数据点，并更新当前窗口内的总数据量。如果窗口内的数据点超过了窗口大小，移除最早的数据点；
isTrafficAbnormal：判断当前窗口内的数据总量是否超过阈值，从而判断流量是否异常；
slideWindow：滑动窗口，将最早的数据点移出窗口，更新当前窗口内的总数据量；
main方法：模拟数据流入和滑动窗口的过程，检测异常流量。

二. 异常处理链路：当检测到数据异常时，数据进入精简的处理链路，去掉耗费资源的处理器，确保系统资源不被耗尽。

public class SimplifiedHandlerChain {
    private Handler firstHandler;    public SimplifiedHandlerChain() {
        // 构建简化的处理链路
        Handler contactFilter = new ContactFilterHandler();
        Handler saveDatabase = new SaveDatabaseHandler();
        contactFilter.setNext(saveDatabase);
        this.firstHandler = contactFilter;
    }
    public void handle(Request request) {
        firstHandler.handle(request);
    }
}

优化效果

通过引入异常检测器，我们能够在业务数据流量异常增大时，自动分流数据，确保系统资源的合理使用，避免系统崩溃。此外，正常业务数据仍然能够按原有处理链路处理，不受异常流量的影响。

06
第二次优化：异常检测器智能化

为了实现不设定固定阈值、而是根据业务流量历史波动曲线判定流量是否异常突增，我们利用滑动窗口结合统计分析的方法来动态检测流量的异常。具体思路是通过滑动窗口计算当前窗口内流量的均值和标准差，然后根据一定的统计规则判断当前流量是否异常。

设计思路

滑动窗口：设定一个固定大小的滑动窗口，用于统计一定时间内的数据量；
历史数据分析：通过滑动窗口计算当前窗口内流量的均值和标准差；
异常检测：利用统计学方法（如3倍标准差原则）判断当前流量是否异常。

代码示例

import java.util.LinkedList;
import java.util.Queue;public class DynamicSlidingWindowDetector {
    private final int windowSize; // 窗口大小（单位：分钟）
    private final Queue<Integer> dataPoints; // 存储窗口内的数据点
    private int currentTotal; // 当前窗口内的总数据量
    private int currentCount; // 当前窗口内的数据点数量
    public DynamicSlidingWindowDetector(int windowSize) {
        this.windowSize = windowSize;
        this.dataPoints = new LinkedList<>();
        this.currentTotal = 0;
        this.currentCount = 0;
    }
    public void addDataPoint(int dataPoint) {
        dataPoints.add(dataPoint);
        currentTotal += dataPoint;
        currentCount++;
        if (currentCount > windowSize) {
            int removed = dataPoints.poll();
            currentTotal -= removed;
            currentCount--;
        }
    }
    public boolean isTrafficAbnormal() {
        if (currentCount < windowSize) {
            // 窗口未满，无法判断是否异常
            return false;
        }
        double mean = currentTotal / (double) currentCount;
        double variance = 0;
        for (int dataPoint : dataPoints) {
            variance += Math.pow(dataPoint - mean, 2);
        }
        variance /= currentCount;
        double stdDev = Math.sqrt(variance);
        int latestDataPoint = dataPoints.peek();
        return Math.abs(latestDataPoint - mean) > 3 * stdDev; // 3倍标准差原则
    }
    public static void main(String[] args) {
        int windowSize = 10; // 窗口大小10分钟
        DynamicSlidingWindowDetector detector = new DynamicSlidingWindowDetector(windowSize);
        // 模拟数据流入
        for (int i = 0; i < 20; i++) {
            int dataPoint = (int) (Math.random() * 200); // 模拟每分钟的数据量
            detector.addDataPoint(dataPoint);
            System.out.println("Minute " + (i + 1) + ": Data Point = " + dataPoint + ", Current Total = " + detector.currentTotal);
            if (detector.isTrafficAbnormal()) {
                System.out.println("Traffic abnormal detected at minute " + (i + 1));
            }
        }
    }
}

详细说明

构造函数：初始化滑动窗口的大小和存储数据点的队列；
addDataPoint：添加新的数据点，并更新当前窗口内的总数据量。如果窗口内的数据点超过了窗口大小，移除最早的数据点；
isTrafficAbnormal：计算当前窗口内数据点的均值和标准差，并根据3倍标准差原则判断最新的数据点是否异常；
main方法：模拟数据流入和异常检测的过程。

总结

利用滑动窗口结合统计分析方法，可以实现基于历史波动曲线的业务数据流量异常检测。通过计算滑动窗口内的均值和标准差，并利用3倍标准差原则判断异常流量，可以智能化的监控数据流量变化，及时发现和处理异常流量，保障系统的稳定运行。

07
第三次优化：回流器的引入

设计思想

虽然第一次优化能够有效分流异常流量，但所有异常流量中的数据都按简化处理链处理，可能导致部分正常数据被误杀。因此，我们引入回流器装置，将异常流量中的正常数据挑选出来，并回流至正常的处理链路进行处理。

实现细节

回流器：通过配置中心加载业务与回流规则，并定期更新规则引擎；

public class BackflowDetector {
    private Map<String, List<Rule>> rulesMap = new HashMap<>();    public void loadRules() {
        // 从配置中心加载回流规则
        rulesMap = ConfigurationCenter.loadRules();
    }
    public boolean shouldBackflow(Request request) {
        String businessLine = request.getBusinessLine();
        List<Rule> rules = rulesMap.get(businessLine);
        if (rules != null) {
            for (Rule rule : rules) {
                if (rule.evaluate(request)) {
                    return true;
                }
            }
        }
        return false;
    }
}

数据回流：如果命中回流规则，则数据回流至完整的处理链。

public class RequestProcessor {
    private BackflowDetector backflowDetector = new BackflowDetector();
    private NormalHandlerChain normalChain = new NormalHandlerChain();
    private SimplifiedHandlerChain simplifiedChain = new SimplifiedHandlerChain();    public void process(Request request) {
        if (backflowDetector.shouldBackflow(request)) {
            normalChain.handle(request);
        } else {
            simplifiedChain.handle(request);
        }
    }
}

优化效果

通过回流器的引入，我们能够将异常流量中的正常数据挑选出来，回流至完整的处理链，避免误杀。同时，系统资源能够得到更好的利用，确保业务的稳定运行。

08
UML类图和流程图

审核系统架构UML类图

@startuml
interface Handler {
    + setNext(Handler next)
    + handle(Request request)
}class ContactFilterHandler {
    + handle(Request request)
}
class HighlightHandler {
    + handle(Request request)
}
class Md5CheckHandler {
    + handle(Request request)
}
class KeywordCheckHandler {
    + handle(Request request)
}
class SaveDatabaseHandler {
    + handle(Request request)
}
Handler <|-- ContactFilterHandler
Handler <|-- HighlightHandler
Handler <|-- Md5CheckHandler
Handler <|-- KeywordCheckHandler
Handler <|-- SaveDatabaseHandler
class TrafficDetector {
    + isTrafficAbnormal(String businessLine)
}
class BackflowDetector {
    + loadRules()
    + shouldBackflow(Request request)
}
class RequestProcessor {
    + process(Request request)
}
TrafficDetector --|> Handler
BackflowDetector --|> Handler
RequestProcessor --|> Handler
@enduml

异常检测器流程图

@startuml
start
:获取当前业务线数据量;
if (数据量超过阈值?) then (是)
    :标记为异常数据;
else (否)
    :标记为正常数据;
endif
:分配到对应处理链路;
end
@enduml

回流器流程图

@startuml
start
:业务数据进入回流器;
:从配置中心加载回流规则;
:检查数据所属业务;
if (业务存在于回流规则中?) then (是)
    :按回流规则检测数据;
    if (命中回流规则?) then (是)
        :数据回流至完整处理链;
    else (否)
        :数据进入简化处理链;
    endif
else (否)
    :数据进入简化处理链;
endif
end
@enduml

09
进一步的优化过程，未来计划

动态调整处理链路

动态调整处理链路的目标是根据系统实时的负载情况和业务需求，动态调整处理器的执行顺序和启用状态。这样可以更灵活地应对不同的业务场景和数据量变化，提升系统的整体性能和稳定性。

动态配置中心：引入一个动态配置中心，用于存储和管理各处理器的配置。系统可以根据实时负载和业务需求，动态调整处理器的启用状态和执行顺序；

public class DynamicConfigCenter {
    private Map<String, List<HandlerConfig>> handlerConfigs = new HashMap<>();    public List<HandlerConfig> getHandlerConfigs(String businessLine) {
        return handlerConfigs.getOrDefault(businessLine, Collections.emptyList());
    }
    public void updateHandlerConfigs(String businessLine, List<HandlerConfig> configs) {
        handlerConfigs.put(businessLine, configs);
    }
}

处理链路调整：在请求处理时，根据动态配置中心的配置，构建适当的处理链路。

public class DynamicHandlerChain {
    private DynamicConfigCenter configCenter;    public DynamicHandlerChain(DynamicConfigCenter configCenter) {
        this.configCenter = configCenter;
    }
    public void handle(Request request) {
        List<HandlerConfig> configs = configCenter.getHandlerConfigs(request.getBusinessLine());
        Handler firstHandler = buildHandlerChain(configs);
        firstHandler.handle(request);
    }
    private Handler buildHandlerChain(List<HandlerConfig> configs) {
        Handler firstHandler = null;
        Handler currentHandler = null;
        for (HandlerConfig config : configs) {
            Handler handler = createHandler(config);
            if (firstHandler == null) {
                firstHandler = handler;
            } else {
                currentHandler.setNext(handler);
            }
            currentHandler = handler;
        }
        return firstHandler;
    }
    private Handler createHandler(HandlerConfig config) {
        // 根据配置创建处理器实例
        // ...
    }
}

智能流量检测与调度

智能流量调度旨在通过机器学习和数据分析技术，预测业务数据流量的变化，并提前做出相应调整。毕竟异常检测是在已经发生了业务流量异常之后才做出反应，有一定的滞后性，我们希望能够通过ai预测，提前预测业务流量变化，提前做出调整。畅想如下：

流量预测模型：构建并训练流量预测模型，预测未来一段时间内的业务数据流量；

from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestRegressor
import numpy as np# 假设我们有历史流量数据
data = np.array([
    # 时间, 业务线1流量, 业务线2流量, ...
])
X = data[:, :-1]
y = data[:, -1]
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)
model = RandomForestRegressor()
model.fit(X_train, y_train)
# 预测未来流量
future_data = np.array([
    # 未来时间点的特征
])
predicted_traffic = model.predict(future_data)

流量调度策略：根据预测的流量数据，动态调整处理链路和系统资源。

public class TrafficScheduler {
    private DynamicConfigCenter configCenter;
    private TrafficPredictor predictor;    public TrafficScheduler(DynamicConfigCenter configCenter, TrafficPredictor predictor) {
        this.configCenter = configCenter;
        this.predictor = predictor;
    }
    public void schedule() {
        Map<String, Integer> predictedTraffic = predictor.predict();
        for (Map.Entry<String, Integer> entry : predictedTraffic.entrySet()) {
            String businessLine = entry.getKey();
            int traffic = entry.getValue();
            adjustHandlerChain(businessLine, traffic);
        }
    }
    private void adjustHandlerChain(String businessLine, int traffic) {
        List<HandlerConfig> configs = configCenter.getHandlerConfigs(businessLine);
        // 根据预测流量调整处理链路
        // ...
    }
}

10
代码示例

处理器接口及实现

public interface Handler {
    void setNext(Handler next);
    void handle(Request request);
}public class ContactFilterHandler implements Handler {
    private Handler next;
    @Override
    public void setNext(Handler next) {
        this.next = next;
    }
    @Override
    public void handle(Request request) {
        // 处理联系方式过滤
        if (next != null) {
            next.handle(request);
        }
    }
}
// 其他处理器类似定义

异常检测器实现

public class TrafficDetector {
    private static final int THRESHOLD = 1000; // 设定阈值
    private Map<String, Integer> trafficData = new HashMap<>();    public boolean isTrafficAbnormal(String businessLine) {
        int currentTraffic = getCurrentTraffic(businessLine);
        return currentTraffic > THRESHOLD;
    }
    private int getCurrentTraffic(String businessLine) {
        // 获取当前窗口期的流量数据
        return trafficData.getOrDefault(businessLine, 0);
    }
}

回流器实现

public class BackflowDetector {
    private Map<String, List<Rule>> rulesMap = new HashMap<>();    public void loadRules() {
        // 从配置中心加载回流规则
        rulesMap = ConfigurationCenter.loadRules();
    }
    public boolean shouldBackflow(Request request) {
        String businessLine = request.getBusinessLine();
        List<Rule> rules = rulesMap.get(businessLine);
        if (rules != null) {
            for (Rule rule : rules) {
                if (rule.evaluate(request)) {
                    return true;
                }
            }
        }
        return false;
    }
}

处理链路实现

public class RequestProcessor {
    private BackflowDetector backflowDetector = new BackflowDetector();
    private NormalHandlerChain normalChain = new NormalHandlerChain();
    private SimplifiedHandlerChain simplifiedChain = new SimplifiedHandlerChain();    public void process(Request request) {
        if (backflowDetector.shouldBackflow(request)) {
            normalChain.handle(request);
        } else {
            simplifiedChain.handle(request);
        }
    }
}

11
综合效果和未来展望

通过以上三次优化和进一步的改进措施，我们的审核系统在应对业务数据流量异常增大时，能够有效分流和处理数据，确保系统资源的合理使用和业务的稳定运行。未来，我们还可以进一步优化处理器的性能，引入更多的智能检测手段，提高系统的处理效率和准确性。

12
总结

本文详细介绍了通过优化审核系统的处理链路，有效应对业务数据流量异常增大的技术方案。通过引入异常检测器和回流器装置，我们能够在保障系统资源不被耗尽的情况下，确保正常业务数据的及时处理。希望本文的内容能够为类似系统的优化提供参考和借鉴。

01引言

02审核系统概述

03现有处理链路及责任链模式

代码示例：

04处理链路的挑战

05第一次优化：异常检测器与处理链分离

设计思想

代码示例

详细说明

优化效果

06第二次优化：异常检测器智能化

设计思路

代码示例

详细说明

总结

07第三次优化：回流器的引入

设计思想

实现细节

优化效果

08UML类图和流程图

审核系统架构UML类图

异常检测器流程图

回流器流程图

09进一步的优化过程，未来计划

动态调整处理链路

智能流量检测与调度

10代码示例

处理器接口及实现

异常检测器实现

回流器实现

处理链路实现

11综合效果和未来展望

12总结

来自同一个声音

RAG命中率从60%到78%经验分享

别再一直调 prompt 了，让 Agent 的 Skills 自己进化！

OpenAI Codex 官方白皮书，最值得抄的 10 个工作流技巧

reified 如何骗过 JVM 类型擦除

领域驱动设计(DDD)在视频付费服务端的实践

未读消息数系统设计

01
引言

02
审核系统概述

03
现有处理链路及责任链模式

04
处理链路的挑战

05
第一次优化：异常检测器与处理链分离

06
第二次优化：异常检测器智能化

07
第三次优化：回流器的引入

08
UML类图和流程图

09
进一步的优化过程，未来计划

10
代码示例

11
综合效果和未来展望

12
总结