社交架构设计 / 即时通讯系统

1. 社交产品架构

社交产品的核心由用户关系、用户资料、内容发布、互动四个子系统构成，各子系统之间通过事件驱动和异步消息进行解耦协作。

text

                    ┌─────────────────────────────────────────────┐
                    │                社交产品架构                    │
                    ├──────────────┬──────────────┬───────────────┤
                    │   用户关系     │   用户资料     │   内容发布      │
                    │  ┌────────┐  │  ┌────────┐  │  ┌─────────┐ │
                    │  │ 好友管理 │  │  │ 资料设置 │  │  │ 动态发布  │ │
                    │  │ 关注/粉丝│  │  │ 用户标签 │  │  │ 图文/视频 │ │
                    │  │ 黑名单   │  │  │ 社交积分 │  │  │ 位置服务 │ │
                    │  └────────┘  │  └────────┘  │  └─────────┘ │
                    ├──────────────┴──────────────┴───────────────┤
                    │                  互动系统                      │
                    │   ┌────────┐ ┌────────┐ ┌────────┐ ┌──────┐ │
                    │   │  点赞   │ │  评论   │ │  转发   │ │ 收藏  │ │
                    │   └────────┘ └────────┘ └────────┘ └──────┘ │
                    └─────────────────────────────────────────────┘

1.1 用户关系

用户关系是社交产品的基石，主要包括好友关系、关注/粉丝关系和黑名单管理。

关系类型定义：

关系类型	方向性	说明	典型场景
好友	双向	双方互相关注，需对方确认	微信、QQ
关注	单向	A关注B，B不一定要关注A	微博、抖音
粉丝	单向	关注的反向关系	微博、抖音
黑名单	单向	屏蔽对方的消息和互动	所有社交产品

好友关系核心接口：

java

public interface FriendRelationService {

    /** 发送好友请求 */
    Result<FriendApplyVO> sendFriendApply(Long userId, Long targetUserId, String remark);

    /** 同意好友请求 */
    Result<Void> acceptFriendApply(Long userId, Long applyId);

    /** 拒绝好友请求 */
    Result<Void> rejectFriendApply(Long userId, Long applyId);

    /** 删除好友 */
    Result<Void> removeFriend(Long userId, Long friendId);

    /** 拉黑用户 */
    Result<Void> blockUser(Long userId, Long targetUserId, BlockType type);

    /** 查询好友列表 */
    Result<List<FriendVO>> listFriends(Long userId, FriendQuery query);

    /** 查询共同好友 */
    Result<List<Long>> listMutualFriends(Long userId, Long targetUserId);
}

1.2 用户资料

用户资料系统管理用户在社交平台上的身份信息、个性化设置、标签体系和社交积分。

资料模型：

java

@Entity
@Table(name = "user_profile")
public class UserProfile {

    @Id
    private Long userId;

    // ========== 基本信息 ==========
    @Column(length = 64)
    private String nickname;                    // 昵称

    @Column(length = 256)
    private String avatar;                      // 头像URL

    @Column(length = 16)
    private String gender;                      // 性别

    @Column(length = 1024)
    private String bio;                         // 个人简介

    // ========== 扩展信息 ==========
    @Column(columnDefinition = "TEXT")
    private String tags;                        // 用户标签，JSON数组存储

    @Column(length = 64)
    private String region;                      // 地区

    @Column(length = 256)
    private String coverImage;                  // 个人主页封面图

    // ========== 社交积分 ==========
    @Column(nullable = false)
    private Integer socialScore;                // 社交积分

    @Column(length = 32)
    private String level;                       // 用户等级

    // ========== 时间字段 ==========
    private LocalDateTime createTime;
    private LocalDateTime updateTime;
}

社交积分规则：

行为	积分变化	每日上限
发布动态	+10	50
获得点赞	+2	100
获得评论	+3	60
被关注	+5	200
每日签到	+1	1
违规行为	-20 ~ -100	无上限

java

@Component
public class SocialScoreService {

    @Autowired
    private StringRedisTemplate redisTemplate;

    private static final String SCORE_KEY_PREFIX = "social:score:";

    /** 增加社交积分（含每日上限检查） */
    public Result<Void> addScore(Long userId, int score, ScoreAction action) {
        String dailyKey = SCORE_KEY_PREFIX + userId + ":daily:" + LocalDate.now();
        Integer todayScore = redisTemplate.opsForValue().get(dailyKey, Integer.class);
        if (todayScore != null && todayScore >= action.getDailyLimit()) {
            return Result.fail("今日积分已达上限");
        }

        // 更新当日累计
        redisTemplate.opsForValue().increment(dailyKey, score);
        // 更新总积分
        redisTemplate.opsForValue().increment(SCORE_KEY_PREFIX + userId, score);

        // 异步持久化到DB
        asyncPersistScore(userId, score);
        return Result.success();
    }
}

1.3 内容发布

内容发布系统支持多种内容类型的发布、审核和分发。

内容类型：

类型	存储格式	说明
文字动态	TEXT	纯文本内容
图文动态	TEXT + JSON(图片列表)	文字配多张图片
视频动态	TEXT + JSON(视频信息)	短视频内容
位置动态	TEXT + 经纬度	带地理位置的内容

内容发布模型：

java

@Entity
@Table(name = "social_post")
public class SocialPost {

    @Id
    private Long postId;

    @Column(nullable = false)
    private Long userId;                        // 发布者ID

    @Column(columnDefinition = "TEXT", nullable = false)
    private String content;                     // 文字内容

    @Column(columnDefinition = "JSON")
    private String mediaList;                   // 媒体列表，JSON数组

    @Column(length = 32)
    private String postType;                    // 内容类型：TEXT/IMAGE/VIDEO/LOCATION

    // ========== 位置信息 ==========
    @Column(length = 128)
    private String locationName;                // 位置名称

    @Column
    private Double latitude;                    // 纬度

    @Column
    private Double longitude;                   // 经度

    // ========== 状态字段 ==========
    @Column(length = 16, nullable = false)
    private String status;                      // 状态：PENDING/APPROVED/REJECTED/DELETED

    @Column(nullable = false)
    private Integer visibleRange;               // 可见范围：0=公开 1=好友 2=私密

    // ========== 计数统计 ==========
    @Column(nullable = false)
    private Integer likeCount = 0;

    @Column(nullable = false)
    private Integer commentCount = 0;

    @Column(nullable = false)
    private Integer shareCount = 0;

    @Column(nullable = false)
    private Integer favoriteCount = 0;

    private LocalDateTime createTime;
    private LocalDateTime updateTime;
}

1.4 互动系统

互动系统包括点赞、评论、转发和收藏四个核心功能，采用异步架构保证写入性能。

互动数据模型：

java

@Entity
@Table(name = "social_interaction",
       uniqueConstraints = @UniqueConstraint(columnNames = {"post_id", "user_id", "type"}))
public class SocialInteraction {

    @Id
    private Long id;

    @Column(nullable = false)
    private Long postId;                        // 目标动态ID

    @Column(nullable = false)
    private Long userId;                        // 操作用户ID

    @Column(length = 16, nullable = false)
    private String type;                        // 互动类型：LIKE/COMMENT/SHARE/FAVORITE

    @Column(columnDefinition = "TEXT")
    private String content;                     // 评论内容（仅评论类型有效）

    @Column
    private Long parentId;                      // 父评论ID，支持嵌套回复

    @Column
    private Long rootId;                        // 根评论ID

    private LocalDateTime createTime;
}

点赞接口（使用 Redis 做计数缓冲）：

java

@Service
public class InteractionService {

    @Autowired
    private StringRedisTemplate redisTemplate;

    private static final String LIKE_COUNT_KEY = "post:like:count:";

    /** 点赞操作，先写Redis后异步刷DB */
    public Result<Void> like(Long userId, Long postId) {
        String likeKey = "post:like:set:" + postId;
        Boolean added = redisTemplate.opsForSet().add(likeKey, String.valueOf(userId));
        if (Boolean.TRUE.equals(added)) {
            redisTemplate.opsForValue().increment(LIKE_COUNT_KEY + postId);
            // 发送异步消息持久化
            sendInteractionMessage(userId, postId, "LIKE");
        }
        return Result.success();
    }

