在线拍卖系统设计：并发出价和拍卖结束

1. 概述

拍卖系统看起来很简单：用户出价，出价最高者获胜。困难在于两个时刻。首先，数百个出价可能在几秒内涌入同一场拍卖。其次，“拍卖已结束”的边界必须绝对清晰。丢失出价或选错获胜者都是金钱问题。

本文将介绍完整的设计：需求、数据模型、API、如何使用PostgreSQL和Redis处理并发出价、防止“最后一刻出价”（anti-sniping）机制，以及在运行多个实例时如何安全地关闭拍卖。读完后，你可以用Spring Boot 3构建一个正确的拍卖系统核心。

2. 需求分析

2.1. 功能需求

• 卖家创建拍卖，包括起始价、最小增幅和结束时间。
• 买家出价。出价有效当拍卖仍在进行且价格不低于当前价格 + 最小增幅。
• 如果在最后2分钟内有有效出价，拍卖将额外延长2分钟。
• 当时间到了，系统关闭拍卖，确定获胜者并发出事件用于支付和通知步骤。
• 观看者实时接收新价格。

2.2. 规模和约束

• 大约同时进行100,000场拍卖。大多数拍卖都很“冷”，每小时只有几个出价。
• 一些“热门”拍卖（名牌货、限时秒杀）在最后一分钟可以接收到2,000个出价/秒。
• 强一致性在每场拍卖：不丢失有效出价、不接受结束时间后的出价、只有一个获胜者。
• 出价必须是幂等的：因为超时导致的客户端重试不应该创建两个出价。
• 出价延迟的p99应低于200毫秒。

简言之，我们需要保证每场拍卖内的一致性。不同拍卖之间不需要。这是扩展的关键点：每场拍卖是一个独立的争夺单位。

3. 数据模型

我们使用PostgreSQL作为源数据。表auction保存当前状态。表bid是仅追加（append-only）日志。

CREATE TABLE auction (
    id             BIGSERIAL PRIMARY KEY,
    seller_id      BIGINT        NOT NULL,
    title          VARCHAR(200)  NOT NULL,
    current_price  NUMERIC(14,2) NOT NULL,
    min_increment  NUMERIC(14,2) NOT NULL,
    leader_id      BIGINT,
    status         VARCHAR(10)   NOT NULL DEFAULT 'OPEN',  -- OPEN | CLOSED
    end_time       TIMESTAMPTZ   NOT NULL
);
CREATE INDEX idx_auction_open_end ON auction (end_time) WHERE status = 'OPEN';

CREATE TABLE bid (
    id          BIGSERIAL PRIMARY KEY,
    auction_id  BIGINT        NOT NULL REFERENCES auction(id),
    bidder_id   BIGINT        NOT NULL,
    amount      NUMERIC(14,2) NOT NULL,
    request_id  UUID          NOT NULL,
    created_at  TIMESTAMPTZ   NOT NULL DEFAULT clock_timestamp(),
    UNIQUE (auction_id, request_id)
);

需要注意的几点：

• current_price和leader_id被反规范化到auction表。这样，检查出价只需读一行，而不是在bid表上执行SELECT MAX(amount)。
• 部分索引idx_auction_open_end只包含进行中的拍卖。关闭拍卖的任务将很快扫描此索引。
• UNIQUE (auction_id, request_id)是数据库层的幂等性机制。

4. API

POST /auctions                     创建拍卖
GET  /auctions/{id}                查看拍卖状态
POST /auctions/{id}/bids           出价（header Idempotency-Key: <uuid>）
WS   /ws/auctions/{id}             实时接收新价格

出价请求：

{ "bidderId": 42, "amount": 1250000 }

返回结果有三种形式。出价被接受时返回201 Created。价格过低或拍卖已关闭时返回409 Conflict。当请求的Idempotency-Key重复时返回200 OK和之前的结果。

5. 并发出价

5.1. 经典错误：读-改-写

最自然的写法反而是错的：

@Transactional
public void placeBidNaive(long auctionId, long bidderId, BigDecimal amount) {
    Auction a = auctionRepository.findById(auctionId).orElseThrow();
    if (amount.compareTo(a.getCurrentPrice().add(a.getMinIncrement())) < 0) {
        throw new BidRejectedException("价格过低");
    }
    a.setCurrentPrice(amount);
    a.setLeaderId(bidderId);
}

