📚 课程目录第4章审计数据收集策略

第四章：审计数据收集策略——同步写入 vs 异步写入、采样策略与流量控制

聊完了审计链路怎么搭，接下来就得解决一个更现实的问题：数据到底怎么收集？

说白了，你设计了一整套审计规则，也埋好了点，但如果数据收集策略没设计好，那这套东西根本跑不起来。我之前就见过一个项目，审计模块上线第二天，直接把业务数据库写挂了——原因就是审计日志的写入逻辑没处理好。

这一章，我带你过一遍审计数据收集中最关键的三个决策点：写入模式选同步还是异步、怎么设计采样策略、以及如何控制流量不把系统打垮。

写入模式：同步 vs 异步

这是个老生常谈的问题，但在审计场景下，选择会更纠结一些。

同步写入

同步写入的意思就是：业务请求处理完后，审计日志必须落盘，才算完事。如果审计库挂了，业务也跟着受影响。

它的优点是——数据可靠性极高。你几乎不会丢数据，因为写入失败就意味着业务请求失败。

缺点也很明显——增加响应延时。我之前在某个金融项目中测过，加上同步审计后，接口平均响应时间多了 20ms 左右。对于高 QPS 的场景，这个成本是致命的。

  适用场景：要求审计数据必须 100% 不丢失的合规场景，比如金融交易、医疗记录变更等。

异步写入

异步写入则是把审计日志先扔进内存队列或者消息队列，业务直接返回，后台再慢慢写。

这么做的好处？几乎不影响主链路的性能。我习惯的做法是用一个独立的线程池或者 Kafka 来消费审计日志，消费者挂了，只要队列不丢，数据总能补回来。

但坏处也很直白——有数据丢失的风险。如果服务突然挂了，内存队列里的数据可能就没了。另外，异步写入会导致审计数据的实时性变差，你没法立刻查到刚刚发生的操作。

注意：异步写入不等于完全没有可靠性保障。至少要做到生产者确认 + 消费者幂等，最好再配一个本地文件兜底，防止服务崩溃导致队列丢失。

下面这个表格帮你快速对比两者的差异：

对比维度	同步写入	异步写入
数据可靠性	极高，近乎不丢失	中等，存在丢失窗口
对主链路影响	增加响应延时	几乎无影响
实时性	立刻可查	延迟可配置
复杂度	低	较高，需要处理重试、幂等
推荐场景	强合规、低 QPS	高 QPS、性能敏感

我的经验：大多数情况下，我建议优先选异步写入。只要把消息队列的可靠性做好，异步就是审计场景下的最优解。同步写入更像个“不得已的选择”。

采样策略：不是所有请求都值得审计

你有没有想过一个问题：每天几千个亿的 API 请求，你真的需要审计每一条吗？

我的答案是不需要。全量审计理论上最保险，但成本太高。不仅存储成本高，查询和分析也慢。这时候，采样策略就派上用场了。

常见的采样策略

固定采样率：比如只记录 10% 的请求。简单粗暴，但可能错过关键事件。
动态采样率：根据流量动态调整采样比例。高峰期降采样，低峰期提采样。
全量采样 + 异常过滤：正常请求低采样，异常或敏感操作全量记录。
基于关键字段采样：对某些用户、操作或时段强制全量，其他按比例采样。

我个人习惯是混合策略。举个例子：

对所有写操作（增、删、改）做全量审计。
对读操作，根据用户等级或操作对象敏感度，设定不同的采样率。
在监控发现系统压力过大时，自动下调采样率。

这样做的好处是：既保证了关键数据不丢，又在高负载时保护了系统。

  核心原则：采样不是一刀切。它应该是一个可配置、可动态调整的策略，而不是一个写死的常量。

流量控制：别让审计把业务拖垮

这里要说的流量控制，不是防止外部攻击，而是防止内部审计模块自己把自己或者下游系统打崩。

我曾经遇到过一个问题：审计系统上线后，业务峰值期间，审计日志的写入量暴增，直接把 Elasticsearch 集群的写入队列撑爆了。结果就是，审计数据写入失败，业务链路倒没受影响，但审计数据丢失了一大块。

那怎么控制审计流量呢？我一般用这么几个手段：

限流：在审计写入端加上令牌桶或漏桶算法，控制每秒写入的日志条数。超过阈值就直接丢弃或降级。
削峰：用消息队列缓冲。业务方只管往队列里写，消费端按自己的节奏慢慢写。这是最常用的方案。
分级丢弃：当流量超过某个阈值，按重要性丢弃日志。比如先丢弃读操作的审计，保留写操作；再不济，只保留关键字段。
背压机制：当下游写入出现积压时，反压到上游，让业务方减缓审计生产速度。这个实现起来成本较高，但效果最好。

避坑：我曾经在项目里只做了限流，没做削峰。结果业务峰值一来，限流直接把审计数据全丢了。后来我才明白，限流只是最后一道防线，不能替代消息队列的削峰能力。

整体策略框架图

为了让你更直观地理解这三者怎么配合，我画了一张流程图。它展示了从业务请求到审计落盘的完整路径，以及同步/异步、采样和流量控制分别在哪个环节起作用。

如何落地？一个简单的伪代码示例

理论说了不少，咱们看一段实际的代码。这是我在项目中用过的采样 + 异步写入的简化版：

class AuditCollector:
    def __init__(self, sample_rate=0.1, max_qps=1000):
        self.sample_rate = sample_rate
        self.limiter = TokenBucket(rate=max_qps, burst=200)

    def should_sample(self, event):
        # 敏感操作全量
        if event.operation in ['DELETE', 'UPDATE']:
            return True
        # 其他操作按采样率
        return random.random() < self.sample_rate

    def collect(self, event):
        # 1. 采样决策
        if not self.should_sample(event):
            return  # 不审计，直接跳过

        # 2. 流量控制
        if not self.limiter.try_acquire():
            # 超过限流阈值，降级为采样
            return self._degrade(event)

        # 3. 异步写入消息队列
        self.queue.put(event)

这段代码很简单，但把采样、限流、降级都串起来了。你想想看，真正在生产里跑的时候，这段逻辑就是审计入口的保护神。