使用 Drools 设计计费系统的高层算法

结论

逐条读取 CDR，将其插入 KieSession，执行规则后删除，同时在会话中保留累计状态，这种做法是合理的。处理数百万条记录时，通常可以采用以下方式：

• 使用数据库游标或分页进行流式读取，避免一次加载全部 CDR。
• 使用一个有状态的 KieSession。
• 为每个计费维度维护一个长期存在的聚合事实，例如 BillingState。
• 将每条 CDR 作为短生命周期事实，处理完立即删除。
• 规则修改聚合事实时，显式调用 modify 或 update。
• 按可独立计费的客户、账户或服务分区，限制单个会话的规模，并支持并行处理。

是否要为每条 CDR 调用一次 fireAllRules()，取决于规则是否要求严格的逐条顺序。逐条调用的语义最清楚，但执行数百万次会产生可观的固定开销。是否改用小批量处理，应以性能测试结果为准。

推荐的事实模型

不要让规则依赖已经删除的历史 CDR 来重新计算累计值。后续规则需要的信息，应保存在独立事实中：

public final class BillingState {
    private final String clientId;
    private final String serviceId;

    // 已经完成计费的 CDR 数量
    private long processedCount;

    private BigDecimal totalAmount = BigDecimal.ZERO;

    public BillingState(String clientId, String serviceId) {
        this.clientId = clientId;
        this.serviceId = serviceId;
    }

    public void recordCharge(BigDecimal amount) {
        processedCount++;
        totalAmount = totalAmount.add(amount);
    }

    public String getClientId() {
        return clientId;
    }

    public String getServiceId() {
        return serviceId;
    }

    public long getProcessedCount() {
        return processedCount;
    }

    public BigDecimal getTotalAmount() {
        return totalAmount;
    }
}

单条待处理记录可以携带当前处理阶段和计算结果：

public final class CdrWork {
    public enum Stage {
        NEW,
        PRICED,
        DONE
    }

    private final String clientId;
    private final String serviceId;
    private final BigDecimal units;

    private Stage stage = Stage.NEW;
    private BigDecimal charge;

    // 构造方法、getter 和 setter 省略
}

BillingState 应插入同一个有状态会话，并在整个计费分区的处理期间保留。Drools 可以据此：

• 匹配累计条件；
• 在状态变化后重新评估相关规则；
• 将累计状态与当前 CDR 联合匹配。

规则示例

假设前 500 条使用标准价格，第 501 条起使用优惠价格。这里的 processedCount 表示当前 CDR 之前已经完成计费的记录数。

rule "Standard price for first 500 CDRs"
when
    $cdr : CdrWork(
        stage == CdrWork.Stage.NEW,
        $clientId : clientId,
        $serviceId : serviceId
    )
    $state : BillingState(
        clientId == $clientId,
        serviceId == $serviceId,
        processedCount < 500
    )
then
    modify($cdr) {
        setCharge(calculateStandardCharge($cdr)),
        setStage(CdrWork.Stage.PRICED)
    }
end

rule "Discounted price after 500 CDRs"
when
    $cdr : CdrWork(
        stage == CdrWork.Stage.NEW,
        $clientId : clientId,
        $serviceId : serviceId
    )
    $state : BillingState(
        clientId == $clientId,
        serviceId == $serviceId,
        processedCount >= 500
    )
then
    modify($cdr) {
        setCharge(calculateDiscountedCharge($cdr)),
        setStage(CdrWork.Stage.PRICED)
    }
end

计价完成后，再更新累计状态：

rule "Record completed charge"
when
    $cdr : CdrWork(
        stage == CdrWork.Stage.PRICED,
        $clientId : clientId,
        $serviceId : serviceId,
        $charge : charge
    )
    $state : BillingState(
        clientId == $clientId,
        serviceId == $serviceId
    )
then
    modify($state) {
        recordCharge($charge)
    }

    modify($cdr) {
        setStage(CdrWork.Stage.DONE)
    }
end

这里用 stage 标明处理阶段，防止计价规则重复执行。流程较复杂时，可以使用规则流、议程组或显式的阶段事实来组织规则。不要主要依靠 salience 控制执行顺序，因为规则增多后，优先级通常很难维护。

Java 侧处理流程

一种典型实现是流式读取数据库记录，并根据计费键创建或查找累计事实：

KieSession session = kieContainer.newKieSession();

Map<BillingKey, BillingState> states = new HashMap<>();