假设当前价格是1000。两个线程都读到价格1000。线程A出价1100，线程B出价1050。两者都通过了检查。如果B最后提交，最终价格是1050，领先者是B。A的更高出价被覆盖了。这是丢失更新错误。

5.2. 原子性条件更新

最简洁的解决方案是将检查和写入合并到一条UPDATE语句中。条件在WHERE中：

public interface AuctionRepository extends JpaRepository<Auction, Long> {

    @Modifying
    @Query(value = """
        UPDATE auction
           SET current_price = :amount,
               leader_id     = :bidderId,
               end_time      = CASE
                   WHEN end_time - clock_timestamp() < INTERVAL '2 minutes'
                   THEN clock_timestamp() + INTERVAL '2 minutes'
                   ELSE end_time END
         WHERE id = :id
           AND status = 'OPEN'
           AND end_time > clock_timestamp()
           AND :amount >= current_price + min_increment
        """, nativeQuery = true)
    int tryPlaceBid(@Param("id") long id,
                    @Param("bidderId") long bidderId,
                    @Param("amount") BigDecimal amount);
}

工作机制如下：

• PostgreSQL在UPDATE时锁定auction行。第二个线程必须等待第一个线程提交。
• 在READ COMMITTED隔离级别下，解锁后，PostgreSQL会在最新的行版本上重新评估WHERE。出价1050遇到新价格1100，条件为假，所以语句返回0。
• 防sniping逻辑（CASE块）就在这条语句中。这样，延期永远不会与引起它的出价相偏离。
• 时间取自数据库的clock_timestamp()，而不是每个应用实例的时钟。数据库是唯一的时间源。

注意：不要在这里使用now()。在PostgreSQL中，now()返回事务开始的时间。一个在关闭时间之前打开但运行缓慢的事务仍然可能“漏掉”一个在关闭后的出价。

Service在同一事务中记录出价日志：

@Service
@RequiredArgsConstructor
public class BidService {

    private final AuctionRepository auctionRepository;
    private final BidRepository bidRepository;

    @Transactional
    public Bid placeBid(long auctionId, long bidderId, BigDecimal amount, UUID requestId) {
        var existing = bidRepository.findByAuctionIdAndRequestId(auctionId, requestId);
        if (existing.isPresent()) {
            return existing.get(); // 客户端重试：返回之前的结果
        }
        int updated = auctionRepository.tryPlaceBid(auctionId, bidderId, amount);
        if (updated == 0) {
            throw new BidRejectedException("拍卖已关闭或价格不足");
        }
        return bidRepository.save(new Bid(auctionId, bidderId, amount, requestId));
    }
}

如果两个具有相同requestId的重试并行运行，两者都可能通过findBy...步骤。此时，UNIQUE约束会对第二个事务抛出DataIntegrityViolationException。该事务会完全回滚，包括价格UPDATE部分。Controller捕获此异常并返回已存在的出价。

5.3. 并发出价测试

下面的测试使用虚拟线程将200个出价并行发送到同一场拍卖，然后检查不变量。需要添加的依赖项：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-testcontainers</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.testcontainers</groupId>
    <artifactId>postgresql</artifactId>
    <scope>test</scope>
</dependency>

@SpringBootTest
@Testcontainers
class BidServiceConcurrencyTest {

    @Container @ServiceConnection
    static PostgreSQLContainer<?> postgres = new PostgreSQLContainer<>("postgres:16");

    @Autowired BidService bidService;
    @Autowired AuctionRepository auctionRepository;
    @Autowired BidRepository bidRepository;

    @Test
    void concurrentBids_keepHighestAcceptedBidAsLeader() throws Exception {
        long id = auctionRepository.save(Auction.open("古董手表",
                new BigDecimal("1000"), new BigDecimal("10"),
                Instant.now().plus(Duration.ofHours(1)))).getId();

        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            for (int i = 1; i <= 200; i++) {
                long bidder = i;
                var amount = new BigDecimal(1000 + i * 10);
                executor.submit(() -> {
                    try {
                        bidService.placeBid(id, bidder, amount, UUID.randomUUID());
                    } catch (BidRejectedException ignored) { }
                });
            }
        }

