Keeping data in memory 的 Design approach

2026-10-24支付阿杰

结论

“内存优先、数据库溢出”这套思路可行，但不建议做成自动淘汰的通用 cache，也不要用 AOP 拦截 put()，再在背后写入数据库。

更稳妥的做法是把每个文件当作独立作业，并明确划分两种处理模式：

文件较小时，全程在内存中完成分组，并以流式方式生成 XML。
内存达到安全阈值时，把当前缓冲区整体迁移到 staging table。切换完成后，该作业的所有数据都写入数据库。
不要让同一个 group 长期处于”一部分在 cache、一部分在 database”的状态。

这种单向降级比逐条淘汰缓存更容易保证一致性，重试、监控和故障恢复也更简单。

为什么普通 cache 方案容易出问题

文件大小不等于内存占用

1 GB 的文本文件转换成 Java 对象后，实际占用的内存可能远超 1 GB。额外开销包括：

String、对象头和引用；
HashMap bucket 与 entry；
grouping key 的重复存储；
XML 构建过程中产生的临时对象；
JVM、线程栈和其他业务所需的内存。

因此，不能根据输入文件大小直接判断是否会发生 OutOfMemoryError，也不应该等到堆内存接近上限时才开始溢写。

按 entry 数量限制并不可靠

不同 payment instruction 的大小可能相差很大。“最多保存 10 万条”并不对应固定的内存上限。即使用了 Guava 的 maximumWeight 和 Weigher，得到的通常也只是业务数据大小的近似值，不是 JVM 对象的实际占用。

淘汰回调不适合负责持久化

如果在 cache eviction listener 或 AOP advice 中写数据库，会带来这些问题：

数据库写入可能阻塞正常的 put()；
写入失败后，很难判断 entry 能否安全淘汰；
批量写入方式和事务边界不明确；
同一个 group 可能同时存在于内存和数据库中；
读取时必须合并两边的数据，不能简单地在 cache miss 后查询数据库；
如果进程在淘汰和提交之间崩溃，可能丢失数据或产生重复数据。

cache eviction 的语义是”数据可以丢弃，之后还能重新获取”。这里的 payment instruction 是尚未处理完的业务数据，两者并不相同。

推荐的处理流程

1. 为每个输入文件创建 job

为每个作业分配唯一的 job_id，所有 staging 数据都带上这个字段。这样可以隔离并发处理的文件，恢复失败作业，并清理临时数据。

2. 小文件先在内存中聚合

可以使用类似下面的数据结构：

Map<GroupKey, List<PaymentInstruction>> groups = new HashMap<>();

不要把 JVM 的 Runtime.freeMemory() 作为主要判断依据。更实际的做法是维护一个保守的估算值，例如累计以下数据：

原始记录的字节数；
key 长度；
payment instruction 主要字段的长度；
为每条记录预留的固定对象开销。

这种估算不可能完全准确，因此阈值必须留出足够余量，并根据实际负载测试进行校准。

3. 超过阈值后整体切换到数据库模式

一旦超过预算，按以下顺序处理：

批量写入当前内存中的全部 group 和 instruction；
确认事务已经成功提交；
清空内存结构；
将当前作业标记为 SPILLED；
后续记录直接批量写入 staging table；
处理结束后，只从数据库生成输出。

模式只能从 IN_MEMORY 切换到 SPILLED，不要在两种模式之间反复切换。

final class AdaptivePaymentProcessor {
    private enum Mode {
        IN_MEMORY,
        SPILLED
    }

    private final long memoryBudgetBytes;
    private final StagingRepository repository;
    private final Map<GroupKey, List<PaymentInstruction>> groups =
            new HashMap<>();

    private Mode mode = Mode.IN_MEMORY;
    private long estimatedBytes;

    AdaptivePaymentProcessor(
            long memoryBudgetBytes,
            StagingRepository repository) {
        this.memoryBudgetBytes = memoryBudgetBytes;
        this.repository = repository;
    }

    void accept(String jobId, PaymentInstruction instruction) {
        if (mode == Mode.SPILLED) {
            repository.addToBatch(jobId, instruction);
            return;
        }

        GroupKey key = GroupKey.from(instruction);
        groups.computeIfAbsent(key, ignored -> new ArrayList<>())
              .add(instruction);

        estimatedBytes += estimateSize(key, instruction);

        if (estimatedBytes >= memoryBudgetBytes) {
            spill(jobId);
        }
    }

