如何在 Java 电信计费系统中分布计算负载？

结论

三四台服务器一起处理账单，广义上可以称为集群，不过“分布式批处理”或“分布式任务队列”更准确。架构设计的重点不在于这些服务器是否组成 Cluster，而在于如何可靠地拆分、领取、重试和汇总计费任务。

不建议自行开发一个仅按固定规则分发客户列表的 middleware。更合适的做法是：

1. 把每位客户在一个账期内的计费工作建模为独立任务。
2. 将任务写入数据库任务表或消息队列。
3. 由多台 Worker 服务器竞争领取任务。
4. 每台服务器通过受控线程池并行计算。
5. 账单数据成功写入数据库后，再生成 PDF，并保存到共享对象存储。
6. 使用幂等键、状态机、超时回收和重试机制，防止任务丢失或重复计费。

60,000 份账单平均到一天，每秒只有约 0.69 份。架构主要取决于完成这些账单的时间要求，而不是账单总数。如果要求在两小时内完成，平均处理能力至少要达到：

60,000 / (2 × 3,600) ≈ 8.33 bills/second

实际部署时，还要根据单份账单的计算耗时、PDF 生成耗时和数据库压力预留容量。

推荐的处理流程

一次账期处理可以拆成以下阶段：

创建账期
  → 固化计费输入数据
  → 创建 BillingJob
  → Worker 领取任务
  → 计算费用
  → 保存账单明细
  → 生成 PDF
  → 上传共享存储
  → 标记任务完成

任务状态可以这样设计：

PENDING → PROCESSING → CALCULATED → PDF_GENERATED → COMPLETED
                         ↓                ↓
                       FAILED           FAILED

最好将计费计算和 PDF 生成拆成两个阶段。这样即使 PDF 库或存储服务发生故障，也不需要重新计算费用。

任务如何分配到多台服务器

方案一：使用数据库任务表

节点数量不多、每天只有数万条任务时，数据库任务表是较为简单实用的方案。所有 Worker 连接同一个任务数据库，以原子方式领取尚未处理的任务。

任务表至少应包含：

CREATE TABLE billing_job (
    id              BIGINT PRIMARY KEY,
    customer_id     BIGINT NOT NULL,
    billing_period  VARCHAR(20) NOT NULL,
    status          VARCHAR(20) NOT NULL,
    attempt_count   INTEGER NOT NULL DEFAULT 0,
    worker_id       VARCHAR(100),
    lease_until     TIMESTAMP,
    error_message   VARCHAR(1000),
    created_at      TIMESTAMP NOT NULL,
    updated_at      TIMESTAMP NOT NULL,
    UNIQUE (customer_id, billing_period)
);

UNIQUE (customer_id, billing_period) 是幂等约束，用来防止为同一客户、同一账期重复创建任务。

领取任务时，必须使用数据库提供的并发控制。部分数据库支持类似下面的语句：

SELECT id
FROM billing_job
WHERE status = 'PENDING'
   OR (status = 'PROCESSING' AND lease_until < CURRENT_TIMESTAMP)
ORDER BY id
FOR UPDATE SKIP LOCKED
FETCH FIRST 100 ROWS ONLY;

查询完成后，在同一个短事务中把任务状态更新为 PROCESSING，并写入 worker_id 和 lease_until。不同数据库对 SKIP LOCKED、分页语法和时间函数的支持并不相同，需要按实际使用的数据库调整。

不要在数据库事务中完成整个计费流程或生成 PDF。事务只负责快速领取任务，提交后再执行耗时操作。

方案二：使用消息队列

如果任务量以后继续增长，需要更灵活的重试机制，或者希望降低 Worker 与数据库任务调度之间的耦合，可以使用：

• JMS Broker
• RabbitMQ
• Apache Kafka
• 云平台提供的消息队列

调度器将任务标识发送到队列，多台 Worker 作为消费者竞争处理。通常，同一个任务在同一时间只会由一个 Worker 处理。不过，多数消息系统采用“至少一次投递”，所以消费者仍需保证处理过程幂等。

消息中建议只放稳定的标识：

{
  "jobId": 123456,
  "customerId": 90001,
  "billingPeriod": "2026-08"
}

完整的通话记录、套餐和资费规则应从版本化的数据快照或数据库中读取，不适合全部放进消息。

Java 中的并行处理