        Auction result = auctionRepository.findById(id).orElseThrow();
        BigDecimal maxAccepted = bidRepository.findMaxAmount(id);

        assertThat(result.getCurrentPrice()).isEqualByComparingTo("3000");
        assertThat(result.getLeaderId()).isEqualTo(200L);
        assertThat(maxAccepted).isEqualByComparingTo(result.getCurrentPrice());
    }
}

出价3000（出价者200）总是有效的，因为没有更高的出价。因此，无论执行顺序如何，它必须是领先者。如果将tryPlaceBid替换为第5.1节的naive版本，此测试将间歇性失败。这正是竞态条件的标志。

6. “热门”拍卖：将争夺移到Redis

条件更新在一行上序列化所有出价。对于一场拍卖，PostgreSQL通常每秒可以处理几百到大约一千个UPDATE。每秒收到2000个出价的拍卖会产生长的锁队列，推高p99。

对于标记为“热门”的拍卖，我们将决定逻辑转移到Redis。Redis以单线程和原子方式运行Lua脚本。需要添加的依赖项：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

@Service
@RequiredArgsConstructor
public class HotBidService {

    private static final RedisScript<Long> PLACE_BID = RedisScript.of("""
        local a = redis.call('HMGET', KEYS[1], 'price', 'inc', 'end', 'status')
        local now = tonumber(ARGV[3])
        if a[4] ~= 'OPEN' or now >= tonumber(a[3]) then return -1 end
        if tonumber(ARGV[1]) < tonumber(a[1]) + tonumber(a[2]) then return 0 end
        local newEnd = tonumber(a[3])
        if newEnd - now < 120000 then newEnd = now + 120000 end
        redis.call('HSET', KEYS[1], 'price', ARGV[1], 'leader', ARGV[2], 'end', newEnd)
        return 1
        """, Long.class);

    private final StringRedisTemplate redis;
    private final KafkaTemplate<String, BidAccepted> kafka;

    public boolean placeBid(long auctionId, long bidderId, long amount, UUID requestId) {
        long now = Long.parseLong(redis.execute(
                (RedisCallback<Long>) c -> c.serverCommands().time()).toString());
        Long r = redis.execute(PLACE_BID, List.of("auction:" + auctionId),
                String.valueOf(amount), String.valueOf(bidderId), String.valueOf(now));
        if (r == null || r != 1L) return false;
        kafka.send("bid-accepted", String.valueOf(auctionId),
                new BidAccepted(auctionId, bidderId, amount, requestId, now));
        return true;
    }
}

简单解释：

• Script读取价格、增幅、结束时间和状态，检查后写入，一切都在一个原子步骤中。返回值：1表示接受，0表示价格过低，-1表示拍卖已关闭。
• 时间取自Redis的TIME命令，以便所有实例共享一个时钟。实际上，你可以在脚本中直接调用redis.call('TIME')。为此，需要启用脚本效果复制模式，这从Redis 5开始是默认的。
• 被接受的出价将被发送到Kafka，key是auctionId。一场拍卖的所有出价都落在同一分区并保持顺序。Consumer以批处理方式将它们写入PostgreSQL。

注意：此时Redis成为整个拍卖期间的源真相。你需要启用AOF，配置appendfsync everysec或always，运行副本，并处理Redis写入后kafka.send失败的情况（重试或定期对账Redis和PostgreSQL）。不要对所有拍卖应用此模型。大多数冷拍卖使用条件更新运行良好，并且简单得多。

7. 拍卖结束

7.1. 用原子语句关闭拍卖

我们不需要为每场拍卖精确到毫秒的调度器。真正的边界已经在出价UPDATE语句的条件end_time > clock_timestamp()中。关闭后，没有出价能通过，即使状态仍然是OPEN。拍卖关闭任务只需在几秒内“结账”：

@Component
@RequiredArgsConstructor
public class AuctionCloser {

    private final JdbcTemplate jdbc;