try (Stream<Cdr> records = repository.streamByMonthAndService(month, serviceId)) {
    records.forEach(cdr -> {
        BillingKey key = new BillingKey(cdr.getClientId(), cdr.getServiceId());

        BillingState state = states.computeIfAbsent(key, ignored -> {
            BillingState newState =
                new BillingState(cdr.getClientId(), cdr.getServiceId());

            session.insert(newState);
            return newState;
        });

        CdrWork work = CdrWork.from(cdr);
        FactHandle cdrHandle = session.insert(work);

        session.fireAllRules();

        if (work.getStage() != CdrWork.Stage.DONE) {
            throw new IllegalStateException(
                "CDR was not fully processed: " + cdr.getId()
            );
        }

        persistBillingResult(cdr, work.getCharge());
        session.delete(cdrHandle);
    });
} finally {
    session.dispose();
}

如果 Java 代码直接修改已经插入会话的 BillingState，必须通知 Drools：

state.setProcessedCount(newCount);
session.update(stateHandle, state);

如果在规则右侧使用 modify($state) { ... }，Drools 会自动发出相应的更新通知。只调用普通 setter，却不执行 modify 或 update，依赖该状态的规则可能无法重新匹配。

是否每条记录都调用 fireAllRules()

以下情况适合逐条调用，因为这样最容易保证结果正确：

• 第 N 条记录的价格依赖前 N−1 条的处理结果；
• CDR 必须按时间、序号或其他确定顺序计费；
• 当前记录产生的状态必须先稳定下来，下一条记录才能进入。

不过，调用数百万次 fireAllRules() 会带来固定成本。可以改用小批量处理，但必须保留批内的顺序语义。简单地先插入 1,000 条 CDR，再调用一次 fireAllRules()，未必与逐条处理等价。多条记录可能同时根据相同的旧 processedCount 创建激活，从而在阈值附近产生错误价格。

如果业务能够预先确定处理顺序，可以在记录进入规则引擎前分配序号：

work.setSequenceNumber(sequence++);

规则随后可以直接根据序号判断价格区间。这种方式更适合批量插入，也能降低规则执行顺序对结果的影响。

聚合状态应该保存什么

BillingState 只需保存后续规则所需的信息，不必保留完整历史，例如：

• 已处理数量；
• 累计时长、流量或金额；
• 各价格区间的使用量；
• 最近一次事件时间；
• 已触发的优惠或封顶标志；
• 按业务窗口维护的计数器。

如果规则需要判断”最近 30 分钟的记录”、“任意两条历史记录之间的关系”，或者要在事后重新计算，那么单个计数器可能不够。可以考虑：

• 使用有界时间窗口和 Drools 事件处理；
• 使用外部聚合存储；
• 通过数据库或流处理系统预聚合；
• 采用两阶段计费，先计算统计信息，再执行定价规则。

不要使用 Drools 的 accumulate 来统计已从会话中删除的 CDR，因为删除后的事实不会继续参与匹配。accumulate 更适合仍保留在会话中的有限事实集合，或者边界明确的时间窗口。

分区与并行化

如果不同客户或账户之间完全独立，最好按计费边界分区，例如：

month + accountId + serviceId

每个分区可以使用独立的 KieSession，并按顺序处理分区内的 CDR。这样能够：

• 限制会话中长期事实的数量；
• 防止不同账户的规则相互匹配；
• 并行处理多个计费分区；
• 简化失败任务的重试和恢复。

不要让多个线程同时操作同一个 KieSession。通常可以让一个工作线程对应一个会话，或者使用会话池，确保同一时刻只有一个线程操作某个会话。

需要特别注意的问题

顺序必须明确

只有排序标准确定后，“第 500 条”才有明确含义。数据库查询必须采用稳定排序，例如按业务时间和唯一 ID 排序：

ORDER BY event_time, cdr_id

不要依赖数据库未声明的自然返回顺序。

明确阈值边界

如果”第 500 条之后”才使用优惠价格，那么应从第 501 条开始优惠，对应条件是：

processedCount >= 500

如果第 500 条本身就应使用优惠价格，则需要调整条件。边界测试应分别覆盖第 499、500 和 501 条记录。

保证幂等性

数据库写入失败、进程重启或消息重复投递，都可能使同一条 CDR 再次进入处理流程。应使用 CDR 唯一标识、计费批次号或唯一约束，防止重复计费，也避免累计值被重复增加。

不要让会话无限增长

除长期保留的 BillingState 外，临时结果、标记事实和辅助对象也应及时删除。否则，即使 CDR 已经删除，其他事实仍可能不断累积并占用大量内存。

复杂规则可能无法单遍计算

只有当累计状态能够完整概括历史，而且决策只依赖”当前记录＋已有状态”时，流式单遍处理才可行。如果价格取决于整月最终总量、跨记录排序、月底排名，或只能在后续确认的条件，就需要预聚合或分两遍处理，不能只靠一个在线计数器解决。

总体来看，在有状态会话中保留长期聚合事实，同时让 CDR 保持较短的生命周期，是合适的基础架构。设计时更需要关注计费顺序、状态模型、规则执行阶段、分区边界和失败恢复，而不只是是否将所有 CDR 插入同一个会话。

备注：内容仅供参考。