3. 数据录制原理：网络请求拦截、数据库变更捕获、日志文件解析

好，咱们进入正题。数据录制，说白了就是「把真实用户的操作过程给录下来」。你想想看，测试的时候最缺什么？缺真实数据。自己构造的数据总感觉差点意思，对吧？

我个人习惯把数据录制分成三个维度：网络层、数据层和文件层。这三个维度基本覆盖了绝大多数场景。今天咱们一个一个拆开讲。

3.1 网络请求拦截——最常用的录制方式

网络请求拦截，是我在项目中用得最多的方法。为什么？因为它简单、直接、覆盖广。前端发起的每一个请求，后端返回的每一个响应，都能被截获。

我记得有一次，要录制一个电商系统的下单流程。用户从浏览商品到支付成功，中间有十几个接口调用。如果用手工抓包，费时费力。但用拦截代理，一次录制，全部搞定。

3.1.1 代理拦截原理

说白了，就是在客户端和服务端之间插一个「中间人」。这个中间人能看到所有经过的流量。

具体怎么做？我常用的工具是 mitmproxy 和 Charles。这里我以 mitmproxy 为例，给你看一段核心代码：

# 一个简单的mitmproxy拦截脚本
from mitmproxy import http

def request(flow: http.HTTPFlow) -> None:
    """拦截请求，记录URL和请求体"""
    # 我只关心API请求，过滤掉静态资源
    if not flow.request.pretty_url.startswith("https://api."):
        return
    
    # 记录请求信息
    print(f"[请求] {flow.request.method} {flow.request.pretty_url}")
    print(f"[请求体] {flow.request.text[:500]}")  # 只取前500字符
    
    # 这里可以写入文件或数据库
    save_request(flow.request)

def response(flow: http.HTTPFlow) -> None:
    """拦截响应，记录状态码和响应体"""
    if not flow.request.pretty_url.startswith("https://api."):
        return
    
    print(f"[响应] {flow.response.status_code}")
    print(f"[响应体] {flow.response.text[:1000]}")
    
    save_response(flow.response)

3.1.2 HTTPS 证书问题

嗯，这里要注意。HTTPS 的流量是加密的，代理要能解密，就得安装根证书。我曾经在这个坑里摔过一次——录制了半小时，发现全是加密乱码，白干了。

3.1.3 录制数据的存储格式

录下来的数据怎么存？我建议用 JSON 格式，结构清晰，方便后续回放。每个请求-响应对作为一个独立记录。

{
  "id": "req_001",
  "timestamp": "2024-01-15T10:30:00.123Z",
  "method": "POST",
  "url": "https://api.example.com/order/create",
  "headers": {
    "Content-Type": "application/json",
    "Authorization": "Bearer xxx"
  },
  "request_body": {
    "product_id": "12345",
    "quantity": 2,
    "address_id": "67890"
  },
  "response_status": 200,
  "response_body": {
    "order_id": "ORD20240115001",
    "status": "success"
  },
  "duration_ms": 235
}

3.2 数据库变更捕获——后端视角的录制

网络请求拦截只能看到「输入输出」，但数据库里发生了什么变化？这个视角也很重要。特别是涉及事务、状态机变更的场景。

我习惯用 MySQL Binlog 或者 PostgreSQL 的 WAL（Write-Ahead Log） 来捕获数据变更。说白了，就是监听数据库的变更日志，把每一次 INSERT、UPDATE、DELETE 都记录下来。

3.2.1 Binlog 监听实战

以 MySQL 为例，用 canal 这个开源工具来监听 Binlog。部署起来很简单：

# 1. 开启 MySQL 的 Binlog（my.cnf 配置）
[mysqld]
log-bin=mysql-bin
binlog-format=ROW
server-id=1

# 2. 启动 canal（docker方式）
docker run -d --name canal \
  -e canal.instance.master.address=192.168.1.100:3306 \
  -e canal.instance.dbUsername=canal \
  -e canal.instance.dbPassword=canal123 \
  -e canal.instance.connectionCharset=UTF-8 \
  -e canal.instance.filter.regex=.*\\..* \
  registry.cn-hangzhou.aliyuncs.com/canal/canal-server:v1.1.6

监听到的数据长这样：

3.3 日志文件解析——最后的兜底方案

有些场景，既不能拦截网络请求，也不能监听数据库。比如：老旧系统、第三方服务、没有源码的遗留系统。这时候怎么办？

我告诉你，看日志。几乎所有系统都会打印日志。日志文件就是最后的宝藏。

3.3.1 日志格式解析

日志格式千奇百怪。我见过最乱的日志，一行里混杂了时间、级别、线程名、类名、消息，还有 JSON 串。怎么解析？用正则表达式。

# 一个典型的 Java 应用日志
2024-01-15 10:30:00.123 [http-nio-8080-exec-1] INFO  c.e.OrderService - 创建订单成功: {"orderId":"ORD001","amount":299.00}

# 对应的正则解析
import re

log_pattern = re.compile(
    r'(?P<timestamp>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d{3}) '
    r'\[(?P<thread>[^\]]+)\] '
    r'(?P<level>\w+) '
    r'(?P<logger>[^\s]+) '
    r'- (?P<message>.*)'
)

def parse_log_line(line: str) -> dict:
    match = log_pattern.match(line)
    if match:
        return match.groupdict()
    return None

3.3.2 关键信息提取

日志里什么都有，但我们要提取的是 业务操作的关键信息。比如：

• 操作类型：创建订单、支付、退款、取消
• 操作对象：订单ID、用户ID、商品ID
• 操作结果：成功、失败、异常信息
• 时间戳：精确到毫秒

3.4 三种录制方式的对比

你可能会问：这三种方式，到底用哪个？我直接给你一张对比表：

我个人习惯是：优先用网络请求拦截，因为它最直接。如果网络拦截搞不定，再考虑数据库变更捕获。日志解析是最后的兜底方案，但往往也是最费力的。

3.5 本章知识体系总览

说了这么多，我画了一张图帮你梳理一下。三种录制方式，各自有各自的定位，但最终目标都一样——把真实数据「录」下来，为后续的回放做准备。

好了，这一章的内容就到这里。三种录制方式，各有各的用武之地。你在实际项目中，可以根据系统的特点灵活选择。记住一个原则：能用网络拦截就用网络拦截，不行再考虑数据库，最后才是日志。这个顺序，是我踩过无数坑之后总结出来的。

可可精品资料库