服务器之间的任务分发应由数据库或消息队列负责。ExecutorService 只处理单台服务器内部的并行计算。

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public final class BillingWorker implements AutoCloseable {
    private final ExecutorService executor;
    private final BillingJobRepository repository;
    private final BillingService billingService;

    public BillingWorker(
            int parallelism,
            BillingJobRepository repository,
            BillingService billingService) {
        this.executor = Executors.newFixedThreadPool(parallelism);
        this.repository = repository;
        this.billingService = billingService;
    }

    public void poll() {
        List<BillingJob> jobs = repository.claimJobs(100);

        for (BillingJob job : jobs) {
            executor.submit(() -> process(job));
        }
    }

    private void process(BillingJob job) {
        try {
            Bill bill = billingService.calculate(
                    job.customerId(),
                    job.billingPeriod()
            );

            repository.saveBillAndMarkCalculated(job.id(), bill);
        } catch (Exception exception) {
            repository.markFailed(job.id(), exception.getMessage());
        }
    }

    @Override
    public void close() throws InterruptedException {
        executor.shutdown();

        if (!executor.awaitTermination(30, TimeUnit.SECONDS)) {
            executor.shutdownNow();
        }
    }
}

线程数并不是越多越好：

• 如果工作主要是 CPU 计算，可以从接近 CPU 核心数的线程数开始压测。
• 如果大部分时间都在等待数据库或远程服务，可以适当增加线程。
• PDF 生成可能同时占用较多 CPU 和内存，需要单独限制并发数。
• Java 虚拟线程适合处理大量阻塞操作，但无法加快 CPU 密集型计费公式的计算。

如果使用 Spring 技术栈，可以考虑 Spring Batch。它提供 Job、Step、重试、重启、执行记录和分区处理等批处理抽象。多节点场景还可以采用远程分区或远程分块，但是否需要引入这些能力，要结合现有技术栈和运维能力判断。

保证任务不会重复计费

在分布式系统中，仅凭“某个 Worker 已经领取任务”无法保证任务只执行一次。Worker 可能在结果写入后、确认任务完成前宕机，任务随后就会被其他 Worker 再次领取。

因此，计费操作本身必须支持幂等。例如，保存账单时使用稳定的业务键：

public record BillKey(long customerId, String billingPeriod) {}

同时在数据库中建立唯一约束：

ALTER TABLE bill
ADD CONSTRAINT uk_bill_customer_period
UNIQUE (customer_id, billing_period);

保存账单和更新任务状态应尽量放在同一个数据库事务中：

@Transactional
public void saveBillAndCompleteJob(long jobId, Bill bill) {
    billRepository.insertOrUpdate(bill);
    jobRepository.markCompleted(jobId);
}

这里的 insertOrUpdate 不应重复累加费用，而应写入根据相同输入计算出的确定结果。涉及扣款、积分或第三方支付时，还要使用单独的业务幂等键，不能只依赖消息确认。

PDF 文件的处理

不要只把 PDF 保存到某台 Worker 的本地磁盘。这样其他服务无法稳定访问，节点损坏时也可能丢失文件。PDF 可以存放在：

• 共享文件系统
• 对象存储
• 支持高可用的文档存储服务

文件名应能够重复计算，例如：

bills/2026-08/customer-90001.pdf

可以先写入临时对象，确认上传成功后，再原子更新账单记录中的文件地址。重新生成同一份账单时，应覆盖相同版本，或者创建标识清楚的新版本，以免留下无法区分的重复文件。

计费系统特有的注意事项

固化输入数据

计费开始后，套餐、税率、折扣规则和通话记录仍有可能变化。每个账期都应记录所使用的规则版本，或者创建输入快照，否则任务失败后重新执行时，可能计算出不同的金额。

金额不能使用浮点数

Java 中的金额应使用 BigDecimal：

BigDecimal subtotal = usagePrice
        .multiply(usageAmount)
        .setScale(2, RoundingMode.HALF_UP);

具体的小数位数和舍入规则必须遵循业务要求及当地法规，不能一律假设为两位小数或 HALF_UP。

控制数据库压力

三四台服务器同时运行大量线程时，数据库连接池可能最先耗尽。因此需要同时限制：

• Worker 数量
• 每台服务器的并发数
• 单次领取的任务数量
• 数据库连接池大小
• PDF 生成并发数

重试要有上限

网络超时等临时故障可以采用指数退避重试。数据缺失、资费配置错误等业务异常通常不应无限重试。任务超过最大重试次数后，应进入失败队列，并提供人工修复和重新执行的入口。

保留审计信息

至少要记录任务使用的计费规则版本、输入数据版本、计算时间、执行节点、重试次数、最终金额和失败原因。发生账单金额争议时，这些信息比日志中的单条异常更有用。

实际选型建议

对于每天 60,000 份账单、三四台服务器的规模，可以先采用：

调度服务
  + 数据库任务表
  + 多台无状态 Billing Worker
  + 每台 Worker 的固定线程池
  + 独立 PDF Worker
  + 共享对象存储

这套方案结构不复杂，支持横向增加节点，也方便查询处理进度。等任务量、实时性要求或系统解耦需求明显提高后，再考虑消息队列和更复杂的远程分区框架。

与其寻找某个 Java class 来“实现集群”，不如先把任务领取、幂等计算、故障恢复、结果存储和审计机制设计完整。Java 并发类负责单机内部的并行处理，数据库、消息队列或批处理框架负责多台服务器之间的可靠协作。

备注：内容仅供参考。