ZH
한국어日本語FrançaisEnglishEspañolPortuguêsDeutschTiếng Việt简体中文
首页 / 架构与设计 / 真实案例

一对一和群组聊天系统设计：消息存储与已读状态

Cris The Dev · 2026/10/12 · 11 分钟阅读 · 94

1. 概述

聊天看起来很简单：发一条消息然后显示它。但当有数百万用户时，你必须回答更难的问题。如何保持消息的正确顺序？如何避免重新发送导致消息重复？如何得知谁读到哪条消息而不记录数十亿条“已查看”的记录？

本文介绍一个完整的一对一和群组聊天设计。设计包括需求、数据模型、API、处理流程和难点。读完后，你将能够使用 Spring Boot 3、PostgreSQL 和 Kafka 构建消息存储和已读状态的核心部分。

2. 需求

2.1. 功能需求

  • 一对一和群组聊天（最多1,000个成员）。
  • 发送文本消息。消息在一个对话中按正确的顺序显示。
  • 查看历史记录，按分页向后翻滚。
  • 对话列表（收件箱）及未读消息数。
  • 已读状态：一对一显示“已查看”，群组显示“已被N个人查看”。

2.2. 规模和约束

  • 1,000万DAU。平均每人每天发送40条消息，即每天4亿条消息，约每秒4,600条。高峰时段可能达到每秒30,000条。
  • 一旦确认发送成功，就不能丢失消息。
  • 顺序只需在单个对话内保证正确。不需要全局顺序。
  • 由于网络不稳定，客户端可能重新发送。服务器必须是幂等的。
  • “已读”事件比发送事件更频繁，因为每次打开应用都会触发。

第三个约束是整个设计的关键。只需要本地顺序，所以所有内容都可以按conversation_id分片。

3. 数据模型

核心思想：每个对话有一个单调递增的序列号（seq）。对话中的第N条消息有seq = N。一个成员的已读状态只是一个数字，last_read_seq。它是一个水位线，不是每条已读消息的列表。

CREATE TABLE conversation (
    id              BIGINT PRIMARY KEY,
    type            VARCHAR(10) NOT NULL,          -- DIRECT | GROUP
    last_seq        BIGINT      NOT NULL DEFAULT 0,
    last_message_at TIMESTAMPTZ
);

CREATE TABLE conversation_member (
    conversation_id BIGINT NOT NULL REFERENCES conversation(id),
    user_id         BIGINT NOT NULL,
    last_read_seq   BIGINT NOT NULL DEFAULT 0,
    PRIMARY KEY (conversation_id, user_id)
);
CREATE INDEX idx_member_user ON conversation_member(user_id);

CREATE TABLE message (
    conversation_id BIGINT      NOT NULL,
    seq             BIGINT      NOT NULL,
    sender_id       BIGINT      NOT NULL,
    client_msg_id   UUID        NOT NULL,
    body            TEXT        NOT NULL,
    created_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
    PRIMARY KEY (conversation_id, seq),
    UNIQUE (conversation_id, client_msg_id)
);