    @Scheduled(fixedDelay = 1000)
    @Transactional
    public void closeExpired() {
        var closed = jdbc.query("""
            UPDATE auction SET status = 'CLOSED'
             WHERE id IN (SELECT id FROM auction
                           WHERE status = 'OPEN' AND end_time <= clock_timestamp()
                           ORDER BY end_time LIMIT 500
                           FOR UPDATE SKIP LOCKED)
            RETURNING id, leader_id, current_price
            """, (rs, i) -> new AuctionClosed(rs.getLong("id"),
                    (Long) rs.getObject("leader_id"), rs.getBigDecimal("current_price")));

        for (var e : closed) {
            jdbc.update("INSERT INTO outbox(topic, aggregate_id, payload) VALUES (?, ?, ?::jsonb)",
                    "auction-closed", e.auctionId(), e.toJson());
        }
    }
}

该语句在运行多个实例时是安全的，因为：

• FOR UPDATE SKIP LOCKED允许实例之间共享拍卖。两个实例永远不会关闭同一场拍卖。
• 如果一个出价持有行锁，SKIP LOCKED会跳过该行。下次运行时会重新检查，此时end_time可能已被延长。
• 状态转换OPEN → CLOSED只发生一次。重新运行任务不会产生副作用，即任务是幂等的。

7.2. 使用Outbox模式发出事件

auction-closed事件触发支付和向获胜者发送邮件。如果在事务中直接调用kafka.send，你可能会遇到两种糟糕的情况。一是已发送事件但事务回滚。二是已提交但发送失败。表outbox与关闭拍卖的操作在同一事务中写入。然后，relay（轮询或Debezium CDC）将outbox行推送到Kafka。Consumer必须按auctionId是幂等的，因为relay保证至少一次。

8. 流量突增和实时fan-out

热门拍卖的最后一分钟不仅有很多出价，还有几十万观看者。一些辅助技术：

• 按用户限速，例如使用Redis中的令牌桶限制最多5个出价/秒。这种技术在spam bot触及争夺层之前就阻止了它们。
• 早期拒绝：在内存中缓存当前价格（几百毫秒），在调用DB或Redis之前立即拒绝明显过低的出价。缓存仅用于拒绝，永远不用于接受。
• 通过pub/sub的fan-out：出价被接受时，发布到Redis频道auction:{id}。每个WebSocket节点订阅并推送给其客户端。如果出价太频繁，每100–200毫秒合并一次更新。观看者只需看到最新价格，无需看到每个出价。
• 分离读写：GET /auctions/{id}可以从副本或缓存读取。只有出价路径需要访问主库。

9. 常见错误和何时不应使用

• 使用应用时钟与end_time比较。实例之间相差几秒是正常的。始终让数据库或Redis决定时间。
• **对热门拍卖使用@Version（乐观锁）。**这仍然是正确的，但2000个出价/秒会产生无数OptimisticLockException和重试循环。条件更新不需要重试，因为WHERE会自动重新评估。
• **使用SELECT ... FOR UPDATE然后在Java中检查。**这是正确的，但保持锁的时间更长，因为需要等待应用程序的往返。只有当检查逻辑太复杂而无法用SQL编写时才使用。
• **围绕整个出价函数的分布式锁（Redisson）。**这增加了单点故障和锁过期风险，而数据库已经有行锁。
• **忽略幂等性。**移动客户端在网络不稳定时重试是常态。没有request_id，同一个人可能“对自己出价”两次。

10. 结论

拍卖系统的关键是将每场拍卖视为一个独立的一致性单位。本文已处理每个难点如下：

• 并发出价：将检查和写入合并到一条原子UPDATE ... WHERE语句中，在同一语句中包含防sniping逻辑。
• 结束时间：从数据库获取时间（clock_timestamp()），而不是从应用程序。
• 热门拍卖：将争夺转移到Redis Lua并通过Kafka持久化，代价是运维复杂性增加。
• 关闭拍卖：使用FOR UPDATE SKIP LOCKED，幂等状态转换，并通过Outbox模式发出事件。

你应该从纯PostgreSQL版本开始。它对大多数情况都足够了，并且易于通过5.3节中的并发测试验证。只有当实际测量显示auction行成为瓶颈时，才将这些拍卖转移到Redis。