    private void spill(String jobId) {
        // repository.spillAll 必须保证：
        // 1. 整个缓冲区成功提交后才返回；
        // 2. 失败时抛出异常，内存数据保持不变。
        repository.spillAll(jobId, groups);

        groups.clear();
        estimatedBytes = 0;
        mode = Mode.SPILLED;
    }

    void finish(String jobId, GroupConsumer consumer) {
        if (mode == Mode.IN_MEMORY) {
            groups.forEach(consumer::accept);
        } else {
            repository.flushBatch();
            repository.forEachGroup(jobId, consumer);
        }
    }

    private long estimateSize(
            GroupKey key,
            PaymentInstruction instruction) {
        return key.estimatedBytes()
                + instruction.estimatedBytes()
                + 128; // 保守的结构开销预算，不代表 JVM 精确对象大小
    }
}

这里的 128 只用于说明估算方式，不能直接作为生产参数。具体数值需要结合 JDK、对象模型、字段结构和实际测试来确定。

数据库结构可以适当简化

如果 group table 只保存分组 key，没有其他状态，可以考虑只保留一张 staging table：

CREATE TABLE payment_staging (
    job_id          VARCHAR(64)  NOT NULL,
    group_key       VARCHAR(256) NOT NULL,
    sequence_no     BIGINT       NOT NULL,
    instruction_data TEXT        NOT NULL,
    PRIMARY KEY (job_id, sequence_no)
);

CREATE INDEX idx_payment_staging_group
    ON payment_staging (job_id, group_key, sequence_no);

处理时，按 job_id, group_key, sequence_no 排序读取：

SELECT group_key, sequence_no, instruction_data
FROM payment_staging
WHERE job_id = ?
ORDER BY group_key, sequence_no;

顺序扫描时可以直接识别 group 边界，不一定要先读取 group table，再逐组查询。如果 group 自身还有状态、汇总金额、发送结果或审计信息，保留独立的 group table 会更合适。

XML 应该流式生成

即使分组数据来自数据库，也不要先把整个 XML 文档构建成 DOM 再发送。可以使用 StAX 等流式 API，读取 group 的同时写入 XML：

XMLOutputFactory factory = XMLOutputFactory.newFactory();
XMLStreamWriter writer =
        factory.createXMLStreamWriter(outputStream, "UTF-8");

writer.writeStartDocument("UTF-8", "1.0");
writer.writeStartElement("Document");

repository.forEachGroup(jobId, (groupKey, instructions) -> {
    try {
        writer.writeStartElement("PaymentGroup");
        writer.writeAttribute("key", groupKey.value());

        for (PaymentInstruction instruction : instructions) {
            writeInstruction(writer, instruction);
        }

        writer.writeEndElement();
    } catch (XMLStreamException e) {
        throw new IllegalStateException("Failed to generate XML", e);
    }
});

writer.writeEndElement();
writer.writeEndDocument();
writer.close();

如果单个 group 本身可能很大，instructions 也必须通过游标、迭代器或流式回调读取，不能先装入完整的 List。

数据库之外的选择

如果 grouping key 能够稳定计算，还可以采用外部排序或 hash partition：

按 group_key 对输入记录进行外部排序，然后顺序生成 XML；
根据 hash(group_key) % N 把记录写入多个临时分区文件，再逐个处理分区。

这些方案可以减轻数据库负载，但需要自行处理临时文件、磁盘空间、崩溃恢复、清理和并发控制。如果已经有可靠的数据库，staging table 通常更容易运维。

必须补上的稳定性设计

无论使用哪种溢出介质，都需要明确以下规则：

数据库切换或批量提交失败时，不能清除内存数据；
staging 数据必须通过 job_id 隔离；
使用唯一键或幂等写入机制，防止重试产生重复记录；
保留 sequence_no，避免分组后丢失必要的输入顺序；
目标系统确认接收成功后，才能把 group 标记为完成；
区分”XML 已生成”和”目标系统已确认”，避免故障恢复后重复付款；
为失败作业设置可控的重试状态和人工处理状态；
定期清理已完成、失败和超时作业的 staging 数据；
数据库读取应使用 cursor/fetch size，避免 JDBC 驱动一次性加载全部结果；
并发处理多个文件时，应按整个进程分配总内存预算，不能按单个文件分别计算。

建议保留数据库 staging 能力，但把它定义为明确的作业执行模式，而不是藏在 HashMap 背后的透明二级存储。小文件使用纯内存路径；大文件超过保守阈值后，整个作业切换到数据库模式，再通过批量写入、顺序读取和流式 XML 输出完成处理。这样既能优化常见的小文件场景，也能避免为了少做几次数据库操作而引入难以验证的一致性问题。