CREATE TABLE outbox (
    id         BIGSERIAL PRIMARY KEY,
    msg_key    VARCHAR(64) NOT NULL,
    payload    JSONB       NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

需要注意的几点：

  • 主键(conversation_id, seq)使得读取历史记录只需扫描索引的连续范围。这个键也是自然的分片键。
  • UNIQUE (conversation_id, client_msg_id)是数据库层的去重防护。
  • 未读消息数 = last_seq - last_read_seq。这个计算不需要COUNT(*)。
  • 群组中的“已被查看”通过计算有last_read_seq >= seq的成员数量。你不需要一个有数十亿行的message_receipt表。
  • 对于一对一聊天，添加表direct_pair(user_low, user_high, conversation_id)并设置唯一约束。两个人同时打开聊天时不会创建两个对话。

注意： 水位线假设用户按顺序阅读，即读了消息N就意味着读了所有之前的消息。这个假设对几乎所有聊天应用都成立。如果你需要知道特定消息是否被打开（例如消息自动销毁），为这类消息添加单独的记录。不要对所有消息这样做。

4. API

方法 端点 描述
POST /conversations/{id}/messages 发送消息。Body包含clientMsgId和body
GET /conversations/{id}/messages?beforeSeq=&limit= 历史记录，按seq分页
POST /conversations/{id}/read Body { "seq": 120 }
GET /inbox?limit= 对话列表及未读消息数
WS /ws 服务器推送message.new和read.updated事件

消息通过REST而不是WebSocket发送。这样你可以使用熟悉的HTTP重试、超时和幂等性机制。WebSocket仅用于推送给客户端。

5. 发送消息的流程

需要的依赖：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-jdbc</artifactId>
</dependency>
<dependency>
    <groupId>org.postgresql</groupId>
    <artifactId>postgresql</artifactId>
    <scope>runtime</scope>
</dependency>
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

整个发送消息的操作都在一个事务内：

public record SentMessage(long conversationId, long seq, UUID clientMsgId) {}

@Service
public class MessageService {

    private final JdbcTemplate jdbc;

    public MessageService(JdbcTemplate jdbc) {
        this.jdbc = jdbc;
    }

    @Transactional
    public SentMessage send(long conversationId, long senderId, UUID clientMsgId, String body) {
        // 1. 锁定conversation行：发送到同一对话的每个请求都被排队
        Long lastSeq = jdbc.queryForObject(
            "SELECT last_seq FROM conversation WHERE id = ? FOR UPDATE",
            Long.class, conversationId);

        // 2. 幂等性：如果客户端重新发送，则返回旧结果
        List<Long> existing = jdbc.queryForList(
            "SELECT seq FROM message WHERE conversation_id = ? AND client_msg_id = ?",
            Long.class, conversationId, clientMsgId);
        if (!existing.isEmpty()) {
            return new SentMessage(conversationId, existing.getFirst(), clientMsgId);
        }

        // 3. 分配新的seq，写入消息并更新对话
        long seq = lastSeq + 1;
        jdbc.update("""
            INSERT INTO message(conversation_id, seq, sender_id, client_msg_id, body)
            VALUES (?, ?, ?, ?, ?)""", conversationId, seq, senderId, clientMsgId, body);
        jdbc.update("UPDATE conversation SET last_seq = ?, last_message_at = now() WHERE id = ?",
            seq, conversationId);

        // 4. 发送者显然已经读了自己的消息
        jdbc.update("""
            UPDATE conversation_member SET last_read_seq = GREATEST(last_read_seq, ?)
            WHERE conversation_id = ? AND user_id = ?""", seq, conversationId, senderId);

        // 5. Outbox：事件与数据在同一事务中写入，永远不会不同步
        jdbc.update("INSERT INTO outbox(msg_key, payload) VALUES (?, ?::jsonb)",
            String.valueOf(conversationId),
            """
            {"type":"message.new","conversationId":%d,"seq":%d,"senderId":%d}"""
                .formatted(conversationId, seq, senderId));

        return new SentMessage(conversationId, seq, clientMsgId);
    }
}

每个步骤的解释：

  • 第1步使用SELECT ... FOR UPDATE。两个请求发送到同一对话时会按顺序执行，所以seq永远不会重复或跳跃。不同的对话不会相互阻塞。
  • 第2步在获取锁之后检查重复。因此同一clientMsgId的两个并发重试也不会通过。如果在获取锁前检查，两个请求都可能看到“还没有”。第二个请求会遇到DuplicateKeyException。
  • 第3步使seq连续递增，没有间隙。这样客户端可以自动检测丢失的消息（见第7.3小节）。
  • 第5步应用事务型Outbox。如果你在事务内直接发送到Kafka，可能会出现Kafka已接收但DB回滚，或反之的情况。

用同一clientMsgId调用send两次：

send(1, 10, a1b2..., "大家好") -> SentMessage[conversationId=1, seq=42, ...]
send(1, 10, a1b2..., "大家好") -> SentMessage[conversationId=1, seq=42, ...]   // 不创建新消息

一个单独的中继读取outbox表按id，推送到chat.events主题，key为conversationId，然后删除已发送的行。你也可以使用Debezium来做这个。按对话的key确保同一对话的所有事件进入同一分区并保持顺序。

消费端将事件推送到WebSocket网关：

@Component
public class FanoutListener {

    private final MemberRepository members;   // 返回对话的userId列表
    private final GatewayRegistry gateways;   // userId -> 当前在线的会话

    public FanoutListener(MemberRepository members, GatewayRegistry gateways) {
        this.members = members;
        this.gateways = gateways;
    }

    @KafkaListener(topics = "chat.events", groupId = "fanout")
    public void onEvent(ConsumerRecord<String, String> record) {
        long conversationId = Long.parseLong(record.key());
        for (long userId : members.findUserIds(conversationId)) {
            gateways.pushIfOnline(userId, record.value());
        }
    }
}

离线用户不会收到推送，这是可以接受的。当重新连接时，客户端调用/inbox并自己加载seq大于其保留的最后一个seq的消息。

6. 已读状态

标记为已读只需要一个简单的单调UPDATE：

@Service
public class ReadStateService {

    private final JdbcTemplate jdbc;

    public ReadStateService(JdbcTemplate jdbc) {
        this.jdbc = jdbc;
    }

    public boolean markRead(long conversationId, long userId, long seq) {
        int updated = jdbc.update("""
            UPDATE conversation_member m
               SET last_read_seq = LEAST(?, c.last_seq)
              FROM conversation c
             WHERE c.id = m.conversation_id
               AND m.conversation_id = ? AND m.user_id = ?
               AND m.last_read_seq < LEAST(?, c.last_seq)""",
            seq, conversationId, userId, seq);
        return updated == 1;   // 如果水位线真的向前推进，返回true
    }

    public long unreadCount(long conversationId, long userId) {
        return jdbc.queryForObject("""
            SELECT c.last_seq - m.last_read_seq
              FROM conversation c JOIN conversation_member m ON m.conversation_id = c.id
             WHERE c.id = ? AND m.user_id = ?""", Long.class, conversationId, userId);
    }
}

这个UPDATE同时解决三个问题：

  • 竞争条件： 手机发送seq=50，笔记本稍后发送seq=45。条件last_read_seq < ...阻止旧请求将水位线拉回。你不需要锁或版本号。
  • 垃圾数据： LEAST(?, c.last_seq)阻止客户端发送超过实际消息的seq。
  • 减负： 当updated == 0时，水位线没变所以不需要发送read.updated事件。无效的扇出被削减。

只有当markRead返回true时，服务才会将read.updated事件写入outbox，类似第5小节。

7. 难点和解决方案

7.1. “热”对话和锁竞争

FOR UPDATE使同一对话内的发送消息操作按顺序执行。对于一对一或小型群组，这个成本微不足道。一个1,000人的群组如果很活跃可能每秒有几十条消息进入同一行。每个事务只持有锁几毫秒，所以这个速度仍然可以承受。

如果一个对话超过每秒几百条消息（如直播频道），将seq分配逻辑移到Redis使用INCR chat:seq:{conversationId}。代价是你必须接受DB写入失败时seq可能有间隙。客户端需要区分“真正的间隙”和“消息还没到”。

注意： 不要用created_at或全局ID（Snowflake）来排序对话中的消息。服务器之间的时钟差偏差几毫秒就足以颠倒两条几乎同时发送的消息的顺序。

7.2. DB、Kafka和客户端之间的一致性

这个设计的保证链：

  1. DB是单一事实来源。消息只在提交后才被确认给客户端。
  2. Outbox对Kafka保证至少一次传递。消费者可能收到重复。
  3. 客户端按(conversationId, seq)去重。已有的消息被忽略。

7.3. 检测丢失的消息

seq是连续的，所以客户端可以自动检测间隙。假设客户端有seq=40并收到推送seq=43。它知道缺少消息41和42，所以调用GET /messages?afterSeq=40。你不需要复杂的服务端ACK机制。

7.4. 突发流量

  • 大群组扇出： 一条消息进入1,000人的群组会产生1,000次推送。对于超过阈值的群组（例如200个成员），只推送给在线的人，在Redis集合online:{conversationId}中查询。Payload只包含轻量级通知{conversationId, seq}，客户端需要时自己拉取内容。
  • “已读”风暴： 网络故障后，数百万客户端重新连接并都发送读状态回执。客户端应该防抖1-2秒并只发送最大seq。服务器可以在Redis中进一步合并，只保存key{conversationId}:{userId}的最大值，然后定期批量刷到DB。
  • Kafka： 按conversationId分区来保持顺序。当流量增加时，你只需增加消费者，最多等于分区数。
  • 存储： 每天4亿消息，每条约300字节，相当于每天约120GB。按conversation_id（哈希）对message表进行分区，并将旧消息移到更便宜的存储如Cassandra或对象存储。查询历史总是带着conversation_id，所以任何模型都适合。

8. 测试核心行为

两个最重要的行为需要测试：重新发送不会创建重复消息，水位线永远不会后退。你应该通过Testcontainers用真实的PostgreSQL测试，因为逻辑依赖于FOR UPDATE和GREATEST/LEAST。

<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>
<dependency>
    <groupId>org.testcontainers</groupId>
    <artifactId>junit-jupiter</artifactId>
    <scope>test</scope>
</dependency>

将第3小节的DDL放到src/main/resources/schema.sql并启用spring.sql.init.mode=always。

@SpringBootTest(properties = {
    "spring.sql.init.mode=always",
    "spring.kafka.listener.auto-startup=false"
})
@Testcontainers
class ChatServiceTest {

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

    @Autowired MessageService messages;
    @Autowired ReadStateService reads;
    @Autowired JdbcTemplate jdbc;

    @BeforeEach
    void setUp() {
        jdbc.execute("TRUNCATE message, outbox, conversation_member, conversation");
        jdbc.update("INSERT INTO conversation(id, type) VALUES (1, 'GROUP')");
        jdbc.update("INSERT INTO conversation_member(conversation_id, user_id) VALUES (1, 10), (1, 20)");
    }

    @Test
    void resendingSameClientMessageIdDoesNotCreateDuplicate() {
        UUID clientMsgId = UUID.randomUUID();

        SentMessage first = messages.send(1, 10, clientMsgId, "大家好");
        SentMessage retry = messages.send(1, 10, clientMsgId, "大家好");

        assertThat(retry.seq()).isEqualTo(first.seq()).isEqualTo(1L);
        assertThat(jdbc.queryForObject("SELECT count(*) FROM message", Long.class)).isEqualTo(1L);
    }

    @Test
    void readWatermarkNeverMovesBackward() {
        for (int i = 0; i < 5; i++) {
            messages.send(1, 10, UUID.randomUUID(), "消息 " + i);
        }

        assertThat(reads.markRead(1, 20, 4)).isTrue();
        assertThat(reads.markRead(1, 20, 2)).isFalse();   // 迟到的请求
        assertThat(reads.markRead(1, 20, 99)).isTrue();   // 被限制为last_seq = 5

        assertThat(reads.unreadCount(1, 20)).isZero();
        assertThat(reads.unreadCount(1, 10)).isZero();    // 发送者自动已读
    }
}

第二个测试检查三种情况。水位线前进到4。请求seq=2迟到被跳过，返回false。超过实际seq的被限制为5，所以未读消息数为0。属性spring.kafka.listener.auto-startup=false使测试不会尝试连接Kafka。

9. 结论

整个设计围绕一个决定：每个对话有一个连续的seq。从seq，你获得消息顺序、历史分页、丢失消息检测和只用一个减法的未读消息数。已读状态是单调的水位线，用有条件的UPDATE更新。这消除了竞争条件和庞大的回执表。

其余的是熟悉的技术。clientMsgId加唯一约束用于幂等性。事务型Outbox配合按对话的Kafka key实现一致性和事件顺序。防抖、只推给在线用户和重连后拉取帮助系统应对突发流量。从PostgreSQL和FOR UPDATE开始。只有当真实数据表明你确实需要时，才将seq分配移到Redis或改变存储。

system-designspring-bootpostgresqlkafkacase-study
作者的应用DeclutterClear similar photos, big videos and inbox clutteriOS · Android

同主题文章