Online payment system 与 Confluent Kafka-Go API 架构问题

2026-10-24支付小周

结论

这种架构有风险。问题不在 ReadMessage 本身，而在消费起点、offset 的提交时机，以及支付操作能否保证幂等。

ReadMessage() 读取的是 consumer 当前 position 之后的下一条消息，并非每次只取 topic 中最新的一条。首次从哪里开始消费，取决于：

consumer group 是否已有 committed offset；
auto.offset.reset 的配置；
offset 是否在支付处理完成前被自动或提前提交。

支付请求不能在收到消息时就视为处理完成。应当在支付成功并持久化结果后，再提交对应的 offset。由于消息可能重复投递，支付操作还必须支持幂等。

只把 auto.offset.reset 从 latest 改成 earliest，解决不了所有问题。

`latest` 到底意味着什么

对于同一个 consumer group：

Kafka 中有 committed offset 时，consumer 会从该位置继续读取；
只有找不到有效的 committed offset 时，auto.offset.reset 才会生效；
latest 表示从当时的分区末尾开始消费。该 consumer group 不会读取此前已有的消息；
earliest 表示从当前仍在保留期内的最早消息开始读取。

因此，新建 consumer group 时使用 latest，可能跳过服务启动前已经进入 topic 的支付请求。需要处理历史积压时，支付消费者通常更适合使用 earliest。

不过，如果请求已经被读取，offset 也已经提交，那么即使支付尚未完成，重启后也不会因为改成 earliest 而重新处理。Kafka 会优先采用该 consumer group 已提交的位置。

最大的风险：异步处理与 offset 提交脱节

假设程序按下面的流程执行：

读取 offset 100 的请求；
将请求交给 goroutine；
consumer 自动提交 offset；
goroutine 尚未完成支付；
服务崩溃。

服务重启后，consumer 会从已提交的位置继续消费，offset 100 对应的请求可能永久遗漏。

异步任务的完成顺序也不确定。例如，在同一分区内：

offset 100 仍在处理；
offset 101 已经处理完成；
程序提交了 offset 101。

Kafka 的提交语义表示此前的消息均已完成。如果服务此时崩溃，offset 100 也可能被跳过。

因此，不能直接让多个 goroutine 处理消息，再由各个 goroutine 独立提交自己的 offset。

最简单可靠的处理方式

支付业务通常应优先保证正确性。可以先让每个 consumer 顺序处理消息，处理成功后再提交 offset：

func consumePayments(
    consumer *kafka.Consumer,
    processor PaymentProcessor,
) error {
    if err := consumer.SubscribeTopics([]string{"payment-requests"}, nil); err != nil {
        return err
    }

    for {
        msg, err := consumer.ReadMessage(-1)
        if err != nil {
            var kafkaErr kafka.Error
            if errors.As(err, &kafkaErr) && kafkaErr.Code() == kafka.ErrAllBrokersDown {
                return err
            }

            log.Printf("failed to read Kafka message: %v", err)
            continue
        }

        request, err := decodePaymentRequest(msg.Value)
        if err != nil {
            // 根据业务策略写入隔离队列或人工处理队列。
            // 不应在没有记录错误请求的情况下直接提交 offset。
            log.Printf(
                "invalid payment request topic=%s partition=%d offset=%v: %v",
                *msg.TopicPartition.Topic,
                msg.TopicPartition.Partition,
                msg.TopicPartition.Offset,
                err,
            )
            continue
        }

        if err := processor.Process(request); err != nil {
            log.Printf(
                "payment processing failed partition=%d offset=%v: %v",
                msg.TopicPartition.Partition,
                msg.TopicPartition.Offset,
                err,
            )

            // 不提交 offset，使消息在重启或重新分配后能够再次消费。
            // 实际系统还应加入有限重试、退避和人工介入机制。
            continue
        }

        if _, err := consumer.CommitMessage(msg); err != nil {
            log.Printf(
                "failed to commit partition=%d offset=%v: %v",
                msg.TopicPartition.Partition,
                msg.TopicPartition.Offset,
                err,
            )
            // 支付可能已经成功，但 offset 提交失败。
            // 消息随后可能被重复处理，因此 Process 必须支持幂等。
            continue
        }
    }
}