    /** 取消点赞 */
    public Result<Void> unlike(Long userId, Long postId) {
        String likeKey = "post:like:set:" + postId;
        Boolean removed = redisTemplate.opsForSet().remove(likeKey, String.valueOf(userId));
        if (Boolean.TRUE.equals(removed)) {
            redisTemplate.opsForValue().decrement(LIKE_COUNT_KEY + postId);
            sendInteractionMessage(userId, postId, "UNLIKE");
        }
        return Result.success();
    }
}

2. 关系链设计

2.1 好友关系表

好友关系采用 MySQL 关系表存储，每条记录表示一条经过确认的双向好友关系。

sql

CREATE TABLE `friend_relation` (
    `id`            BIGINT       NOT NULL AUTO_INCREMENT COMMENT '主键',
    `user_id`       BIGINT       NOT NULL COMMENT '用户ID',
    `friend_id`     BIGINT       NOT NULL COMMENT '好友ID',
    `status`        TINYINT      NOT NULL DEFAULT 0 COMMENT '状态：0=待确认 1=已确认 2=已删除',
    `group_id`      BIGINT       DEFAULT NULL COMMENT '好友分组ID',
    `remark`        VARCHAR(64)  DEFAULT NULL COMMENT '好友备注',
    `source`        VARCHAR(32)  DEFAULT NULL COMMENT '来源：search/qrcode/contact/group',
    `create_time`   DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
    `update_time`   DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
    PRIMARY KEY (`id`),
    UNIQUE KEY `uk_user_friend` (`user_id`, `friend_id`),
    KEY `idx_user_id_status` (`user_id`, `status`),
    KEY `idx_friend_id` (`friend_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='好友关系表';

好友申请表：

sql

CREATE TABLE `friend_apply` (
    `id`            BIGINT       NOT NULL AUTO_INCREMENT COMMENT '主键',
    `from_user_id`  BIGINT       NOT NULL COMMENT '申请人ID',
    `to_user_id`    BIGINT       NOT NULL COMMENT '被申请人ID',
    `remark`        VARCHAR(256) DEFAULT NULL COMMENT '申请附言',
    `status`        TINYINT      NOT NULL DEFAULT 0 COMMENT '状态：0=待处理 1=已同意 2=已拒绝 3=已过期',
    `create_time`   DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '申请时间',
    `handle_time`   DATETIME     DEFAULT NULL COMMENT '处理时间',
    PRIMARY KEY (`id`),
    KEY `idx_to_user_status` (`to_user_id`, `status`),
    KEY `idx_from_user` (`from_user_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='好友申请表';

好友分组表：

sql

CREATE TABLE `friend_group` (
    `id`            BIGINT       NOT NULL AUTO_INCREMENT COMMENT '主键',
    `user_id`       BIGINT       NOT NULL COMMENT '用户ID',
    `group_name`    VARCHAR(32)  NOT NULL COMMENT '分组名称',
    `sort_order`    INT          NOT NULL DEFAULT 0 COMMENT '排序',
    `create_time`   DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP,
    PRIMARY KEY (`id`),
    KEY `idx_user_id` (`user_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='好友分组表';

2.2 关注/粉丝表

关注/粉丝关系适合三种存储方案，不同方案在性能、开发成本和查询灵活性上各有优劣。

方案一：JSON 数组存储（适合小规模）

直接在用户表中用 JSON 字段存储关注列表和粉丝列表。

sql

CREATE TABLE `user_relation_json` (
    `user_id`         BIGINT       NOT NULL PRIMARY KEY COMMENT '用户ID',
    `follow_ids`      JSON         NOT NULL COMMENT '关注用户ID列表，如 [1001,1002,1003]',
    `follower_ids`    JSON         NOT NULL COMMENT '粉丝用户ID列表',
    `follow_count`    INT          NOT NULL DEFAULT 0 COMMENT '关注数',
    `follower_count`  INT          NOT NULL DEFAULT 0 COMMENT '粉丝数',
    `create_time`     DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP,
    `update_time`     DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='用户关系统计表(JSON方案)';

优点： 实现简单，单行读取所有关系，适合用户量较少（万级以下）的场景。 缺点： JSON 字段更新需要全量替换，无法支持分页查询，不适合大规模数据。

方案二：Redis Set 存储（适合高并发）

使用 Redis 的 Set 数据结构存储关注/粉丝关系。

java

@Component
public class FollowRedisService {

    @Autowired
    private StringRedisTemplate redisTemplate;

    private static final String FOLLOW_KEY = "social:follow:";
    private static final String FAN_KEY = "social:fan:";

    /** 关注用户 */
    public void follow(Long userId, Long targetUserId) {
        // 使用lua脚本保证原子性
        String lua = "redis.call('SADD', KEYS[1], ARGV[1]); redis.call('SADD', KEYS[2], ARGV[2]); return 1;";
        redisTemplate.execute(
            new DefaultRedisScript<>(lua, Long.class),
            Arrays.asList(FOLLOW_KEY + userId, FAN_KEY + targetUserId),
            String.valueOf(targetUserId), String.valueOf(userId)
        );
    }

    /** 取消关注 */
    public void unfollow(Long userId, Long targetUserId) {
        String lua = "redis.call('SREM', KEYS[1], ARGV[1]); redis.call('SREM', KEYS[2], ARGV[2]); return 1;";
        redisTemplate.execute(
            new DefaultRedisScript<>(lua, Long.class),
            Arrays.asList(FOLLOW_KEY + userId, FAN_KEY + targetUserId),
            String.valueOf(targetUserId), String.valueOf(userId)
        );
    }

    /** 是否已关注 */
    public boolean isFollowing(Long userId, Long targetUserId) {
        return Boolean.TRUE.equals(
            redisTemplate.opsForSet().isMember(FOLLOW_KEY + userId, String.valueOf(targetUserId))
        );
    }

    /** 分页查询关注列表 */
    public List<Long> listFollows(Long userId, int page, int size) {
        int start = (page - 1) * size;
        int end = start + size - 1;
        Set<String> members = redisTemplate.opsForSet().members(FOLLOW_KEY + userId);
        // 实际使用 SCAN 或 SRANDMEMBER 做分页展示
        return members.stream().map(Long::valueOf).collect(Collectors.toList());
    }

    /** 查询共同关注 */
    public List<Long> intersectFollows(Long userId1, Long userId2) {
        Set<String> intersect = redisTemplate.opsForSet().intersect(
            FOLLOW_KEY + userId1, FOLLOW_KEY + userId2
        );
        return intersect.stream().map(Long::valueOf).collect(Collectors.toList());
    }
}

优点： 读写性能极高（微秒级），天然支持集合运算（交集/并集/差集），适合高并发场景。 缺点： 数据可靠性依赖 Redis 持久化配置，内存成本较高，复杂的业务查询能力有限。

方案三：MySQL 关系表存储（适合大规模、复杂查询）

sql

CREATE TABLE `user_follow` (
    `id`              BIGINT       NOT NULL AUTO_INCREMENT COMMENT '主键',
    `user_id`         BIGINT       NOT NULL COMMENT '关注者ID',
    `target_user_id`  BIGINT       NOT NULL COMMENT '被关注者ID',
    `source`          VARCHAR(32)  DEFAULT NULL COMMENT '关注来源',
    `create_time`     DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP,
    PRIMARY KEY (`id`),
    UNIQUE KEY `uk_user_target` (`user_id`, `target_user_id`),
    KEY `idx_user_id_time` (`user_id`, `create_time`),
    KEY `idx_target_user_id_time` (`target_user_id`, `create_time`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='关注关系表';

CREATE TABLE `user_follower` (
    `id`              BIGINT       NOT NULL AUTO_INCREMENT COMMENT '主键',
    `user_id`         BIGINT       NOT NULL COMMENT '用户ID（被关注者）',
    `follower_id`     BIGINT       NOT NULL COMMENT '粉丝ID',
    `create_time`     DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP,
    PRIMARY KEY (`id`),
    UNIQUE KEY `uk_user_follower` (`user_id`, `follower_id`),
    KEY `idx_user_id_time` (`user_id`, `create_time`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='粉丝关系表';

优点： 数据可靠，支持复杂查询（按时间排序、分页、聚合统计），存储成本低。 缺点： 写入性能低于 Redis，大 V 用户的粉丝表可能出现热点行竞争。

三种方案对比总结

维度	JSON 数组	Redis Set	MySQL 关系表
数据规模	万级以下	百万~千万级	亿级
读取性能	一般（全量读取）	极高（微秒级）	较好（毫秒级）
写入性能	差（全量覆盖）	极高	一般
集合运算	不支持	原生支持	SQL JOIN 实现
数据可靠性	低	中（依赖RDB/AOF）	高
存储成本	低	高（内存）	低
分页查询	不支持	有限支持	完善支持
推荐场景	原型验证	热数据缓存	核心关系存储

实际项目推荐方案： 采用 Redis Set + MySQL 关系表 的组合模式。Redis 负责高并发读写（关注/取关/判断），MySQL 负责数据持久化和复杂查询，通过异步消息保持最终一致性。

java

@Service
public class FollowService {

    @Autowired
    private FollowRedisService followRedisService;

    @Autowired
    private UserFollowMapper followMapper;

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Transactional
    public Result<Void> follow(Long userId, Long targetUserId) {
        // 1. 写入DB
        UserFollow follow = new UserFollow();
        follow.setUserId(userId);
        follow.setTargetUserId(targetUserId);
        followMapper.insert(follow);

        UserFollower follower = new UserFollower();
        follower.setUserId(targetUserId);
        follower.setFollowerId(userId);
        followerMapper.insert(follower);

        // 2. 写Redis
        followRedisService.follow(userId, targetUserId);

        // 3. 发送事件通知
        FollowEvent event = new FollowEvent(userId, targetUserId);
        rocketMQTemplate.send("social-follow-topic", event);

        // 4. 更新计数器
        followerMapper.incrementFollowerCount(targetUserId);
        followMapper.incrementFollowCount(userId);

        return Result.success();
    }
}

2.3 关系链推送

当用户发布新动态时，系统需要将动态推送给其粉丝。关系链推送有推送模式和拉取模式两种。

推模式（Fanout-on-Write）： 发布者发布动态时，将动态写入所有粉丝的收件箱（Timeline）。

java

@Service
public class TimelinePushService {

    @Autowired
    private StringRedisTemplate redisTemplate;

    private static final String TIMELINE_KEY = "timeline:inbox:";

    /** 发布动态时推送给粉丝 */
    public void fanoutPost(SocialPost post, List<Long> followerIds) {
        // 每个粉丝的收件箱写入一条消息ID
        for (Long followerId : followerIds) {
            String key = TIMELINE_KEY + followerId;
            redisTemplate.opsForZSet().add(key,
                String.valueOf(post.getPostId()),
                (double) post.getCreateTime().toInstant(ZoneOffset.UTC).toEpochMilli());
            // 控制收件箱长度，超出部分裁剪
            redisTemplate.opsForZSet().removeRangeByScore(key, 0,
                System.currentTimeMillis() - 30L * 24 * 3600 * 1000);
        }
    }

    /** 拉取收件箱动态（分页） */
    public List<Long> pullTimeline(Long userId, int page, int size) {
        String key = TIMELINE_KEY + userId;
        long end = System.currentTimeMillis();
        long start = end - 30L * 24 * 3600 * 1000;
        Set<String> postIds = redisTemplate.opsForZSet().reverseRangeByScore(key, start, end,
            (page - 1) * size, size);
        return postIds.stream().map(Long::valueOf).collect(Collectors.toList());
    }
}

拉模式（Fanout-on-Read）： 发布者只将动态写入自己的发件箱，粉丝拉取时实时聚合关注列表的动态。

java

@Service
public class TimelinePullService {

    /** 拉取粉丝时间线（合并关注列表的发件箱） */
    public List<SocialPost> pullTimeline(Long userId, int page, int size) {
        // 1. 获取关注列表
        List<Long> followIds = followService.listFollowIds(userId);

        // 2. 从每个关注者的发件箱拉取最新动态
        List<SocialPost> result = new ArrayList<>();
        for (Long followId : followIds) {
            String outboxKey = "timeline:outbox:" + followId;
            Set<String> postIds = redisTemplate.opsForZSet()
                .reverseRange(outboxKey, 0, 10);
            for (String postId : postIds) {
                SocialPost post = postService.getPostById(Long.valueOf(postId));
                if (post != null) {
                    result.add(post);
                }
            }
        }

        // 3. 按时间排序返回
        result.sort((a, b) -> b.getCreateTime().compareTo(a.getCreateTime()));
        return result.stream().skip((page - 1L) * size).limit(size).collect(Collectors.toList());
    }
}

推拉对比：

特性	推模式	拉模式	混合模式
写入成本	高（粉丝越多成本越高）	低（只写一份）	普通用户推，大V拉
读取延迟	低（预计算）	高（实时聚合）	较低
适合场景	普通用户	大V用户/百万粉丝	综合方案
存储成本	高（N份拷贝）	低（1份）	中等

3. 即时通讯

3.1 在线状态管理

在线状态通过 Redis 心跳机制实现，使用 ZSet 存储在线用户列表，利用有序集合的 Score 存储最后心跳时间。

java

@Component
public class OnlineStatusManager {

    @Autowired
    private StringRedisTemplate redisTemplate;

    private static final String ONLINE_SET = "im:online:users";
    private static final long HEARTBEAT_TIMEOUT = 60L;   // 60秒无心跳视为离线
    private static final long HEARTBEAT_INTERVAL = 30L;  // 30秒发送一次心跳

    /** 用户上线，记录在线状态和连接信息 */
    public void userOnline(Long userId, String serverId, String sessionId) {
        long now = System.currentTimeMillis();

        // 1. 更新在线ZSet
        redisTemplate.opsForZSet().add(ONLINE_SET, String.valueOf(userId), now);

        // 2. 记录会话与服务器的映射关系
        String sessionKey = "im:session:" + userId;
        Map<String, String> sessionInfo = new HashMap<>();
        sessionInfo.put("serverId", serverId);
        sessionInfo.put("sessionId", sessionId);
        sessionInfo.put("onlineTime", String.valueOf(now));
        redisTemplate.opsForHash().putAll(sessionKey, sessionInfo);

        // 3. 设置会话过期时间
        redisTemplate.expire(sessionKey, Duration.ofSeconds(HEARTBEAT_TIMEOUT * 3));
    }

    /** 心跳更新 */
    public void heartbeat(Long userId) {
        long now = System.currentTimeMillis();
        redisTemplate.opsForZSet().add(ONLINE_SET, String.valueOf(userId), now);

        // 更新会话过期时间
        String sessionKey = "im:session:" + userId;
        redisTemplate.expire(sessionKey, Duration.ofSeconds(HEARTBEAT_TIMEOUT * 3));
    }

    /** 用户离线 */
    public void userOffline(Long userId) {
        redisTemplate.opsForZSet().remove(ONLINE_SET, String.valueOf(userId));
        redisTemplate.delete("im:session:" + userId);
    }

    /** 清理超时离线用户（定时任务调用） */
    @Scheduled(fixedRate = 30000)
    public void cleanOfflineUsers() {
        long deadline = System.currentTimeMillis() - HEARTBEAT_TIMEOUT * 1000;
        redisTemplate.opsForZSet().removeRangeByScore(ONLINE_SET, 0, deadline);
    }

    /** 判断用户是否在线 */
    public boolean isOnline(Long userId) {
        Double score = redisTemplate.opsForZSet().score(ONLINE_SET, String.valueOf(userId));
        if (score == null) {
            return false;
        }
        return (System.currentTimeMillis() - score.longValue()) < HEARTBEAT_TIMEOUT * 1000;
    }

    /** 获取在线用户数 */
    public long getOnlineCount() {
        Long count = redisTemplate.opsForZSet().size(ONLINE_SET);
        return count != null ? count : 0;
    }

    /** 获取用户所在网关服务器 */
    public String getUserServer(Long userId) {
        String sessionKey = "im:session:" + userId;
        Object serverId = redisTemplate.opsForHash().get(sessionKey, "serverId");
        return serverId != null ? serverId.toString() : null;
    }
}

3.2 消息模型

即时通讯的消息分为三种类型：单聊消息、群聊消息和系统消息。

消息统一模型：

java

@Data
public class IMMessage {

    // ========== 消息标识 ==========
    private Long msgId;                         // 消息ID，全局唯一
    private Long seqId;                         // 时序ID，全局递增
    private String msgKey;                      // 消息去重Key

    // ========== 消息路由 ==========
    private Long fromUserId;                    // 发送者ID
    private Long toUserId;                      // 接收者ID（单聊）/ 群ID（群聊）
    private Integer msgType;                    // 消息类型：1=单聊 2=群聊 3=系统消息

    // ========== 消息内容 ==========
    private Integer contentType;                // 内容类型：1=文本 2=图片 3=语音 4=视频 5=位置 6=文件
    private String content;                     // 消息内容（文本内容或媒体URL）
    private String extra;                       // 扩展字段，JSON格式

    // ========== 消息状态 ==========
    private Integer status;                     // 状态：0=发送中 1=已送达 2=已读 3=撤回
    private Long clientSeq;                     // 客户端序列号

    // ========== 时间字段 ==========
    private Long timestamp;                     // 消息时间戳（毫秒）
}

消息内容类型枚举：

java

public enum MsgContentType {
    TEXT(1, "文本"),
    IMAGE(2, "图片"),
    VOICE(3, "语音"),
    VIDEO(4, "视频"),
    LOCATION(5, "位置"),
    FILE(6, "文件"),
    SYSTEM(99, "系统消息");

    private final int code;
    private final String desc;
}

3.3 消息投递

消息投递策略根据接收方在线状态动态选择：在线走 WebSocket 实时推送，离线走 Push 通知。

java

@Component
public class MessageDeliveryService {

    @Autowired
    private OnlineStatusManager onlineStatusManager;

    @Autowired
    private WebSocketSender webSocketSender;

    @Autowired
    private PushNotificationService pushService;

    @Autowired
    private MessageStoreService messageStoreService;

    /** 投递消息 */
    public void deliverMessage(IMMessage message) {
        // 1. 持久化消息
        messageStoreService.saveMessage(message);

        // 2. 更新最近会话
        messageStoreService.updateRecentConversation(message);

        // 3. 根据在线状态决定投递方式
        boolean isOnline = onlineStatusManager.isOnline(message.getToUserId());

        if (isOnline) {
            // 在线走WebSocket
            String serverId = onlineStatusManager.getUserServer(message.getToUserId());
            webSocketSender.sendToUser(message.getToUserId(), message, serverId);
        } else {
            // 离线走Push通知
            pushService.sendPushNotification(message.getToUserId(), message);
        }
    }

    /** 群聊消息投递 */
    public void deliverGroupMessage(IMMessage message, List<Long> memberIds) {
        // 1. 持久化群消息
        messageStoreService.saveGroupMessage(message);

        // 2. 批量查询在线状态
        for (Long memberId : memberIds) {
            if (memberId.equals(message.getFromUserId())) {
                continue; // 不给自己发
            }
            IMMessage copy = message.clone();
            copy.setToUserId(memberId);

            if (onlineStatusManager.isOnline(memberId)) {
                String serverId = onlineStatusManager.getUserServer(memberId);
                webSocketSender.sendToUser(memberId, copy, serverId);
            } else {
                pushService.sendPushNotification(memberId, copy);
            }
        }
    }
}

3.4 消息时序

消息时序通过全局递增的 seq_id 保证，使用 Redis 自增或数据库序列实现。

java

@Component
public class SeqIdGenerator {

    @Autowired
    private StringRedisTemplate redisTemplate;

    private static final String GLOBAL_SEQ_KEY = "im:global:seq";

    /** 获取全局递增seq_id */
    public long nextSeq() {
        Long seq = redisTemplate.opsForValue().increment(GLOBAL_SEQ_KEY);
        if (seq == null) {
            throw new RuntimeException("seq_id生成失败");
        }
        return seq;
    }

    /** 获取用户的会话seq（用于客户端对比消息连续性） */
    public long getUserConversationSeq(Long userId, Long targetUserId) {
        String key = "im:conv:seq:" + Math.min(userId, targetUserId)
            + ":" + Math.max(userId, targetUserId);
        Long seq = redisTemplate.opsForValue().increment(key);
        return seq != null ? seq : 0;
    }
}

seq_id 的作用：

保证消息的全局有序性，接收方根据 seq_id 判断消息顺序
用于客户端去重，同一 seq_id 的消息只展示一次
离线消息拉取时作为分页游标，避免重复

3.5 消息存储

消息存储采用三表设计：消息内容表、消息索引表和最近会话表。

消息内容表：

sql

CREATE TABLE `im_message_content` (
    `msg_id`        BIGINT       NOT NULL COMMENT '消息ID（分布式ID）',
    `from_user_id`  BIGINT       NOT NULL COMMENT '发送者ID',
    `msg_type`      TINYINT      NOT NULL COMMENT '消息类型：1=单聊 2=群聊 3=系统',
    `content_type`  TINYINT      NOT NULL COMMENT '内容类型：1=文本 2=图片 3=语音 4=视频',
    `content`       TEXT         NOT NULL COMMENT '消息内容',
    `extra`         JSON         DEFAULT NULL COMMENT '扩展字段',
    `create_time`   DATETIME(3)  NOT NULL COMMENT '消息时间',
    PRIMARY KEY (`msg_id`),
    KEY `idx_from_user_time` (`from_user_id`, `create_time`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='消息内容表';

消息索引表：

sql

CREATE TABLE `im_message_index` (
    `id`            BIGINT       NOT NULL AUTO_INCREMENT COMMENT '主键',
    `msg_id`        BIGINT       NOT NULL COMMENT '消息ID',
    `seq_id`        BIGINT       NOT NULL COMMENT '全局递增序列号',
    `owner_id`      BIGINT       NOT NULL COMMENT '所属用户ID（消息接收方）',
    `peer_id`       BIGINT       NOT NULL COMMENT '对方ID（用户或群ID）',
    `msg_type`      TINYINT      NOT NULL COMMENT '消息类型',
    `status`        TINYINT      NOT NULL DEFAULT 0 COMMENT '状态：0=未读 1=已读 2=已撤回',
    `create_time`   DATETIME(3)  NOT NULL COMMENT '消息时间',
    PRIMARY KEY (`id`),
    UNIQUE KEY `uk_owner_msg` (`owner_id`, `msg_id`),
    KEY `idx_owner_peer_time` (`owner_id`, `peer_id`, `create_time`),
    KEY `idx_owner_seq` (`owner_id`, `seq_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='消息索引表';

最近会话表：

sql

CREATE TABLE `im_recent_conversation` (
    `id`            BIGINT       NOT NULL AUTO_INCREMENT COMMENT '主键',
    `user_id`       BIGINT       NOT NULL COMMENT '用户ID',
    `peer_id`       BIGINT       NOT NULL COMMENT '对方ID（用户或群ID）',
    `conversation_type` TINYINT  NOT NULL COMMENT '会话类型：1=单聊 2=群聊',
    `last_msg_id`   BIGINT       NOT NULL COMMENT '最后一条消息ID',
    `last_content`  VARCHAR(256) DEFAULT NULL COMMENT '最后一条消息预览',
    `last_time`     DATETIME(3)  NOT NULL COMMENT '最后消息时间',
    `unread_count`  INT          NOT NULL DEFAULT 0 COMMENT '未读数',
    `top_time`      DATETIME     DEFAULT NULL COMMENT '置顶时间（NULL=未置顶）',
    `create_time`   DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP,
    `update_time`   DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    PRIMARY KEY (`id`),
    UNIQUE KEY `uk_user_peer` (`user_id`, `peer_id`, `conversation_type`),
    KEY `idx_user_last_time` (`user_id`, `last_time`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='最近会话表';

4. 即时通讯架构设计

IM 系统采用四层架构：接入层、路由层、逻辑层和存储层，各层职责清晰，通过服务发现和消息队列解耦。

text

┌─────────────────────────────────────────────────────────────────────────────┐
│                             接入层 (Access Layer)                            │
│                ┌─────────────────────────────────────────┐                  │
│                │          WebSocket 网关集群                │                  │
│                │  ┌──────────┐ ┌──────────┐ ┌──────────┐ │                  │
│                │  │ Gateway-1 │ │ Gateway-2 │ │ Gateway-N │ │                  │
│                │  │(连接管理) │ │(连接管理) │ │(连接管理) │ │                  │
│                │  └─────┬────┘ └─────┬────┘ └─────┬────┘ │                  │
│                │        │            │            │       │                  │
│                │   ┌────┴────────────┴────────────┴───┐   │                  │
│                │   │    连接路由表 (Redis Hash)         │   │                  │
│                │   │  userId -> serverId:sessionId     │   │                  │
│                │   └───────────────────────────────────┘   │                  │
│                └─────────────────────────────────────────┘                  │
└──────────────────────────────┬──────────────────────────────────────────────┘
                               │
                               ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                         路由层 (Router Layer)                                 │
│                ┌─────────────────────────────────────────┐                  │
│                │   一致性哈希环 (Consistent Hash Ring)     │                  │
│                │    ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐ │                  │
│                │    │Channel│ │Channel│ │Channel│ │Channel│ │                  │
│                │    │  1   │ │  2   │ │  3   │ │  N   │ │                  │
│                │    └──────┘ └──────┘ └──────┘ └──────┘ │                  │
│                │  分配策略: hash(conversation_id) % N    │                  │
│                └─────────────────────────────────────────┘                  │
└──────────────────────────────┬──────────────────────────────────────────────┘
                               │
                               ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                         逻辑层 (Logic Layer)                                  │
│  ┌─────────────────┐  ┌─────────────────┐  ┌─────────────────────┐        │
│  │   消息处理服务     │  │   关系链服务       │  │    群聊管理服务       │        │
│  │  - 消息校验/过滤  │  │  - 好友管理       │  │  - 群组CRUD        │        │
│  │  - 消息路由      │  │  - 关注/粉丝      │  │  - 成员管理         │        │
│  │  - 消息持久化    │  │  - 黑名单检查     │  │  - 群公告/管理     │        │
│  │  - 消息投递      │  │  - 关系链推送     │  │  - @所有人         │        │
│  └────────┬────────┘  └────────┬────────┘  └──────────┬──────────┘        │
└───────────┼────────────────────┼───────────────────────┼───────────────────┘
            │                    │                       │
            ▼                    ▼                       ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                         存储层 (Storage Layer)                               │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐   │
│  │   Redis       │  │    MySQL      │  │  消息队列     │  │   对象存储    │   │
│  │  - 在线状态   │  │  - 消息内容   │  │(RocketMQ/    │  │  - 图片      │   │
│  │  - 会话缓存   │  │  - 消息索引   │  │  Kafka)      │  │  - 语音      │   │
│  │  - seq_id     │  │  - 关系链     │  │ - 消息异步处理│  │  - 文件      │   │
│  │  - 未读数     │  │  - 群组信息   │  │ - 事件通知   │  │              │   │
│  └──────────────┘  └──────────────┘  └──────────────┘  └──────────────┘   │
└─────────────────────────────────────────────────────────────────────────────┘

4.1 接入层 (Access Layer)

接入层负责 WebSocket 连接管理和协议解析，是一个无状态网关集群。

java

@Component
public class WebSocketServer {

    private static final Map<String, Channel> ONLINE_CHANNELS = new ConcurrentHashMap<>();

    /** WebSocket连接建立 */
    @ServerEndpoint(value = "/im/ws")
    public void onOpen(Session session, @PathParam("userId") Long userId) {
        String sessionId = session.getId();
        String serverId = getLocalServerId();

        // 注册到在线管理器
        onlineStatusManager.userOnline(userId, serverId, sessionId);
        ONLINE_CHANNELS.put(sessionId, session);

        log.info("用户[{}]上线，sessionId={}, serverId={}", userId, sessionId, serverId);
    }

    /** 接收消息 */
    @OnMessage
    public void onMessage(Session session, String messageText) {
        // 1. 解析协议（支持Protobuf/JSON）
        IMMessage message = MessageProtocol.decode(messageText);

        // 2. 校验消息合法性
        if (!validateMessage(message)) {
            sendError(session, "消息格式非法");
            return;
        }

        // 3. 发送到消息队列，由逻辑层处理
        messageProducer.send(message);
    }

    /** 连接关闭 */
    @OnClose
    public void onClose(Session session) {
        String sessionId = session.getId();
        Long userId = getUserIdBySession(sessionId);
        ONLINE_CHANNELS.remove(sessionId);
        onlineStatusManager.userOffline(userId);

        log.info("用户[{}]离线", userId);
    }

    /** 向用户发送消息 */
    public void sendMessage(Long userId, IMMessage message) {
        String sessionId = onlineStatusManager.getUserSession(userId);
        Channel channel = ONLINE_CHANNELS.get(sessionId);
        if (channel != null && channel.isActive()) {
            channel.writeAndFlush(MessageProtocol.encode(message));
        }
    }
}

4.2 路由层 (Router Layer)

路由层通过一致性哈希算法将消息路由到对应的 Channel，保证同一会话的消息路由到同一处理节点。

java

@Component
public class MessageRouter {

    private final ConsistentHashRouter<String> router;

    private static final int VIRTUAL_NODES = 160;

    public MessageRouter(List<String> channelNodes) {
        this.router = new ConsistentHashRouter<>(channelNodes, VIRTUAL_NODES);
    }

    /** 根据会话ID路由到具体的Channel */
    public String routeChannel(String conversationId) {
        return router.routeNode(conversationId);
    }

    /** 消息路由处理 */
    public void routeMessage(IMMessage message) {
        String conversationId = getConversationId(message);
        String targetChannel = routeChannel(conversationId);

        // 发送到目标Channel的消息队列
        messageProducer.sendToChannel(targetChannel, message);
    }

    private String getConversationId(IMMessage message) {
        if (message.getMsgType() == 1) {
            // 单聊: min(userId, toUserId):max(userId, toUserId)
            long min = Math.min(message.getFromUserId(), message.getToUserId());
            long max = Math.max(message.getFromUserId(), message.getToUserId());
            return "single:" + min + ":" + max;
        } else if (message.getMsgType() == 2) {
            // 群聊: group:groupId
            return "group:" + message.getToUserId();
        }
        return "system:" + message.getToUserId();
    }
}

一致性哈希实现：

java

public class ConsistentHashRouter<T extends Node> {

    private final TreeMap<Long, T> virtualNodes = new TreeMap<>();
    private final HashFunction hashFunction;

    public ConsistentHashRouter(Collection<T> nodes, int virtualNodeCount) {
        this.hashFunction = new MD5Hash();
        for (T node : nodes) {
            for (int i = 0; i < virtualNodeCount; i++) {
                virtualNodes.put(hashFunction.hash(node.getKey() + "-VN-" + i), node);
            }
        }
    }

    public T routeNode(String key) {
        Long hash = hashFunction.hash(key);
        Map.Entry<Long, T> entry = virtualNodes.ceilingEntry(hash);
        if (entry == null) {
            entry = virtualNodes.firstEntry();
        }
        return entry.getValue();
    }
}

4.3 逻辑层 (Logic Layer)

逻辑层是 IM 系统的核心处理层，负责消息校验、关系链检查、消息持久化和投递决策。

java

@Service
public class MessageLogicService {

    @Autowired
    private MessageValidator messageValidator;

    @Autowired
    private FriendRelationService friendRelationService;

    @Autowired
    private MessageStoreService messageStoreService;

    @Autowired
    private MessageDeliveryService deliveryService;

    @Autowired
    private MessageQueueProducer mqProducer;

    /** 处理消息（由MQ消费者调用） */
    @RocketMQMessageListener(topic = "im-message-topic", consumerGroup = "im-logic-group")
    public void handleMessage(IMMessage message) {
        // 1. 消息校验
        Result<Void> validateResult = messageValidator.validate(message);
        if (validateResult.isFailed()) {
            log.warn("消息校验失败: {}", validateResult.getMessage());
            return;
        }

        // 2. 关系链检查（黑名单过滤）
        if (message.getMsgType() == 1) {
            boolean isBlocked = friendRelationService.isBlocked(
                message.getToUserId(), message.getFromUserId());
            if (isBlocked) {
                log.info("消息被拦截，接收方已拉黑发送方");
                return;
            }
        }

        // 3. 生成seq_id
        long seqId = seqIdGenerator.nextSeq();
        message.setSeqId(seqId);

        // 4. 持久化消息
        messageStoreService.saveMessage(message);

        // 5. 投递消息
        deliveryService.deliverMessage(message);

        // 6. 发送回执给发送方
        sendAckToSender(message);
    }
}

4.4 存储层 (Storage Layer)

存储层组合使用 Redis、MySQL 和消息队列，分别承担缓存、持久化和异步处理的职责。

Redis 缓存设计：

Key	数据结构	用途	过期时间
`im:online:users`	ZSet	在线用户列表	无
`im:session:{userId}`	Hash	用户会话与服务器映射	3min
`im:global:seq`	String	全局递增seq_id	无
`im:unread:{userId}:{peerId}`	String	单聊未读数	无
`im:group:unread:{groupId}:{userId}`	String	群聊未读数	无
`im:msg:ack:{msgId}`	String	消息ACK状态	1h

5. 群聊设计

5.1 群组表

sql

CREATE TABLE `im_group` (
    `group_id`       BIGINT       NOT NULL AUTO_INCREMENT COMMENT '群ID',
    `group_name`     VARCHAR(128) NOT NULL COMMENT '群名称',
    `group_avatar`   VARCHAR(256) DEFAULT NULL COMMENT '群头像',
    `owner_id`       BIGINT       NOT NULL COMMENT '群主ID',
    `group_type`     TINYINT      NOT NULL DEFAULT 0 COMMENT '群类型：0=普通群 1=全员群 2=部门群',
    `max_members`    INT          NOT NULL DEFAULT 200 COMMENT '最大成员数',
    `notice`         TEXT         DEFAULT NULL COMMENT '群公告',
    `status`         TINYINT      NOT NULL DEFAULT 1 COMMENT '状态：0=解散 1=正常 2=全员禁言',
    `create_time`    DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP,
    `update_time`    DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    PRIMARY KEY (`group_id`),
    KEY `idx_owner_id` (`owner_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='群组表';

群成员表：

sql

CREATE TABLE `im_group_member` (
    `id`            BIGINT       NOT NULL AUTO_INCREMENT COMMENT '主键',
    `group_id`      BIGINT       NOT NULL COMMENT '群ID',
    `user_id`       BIGINT       NOT NULL COMMENT '用户ID',
    `role`          TINYINT      NOT NULL DEFAULT 0 COMMENT '角色：0=普通成员 1=管理员 2=群主',
    `nickname`      VARCHAR(64)  DEFAULT NULL COMMENT '群昵称',
    `join_time`     DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '入群时间',
    `mute_time`     DATETIME     DEFAULT NULL COMMENT '禁言截止时间（NULL=未禁言）',
    `status`        TINYINT      NOT NULL DEFAULT 1 COMMENT '状态：0=退群 1=正常',
    PRIMARY KEY (`id`),
    UNIQUE KEY `uk_group_user` (`group_id`, `user_id`),
    KEY `idx_user_id` (`user_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='群成员表';

5.2 消息扩散模式

群聊消息的扩散有三种模式，分别适用于不同的场景。

写扩散 (Write Fanout)

发送者将消息写入群聊的中央消息存储，系统为每个群成员在消息索引表中插入一条索引记录。

java

@Service
public class WriteFanoutStrategy {

    /** 写扩散模式发送群消息 */
    public void sendGroupMessage(IMMessage message, Long groupId, List<Long> memberIds) {
        // 1. 保存消息内容（仅一份）
        messageStoreService.saveMessageContent(message);

        // 2. 为每个成员写索引（写扩散）
        List<ImMessageIndex> indices = new ArrayList<>();
        for (Long memberId : memberIds) {
            if (memberId.equals(message.getFromUserId())) {
                continue;
            }
            ImMessageIndex index = new ImMessageIndex();
            index.setMsgId(message.getMsgId());
            index.setOwnerId(memberId);
            index.setPeerId(groupId);
            index.setMsgType(2);
            index.setSeqId(message.getSeqId());
            index.setCreateTime(LocalDateTime.now());
            indices.add(index);
        }
        messageStoreService.batchSaveIndices(indices);
    }
}

优点： 读取时无需聚合，直接按时间倒序查询索引表即可，查询性能好。 缺点： 大群场景下写放大严重（1000人群需要写1000条索引）。

读扩散 (Read Fanout)

消息内容只存一份，不写索引。用户拉取消息时实时聚合群聊消息。

java

@Service
public class ReadFanoutStrategy {

    /** 读取扩散模式拉取群消息 */
    public List<IMMessage> pullGroupMessages(Long userId, Long groupId, Long cursor, int limit) {
        // 1. 获取用户入群时间
        LocalDateTime joinTime = groupMemberMapper.getJoinTime(groupId, userId);

        // 2. 查询群聊消息（按seq_id分页）
        List<ImMessageContent> contents = messageContentMapper.selectByGroup(
            groupId, cursor, limit, joinTime);

        return contents.stream().map(this::convert).collect(Collectors.toList());
    }
}

优点： 写入成本极低，没有写放大问题。 缺点： 读取需要实时聚合，延迟较高，需要额外过滤退群前的消息。

混合模式

结合写扩散和读扩散的优点，根据群规模和活跃度动态选择策略。

java

@Service
public class HybridFanoutStrategy {

    private static final int WRITE_FANOUT_THRESHOLD = 500;  // 低于500人使用写扩散

    @Autowired
    private WriteFanoutStrategy writeFanout;

    @Autowired
    private ReadFanoutStrategy readFanout;

    /** 动态选择扩散策略 */
    public void sendMessage(IMMessage message, Long groupId) {
        int memberCount = groupMemberMapper.countMembers(groupId);

        if (memberCount <= WRITE_FANOUT_THRESHOLD) {
            // 小群使用写扩散
            List<Long> memberIds = groupMemberMapper.listMemberIds(groupId);
            writeFanout.sendGroupMessage(message, groupId, memberIds);
        } else {
            // 大群使用读扩散
            readFanout.sendGroupMessage(message, groupId);
        }
    }
}

三种模式对比：

特性	写扩散	读扩散	混合模式
写入成本	高（O(N)）	低（O(1)）	动态选择
读取成本	低（O(1)）	高（O(N)）	动态平衡
适合场景	小群（<500人）	大群（>500人）	综合场景
存储开销	大（N份索引）	小（1份内容）	中等
实现复杂度	低	低	中

5.3 群公告 / @所有人 / 群管理

群公告管理：

java

@Service
public class GroupNoticeService {

    /** 发布群公告 */
    public Result<Void> publishNotice(Long groupId, Long operatorId, String title, String content) {
        // 1. 校验操作权限（群主或管理员）
        if (!checkAdminPermission(groupId, operatorId)) {
            return Result.fail("无操作权限");
        }

        // 2. 保存公告
        GroupNotice notice = new GroupNotice();
        notice.setGroupId(groupId);
        notice.setTitle(title);
        notice.setContent(content);
        notice.setPublisherId(operatorId);
        groupNoticeMapper.insert(notice);

        // 3. 更新群公告字段
        groupMapper.updateNotice(groupId, title);

        // 4. 发送系统通知给所有群成员
        sendNoticeToMembers(groupId, notice);

        return Result.success();
    }

    /** 发送公告通知 */
    private void sendNoticeToMembers(Long groupId, GroupNotice notice) {
        List<Long> memberIds = groupMemberMapper.listMemberIds(groupId);
        IMMessage sysMsg = new IMMessage();
        sysMsg.setMsgType(3);  // 系统消息
        sysMsg.setContentType(99);
        sysMsg.setContent(JSON.toJSONString(notice));
        sysMsg.setTimestamp(System.currentTimeMillis());

        for (Long memberId : memberIds) {
            sysMsg.setToUserId(memberId);
            messageDeliveryService.deliverMessage(sysMsg);
        }
    }
}

@所有人实现：

java

@Service
public class AtAllService {

    /** @所有人权限检查 */
    public boolean canAtAll(Long groupId, Long userId) {
        GroupMember member = groupMemberMapper.selectByGroupAndUser(groupId, userId);
        if (member == null) {
            return false;
        }
        // 群主和管理员可以@所有人，或者群配置允许所有成员@所有人
        return member.getRole() >= 1 || groupConfigMapper.canAtAll(groupId);
    }

    /** 处理@所有人消息 */
    public void processAtAll(IMMessage message, Long groupId) {
        if (!canAtAll(groupId, message.getFromUserId())) {
            throw new BusinessException("无@所有人权限");
        }

        // 标记消息为@所有人类型
        message.setContentType(100);

        // 推送时单独处理@所有人标记
        List<Long> memberIds = groupMemberMapper.listMemberIds(groupId);
        for (Long memberId : memberIds) {
            if (memberId.equals(message.getFromUserId())) {
                continue;
            }
            // 在消息扩展中标记@了该成员
            Map<String, Object> extra = new HashMap<>();
            extra.put("atType", "ALL");
            extra.put("atUserId", memberId);
            message.setExtra(JSON.toJSONString(extra));

            IMMessage copy = message.clone();
            copy.setToUserId(memberId);
            messageDeliveryService.deliverMessage(copy);
        }
    }
}

群管理功能：

java

@Service
public class GroupAdminService {

    /** 转让群主 */
    @Transactional
    public Result<Void> transferOwner(Long groupId, Long currentOwner, Long newOwnerId) {
        validateOwner(groupId, currentOwner);
        validateMember(groupId, newOwnerId);

        // 原群主降为管理员，新群主升为群主
        groupMemberMapper.updateRole(groupId, currentOwner, 1);
        groupMemberMapper.updateRole(groupId, newOwnerId, 2);
        groupMapper.updateOwner(groupId, newOwnerId);

        // 发送群变更通知
        sendGroupChangeNotification(groupId, "群主变更为 " + newOwnerId);
        return Result.success();
    }

    /** 踢出群成员 */
    @Transactional
    public Result<Void> kickMember(Long groupId, Long operatorId, Long targetUserId) {
        GroupMember operator = groupMemberMapper.selectByGroupAndUser(groupId, operatorId);
        GroupMember target = groupMemberMapper.selectByGroupAndUser(groupId, targetUserId);

        validateKickPermission(operator, target);

        // 移除群成员
        groupMemberMapper.updateStatus(groupId, targetUserId, 0);
        // 清理群会话
        recentConversationMapper.deleteByUserAndPeer(targetUserId, groupId);

        return Result.success();
    }

    /** 设置管理员 */
    public Result<Void> setAdmin(Long groupId, Long ownerId, Long targetUserId, boolean isAdmin) {
        validateOwner(groupId, ownerId);
        int newRole = isAdmin ? 1 : 0;
        groupMemberMapper.updateRole(groupId, targetUserId, newRole);
        return Result.success();
    }

    /** 全员禁言/解禁 */
    public Result<Void> toggleMuteAll(Long groupId, Long operatorId, boolean mute) {
        validateAdminPermission(groupId, operatorId);
        groupMapper.updateStatus(groupId, mute ? 2 : 1);
        // 发送系统通知
        String msg = mute ? "全员禁言已开启" : "全员禁言已关闭";
        sendGroupChangeNotification(groupId, msg);
        return Result.success();
    }
}

6. 消息可靠性

6.1 ACK 机制

消息 ACK 机制确保消息从发送端到接收端的可靠传递，包括客户端确认、重试和超时处理。

text

发送端                         服务端                         接收端
  │                             │                             │
  │  1. 发送消息(msgId, seq)     │                             │
  │ ───────────────────────────> │                             │
  │                             │  2. 持久化消息               │
  │                             │  3. 投递消息                 │
  │                             │ ──────────────────────────> │
  │                             │                             │  4. 收到消息
  │                             │                             │  5. 发送ACK
  │                             │  <────────────────────────── │
  │                             │                             │
  │  6. ACK回执确认              │                             │
  │ <─────────────────────────── │                             │
  │                             │                             │

服务端 ACK 管理器：

java

@Component
public class AckManager {

    @Autowired
    private StringRedisTemplate redisTemplate;

    private static final String ACK_KEY_PREFIX = "im:ack:";
    private static final long ACK_TIMEOUT_MS = 10000;   // 等待ACK超时时间
    private static final int MAX_RETRY_COUNT = 3;        // 最大重试次数

    /** 等待消息ACK */
    public void waitAck(IMMessage message) {
        String ackKey = ACK_KEY_PREFIX + message.getMsgId();

        // 记录待ACK消息
        Map<String, String> ackInfo = new HashMap<>();
        ackInfo.put("msgId", String.valueOf(message.getMsgId()));
        ackInfo.put("toUserId", String.valueOf(message.getToUserId()));
        ackInfo.put("retryCount", "0");
        ackInfo.put("sendTime", String.valueOf(System.currentTimeMillis()));
        redisTemplate.opsForHash().putAll(ackKey, ackInfo);
        redisTemplate.expire(ackKey, Duration.ofMinutes(5));

        // 延迟检查ACK状态
        scheduleAckCheck(message);
    }

    /** 接收端确认ACK */
    public void receiveAck(Long msgId, Long userId) {
        String ackKey = ACK_KEY_PREFIX + msgId;
        redisTemplate.opsForHash().put(ackKey, "status", "ACKED");
        redisTemplate.opsForHash().put(ackKey, "ackTime", String.valueOf(System.currentTimeMillis()));
    }

    /** 定时检查ACK（未ACK则重试） */
    @Scheduled(fixedDelay = 3000)
    public void checkPendingAcks() {
        // 扫描待ACK消息（实际可通过延迟队列实现）
        Set<String> keys = redisTemplate.keys("im:ack:*");
        if (keys == null || keys.isEmpty()) {
            return;
        }

        for (String key : keys) {
            Map<Object, Object> ackInfo = redisTemplate.opsForHash().entries(key);
            if (ackInfo.isEmpty()) {
                continue;
            }

            String status = (String) ackInfo.get("status");
            if ("ACKED".equals(status)) {
                continue;  // 已确认
            }

            long sendTime = Long.parseLong((String) ackInfo.get("sendTime"));
            int retryCount = Integer.parseInt((String) ackInfo.getOrDefault("retryCount", "0"));

            if (System.currentTimeMillis() - sendTime > ACK_TIMEOUT_MS) {
                if (retryCount >= MAX_RETRY_COUNT) {
                    // 超过最大重试次数，标记为失败
                    redisTemplate.opsForHash().put(key, "status", "FAILED");
                    log.warn("消息[{}]投递失败，已达最大重试次数", key);
                } else {
                    // 重试投递
                    retryCount++;
                    redisTemplate.opsForHash().put(key, "retryCount", String.valueOf(retryCount));
                    redisTemplate.opsForHash().put(key, "sendTime",
                        String.valueOf(System.currentTimeMillis()));

                    Long msgId = Long.parseLong(key.replace("im:ack:", ""));
                    IMMessage message = messageStoreService.getMessage(msgId);
                    if (message != null) {
                        messageDeliveryService.deliverMessage(message);
                        log.info("消息[{}]第{}次重试", msgId, retryCount);
                    }
                }
            }
        }
    }
}

客户端 ACK 发送：

java

// WebSocket客户端收到消息后自动发送ACK
@OnMessage
public void onMessage(Session session, String messageText) {
    IMMessage message = MessageProtocol.decode(messageText);

    // 处理消息
    displayMessage(message);

    // 发送ACK回执
    IMACK ack = new IMACK();
    ack.setMsgId(message.getMsgId());
    ack.setUserId(currentUserId);
    ack.setTimestamp(System.currentTimeMillis());
    session.getBasicRemote().sendText(MessageProtocol.encode(ack));
}

6.2 消息去重

消息去重通过全局唯一的消息 ID 实现，在服务端和客户端双重保证。

服务端去重：

java

@Component
public class MessageDeduplication {

    @Autowired
    private StringRedisTemplate redisTemplate;

    private static final String DEDUP_KEY_PREFIX = "im:dedup:";

    /** 检查消息是否已处理（使用Redis SET NX实现） */
    public boolean tryProcess(String msgKey) {
        String key = DEDUP_KEY_PREFIX + msgKey;
        Boolean success = redisTemplate.opsForValue()
            .setIfAbsent(key, "1", Duration.ofHours(24));
        return Boolean.TRUE.equals(success);
    }

    /** 生成消息去重Key */
    public String buildMsgKey(IMMessage message) {
        // 客户端发送时携带 msgKey = MD5(userId + timestamp + random)
        if (StringUtils.hasText(message.getMsgKey())) {
            return message.getMsgKey();
        }
        // 服务端生成
        return DigestUtils.md5DigestAsHex(
            (message.getFromUserId() + "-" + message.getTimestamp() + "-"
             + ThreadLocalRandom.current().nextInt()).getBytes()
        );
    }
}

客户端去重：

java

// 客户端维护已处理消息ID集合，防止重复展示
public class MessageDisplayManager {

    private final Set<Long> processedMsgIds = new ConcurrentSkipListSet<>();

    /** 展示消息（去重后） */
    public boolean tryDisplay(IMMessage message) {
        if (processedMsgIds.contains(message.getMsgId())) {
            log.debug("消息已展示，跳过：msgId={}", message.getMsgId());
            return false;
        }
        processedMsgIds.add(message.getMsgId());

        // 限制集合大小，防止内存泄漏
        if (processedMsgIds.size() > 10000) {
            processedMsgIds.clear();
        }

        // 展示消息
        renderMessage(message);
        return true;
    }
}

6.3 离线消息拉取

用户上线后需要拉取离线期间的消息，采用分页按时间倒序的方式拉取。

java

@Service
public class OfflineMessageService {

    @Autowired
    private MessageIndexMapper messageIndexMapper;

    @Autowired
    private MessageContentMapper messageContentMapper;

    /** 拉取离线消息（按时间倒序分页） */
    public Result<OfflineMessageVO> pullOfflineMessages(Long userId, Long cursor, int limit) {
        // 参数默认值
        if (cursor == null || cursor <= 0) {
            cursor = Long.MAX_VALUE;
        }
        if (limit <= 0 || limit > 100) {
            limit = 20;
        }

        // 1. 查询消息索引（按seq_id倒序）
        List<ImMessageIndex> indices = messageIndexMapper.selectByOwner(
            userId, cursor, limit);

        if (indices.isEmpty()) {
            return Result.success(OfflineMessageVO.empty());
        }

        // 2. 批量查询消息内容
        List<Long> msgIds = indices.stream()
            .map(ImMessageIndex::getMsgId)
            .collect(Collectors.toList());
        List<ImMessageContent> contents = messageContentMapper.selectBatch(msgIds);
        Map<Long, ImMessageContent> contentMap = contents.stream()
            .collect(Collectors.toMap(ImMessageContent::getMsgId, Function.identity()));

        // 3. 组装消息
        List<IMMessage> messages = new ArrayList<>();
        for (ImMessageIndex index : indices) {
            ImMessageContent content = contentMap.get(index.getMsgId());
            if (content != null) {
                IMMessage msg = convert(index, content);
                messages.add(msg);
            }
        }

        // 4. 更新未读数为0
        messageIndexMapper.markAllRead(userId);

        // 5. 返回分页结果
        long nextCursor = indices.get(indices.size() - 1).getSeqId();
        boolean hasMore = indices.size() >= limit;

        OfflineMessageVO vo = new OfflineMessageVO();
        vo.setMessages(messages);
        vo.setNextCursor(nextCursor);
        vo.setHasMore(hasMore);

        return Result.success(vo);
    }

    /** 拉取离线消息（按会话分组） */
    public Result<List<ConversationMessagesVO>> pullByConversation(Long userId) {
        // 1. 查询有未读消息的会话列表
        List<RecentConversation> conversations =
            recentConversationMapper.selectByUserWithUnread(userId);

        List<ConversationMessagesVO> result = new ArrayList<>();
        for (RecentConversation conv : conversations) {
            // 2. 拉取每个会话最近的消息
            List<ImMessageIndex> indices = messageIndexMapper.selectByOwnerAndPeer(
                userId, conv.getPeerId(), conv.getConversationType(), 20);

            List<Long> msgIds = indices.stream()
                .map(ImMessageIndex::getMsgId)
                .collect(Collectors.toList());
            List<ImMessageContent> contents = messageContentMapper.selectBatch(msgIds);

            ConversationMessagesVO vo = new ConversationMessagesVO();
            vo.setPeerId(conv.getPeerId());
            vo.setConversationType(conv.getConversationType());
            vo.setMessages(convertToMsgList(indices, contents));
            result.add(vo);
        }

        return Result.success(result);
    }
}

离线消息拉取流程：

text

用户上线
    │
    ▼
┌─────────────────────────────────────┐
│  1. 查询最近会话列表（含未读数）        │
│     SELECT * FROM im_recent_        │
│     conversation WHERE user_id=?    │
│     AND unread_count > 0            │
└────────────────┬────────────────────┘
                 │
                 ▼
┌─────────────────────────────────────┐
│  2. 按会话拉取离线消息                 │
│     按 seq_id 倒序，每次拉取20条       │
│     SELECT * FROM im_message_index   │
│     WHERE owner_id=? AND seq_id < ?  │
│     ORDER BY seq_id DESC LIMIT 20    │
└────────────────┬────────────────────┘
                 │
                 ▼
┌─────────────────────────────────────┐
│  3. 批量查询消息内容                   │
│     SELECT * FROM im_message_content │
│     WHERE msg_id IN (?, ?, ...)      │
└────────────────┬────────────────────┘
                 │
                 ▼
┌─────────────────────────────────────┐
│  4. 标记消息为已读                     │
│     UPDATE im_message_index          │
│     SET status=1 WHERE owner_id=?    │
│     AND status=0                     │
└────────────────┬────────────────────┘
                 │
                 ▼
┌─────────────────────────────────────┐
│  5. 更新未读数                        │
│     UPDATE im_recent_conversation    │
│     SET unread_count=0               │
│     WHERE user_id=? AND peer_id=?    │
└─────────────────────────────────────┘
                 │
                 ▼
          返回消息列表
          (含 nextCursor 用于分页)

离线消息 SQL 查询：

sql

-- 按游标分页拉取离线消息
SELECT
    idx.msg_id,
    idx.seq_id,
    idx.owner_id,
    idx.peer_id,
    idx.msg_type,
    idx.status AS read_status,
    content.from_user_id,
    content.content_type,
    content.content,
    content.create_time
FROM im_message_index idx
INNER JOIN im_message_content content ON idx.msg_id = content.msg_id
WHERE idx.owner_id = ?
  AND idx.seq_id < ?          -- 游标，首次拉取传 MAX_VALUE
  AND idx.status = 0          -- 只拉取未读消息
ORDER BY idx.seq_id DESC
LIMIT ?;

-- 按会话分组拉取离线消息
SELECT
    idx.msg_id,
    idx.seq_id,
    idx.peer_id,
    idx.msg_type,
    content.from_user_id,
    content.content_type,
    content.content,
    content.create_time
FROM im_message_index idx
INNER JOIN im_message_content content ON idx.msg_id = content.msg_id
WHERE idx.owner_id = ?
  AND idx.peer_id = ?
  AND idx.msg_type = ?
  AND idx.status = 0
ORDER BY idx.seq_id DESC
LIMIT 20;

社交架构设计 / 即时通讯系统 ​

1. 社交产品架构 ​

1.1 用户关系 ​

1.2 用户资料 ​

1.3 内容发布 ​

1.4 互动系统 ​

2. 关系链设计 ​

2.1 好友关系表 ​

2.2 关注/粉丝表 ​

方案一：JSON 数组存储（适合小规模） ​

方案二：Redis Set 存储（适合高并发） ​

方案三：MySQL 关系表存储（适合大规模、复杂查询） ​

三种方案对比总结 ​

2.3 关系链推送 ​

3. 即时通讯 ​

3.1 在线状态管理 ​

3.2 消息模型 ​

3.3 消息投递 ​

3.4 消息时序 ​

3.5 消息存储 ​

4. 即时通讯架构设计 ​

4.1 接入层 (Access Layer) ​

4.2 路由层 (Router Layer) ​

4.3 逻辑层 (Logic Layer) ​

4.4 存储层 (Storage Layer) ​

5. 群聊设计 ​

5.1 群组表 ​

5.2 消息扩散模式 ​

写扩散 (Write Fanout) ​

读扩散 (Read Fanout) ​

混合模式 ​

5.3 群公告 / @所有人 / 群管理 ​

6. 消息可靠性 ​

6.1 ACK 机制 ​

6.2 消息去重 ​

6.3 离线消息拉取 ​

社交架构设计 / 即时通讯系统

1. 社交产品架构

1.1 用户关系

1.2 用户资料

1.3 内容发布

1.4 互动系统

2. 关系链设计

2.1 好友关系表

2.2 关注/粉丝表

方案一：JSON 数组存储（适合小规模）

方案二：Redis Set 存储（适合高并发）

方案三：MySQL 关系表存储（适合大规模、复杂查询）

三种方案对比总结

2.3 关系链推送

3. 即时通讯

3.1 在线状态管理

3.2 消息模型

3.3 消息投递

3.4 消息时序

3.5 消息存储

4. 即时通讯架构设计

4.1 接入层 (Access Layer)

4.2 路由层 (Router Layer)

4.3 逻辑层 (Logic Layer)

4.4 存储层 (Storage Layer)

5. 群聊设计

5.1 群组表

5.2 消息扩散模式

写扩散 (Write Fanout)

读扩散 (Read Fanout)

混合模式

5.3 群公告 / @所有人 / 群管理

6. 消息可靠性

6.1 ACK 机制

6.2 消息去重

6.3 离线消息拉取