这段示例只用于说明处理顺序，并非完整的生产代码。实际项目还要处理退出信号、consumer group rebalance、超时、日志脱敏、监控和重试上限。

支付操作必须幂等

处理成功后再提交 offset，可以提供接近 at-least-once 的消费语义。请求通常不会丢失，但可能被重复处理。

典型的故障过程如下：

支付渠道已经扣款成功；
服务尚未提交 Kafka offset；
服务崩溃；
消息被再次消费；
系统再次发起扣款。

因此，每个支付请求都要有稳定且全局唯一的业务标识，例如 payment_request_id。重复消费时必须继续使用同一个标识，不能重新生成。

可以在数据库中为该标识建立唯一约束：

CREATE TABLE payment_requests (
    payment_request_id VARCHAR(128) PRIMARY KEY,
    status VARCHAR(32) NOT NULL,
    provider_reference VARCHAR(128),
    updated_at TIMESTAMP NOT NULL
);

处理请求时，应先查询状态或通过原子操作写入状态：

type PaymentRequest struct {
    ID     string
    Amount int64
    UserID string
}

func (p *Service) Process(req PaymentRequest) error {
    payment, err := p.repository.FindByRequestID(req.ID)
    if err != nil && !errors.Is(err, ErrNotFound) {
        return err
    }

    if payment != nil && payment.Status == "SUCCEEDED" {
        return nil
    }

    result, err := p.provider.Charge(
        context.Background(),
        req.ID, // 作为支付渠道的 idempotency key
        req.Amount,
    )
    if err != nil {
        return err
    }

    return p.repository.MarkSucceeded(req.ID, result.Reference)
}

如果支付提供方支持 idempotency key，应把同一个业务请求 ID 传给它。只依靠本地数据库去重，仍然无法消除“外部扣款成功，但本地状态尚未保存”这个故障窗口。

如果支付渠道既不支持幂等请求，也不支持状态查询，就需要设计更严格的交易状态机和对账补偿流程。Kafka offset 无法解决这类问题。

需要并发时如何处理

可以继续使用 goroutine 和 channel，但必须保证分区内的 offset 按顺序提交。常见做法包括：

每个 Kafka partition 交给一个 worker 顺序处理；
不同 partition 之间并行处理；
同一 partition 并行处理时，记录每个 offset 的完成状态；
只有此前的所有 offset 都已完成，才能推进该分区的提交位置；
rebalance 或关闭 consumer 前，停止接收新任务，并等待或取消正在执行的任务。

如果支付吞吐量允许，按分区顺序处理更容易保证正确性。增加 topic partition 数量，可以在维持分区内顺序的同时提高整体并行度。

消息 key 也要合理选择。比如，需要保证同一订单的支付请求严格有序时，可以用订单 ID 作为 Kafka message key，让同一订单的消息进入同一个 partition。

失败请求不能无限阻塞

不提交失败消息可以防止请求遗漏，但一条永久失败的消息可能反复重试，阻塞后续请求。需要按错误类型分别处理：

临时错误：网络超时、服务限流，可以使用指数退避重试；
明确失败：余额不足、订单已取消，应记录最终业务状态；
数据错误：字段缺失、格式错误，可以写入隔离 topic；
状态不确定：支付请求超时且无法确认是否扣款，应进入查询、对账或人工处理流程。

先把消息可靠地写入 dead-letter topic，再提交原 offset，是一种可行的处理方式。但隔离消息必须写入成功，还要配套监控和后续处理流程，否则只是把遗漏问题转移到了另一个位置。

建议的完整流程

较稳妥的支付消费链路可以按以下方式设计：

使用固定的 consumer group；
新消费组根据业务需求设置 auto.offset.reset=earliest；
关闭自动提交；
校验消息，取得稳定的支付请求 ID；
通过数据库唯一约束和支付渠道 idempotency key 防止重复扣款；
持久化支付结果或明确的待确认状态；
完成上述步骤后提交 Kafka offset；
对失败、超时和未知状态执行重试、查询和对账；
监控 consumer lag、处理时延、失败次数及长期未完成的支付请求。

目标不应是让每条消息只处理一次，因为 Kafka 消费、数据库事务和外部支付调用通常无法组成一个简单的原子事务。更可行的目标是允许消息重复投递，同时保证重复处理不会造成重复扣款，并且每个未完成的请求都能够被追踪和恢复。