基于 Outbox 事务表 + Canal 监听+kafka+多级缓存：高并发社交关注系统全链路架构设计

    在社交类业务系统中，用户关注与取关是访问量极高、并发压力大且对数据一致性要求严苛的核心场景，不仅需要应对用户高频点击、恶意刷接口等流量问题，还要解决消息丢失、重复消费、缓存与数据库数据偏差、粉丝关注计数不准等一系列线上常见难题。传统直接发送消息队列的开发模式，极易出现业务入库成功但消息投递失败，最终引发业务数据不一致的隐患。为此我基于实际业务场景，设计并落地了一套限流防护 + 多级缓存 + 本地事务事件表 + Canal 异步转发的完整高并发社交关系架构，通过前置流量拦截、事务保障事件可靠落地、分层缓存提升查询性能、异步解耦完成后置业务处理，并搭配定时数据校验机制实现数据兜底纠偏，从请求入口到最终数据同步全链路完成优化，在保证系统高性能运行的同时，彻底解决幂等性、消息可靠性、缓存一致性等核心痛点，本文将完整拆解整套架构设计思路、核心代码实现与落地实践经验。

一、整体技术架构总览

分层架构设计：客户端层、API 接入层、业务服务层、多级缓存层、数据持久层、异步事件层、计数纠偏层 核心技术栈梳理：SpringBoot、Redis、Lua 限流、Caffeine 本地缓存、MyBatis、Outbox 事务表、Canal、Kafka

全链路核心流转简述：用户请求→鉴权限流→业务事务→库表写入→事件投递→异步消费→数据更新

二、API 接入层设计：

lua令牌桶设计，限流，防止一个用户指定时间内最多发多少请求

初始桶中有100个令牌，以后每秒生成1个，如果一瞬间有100个请求，消耗完所有令牌，就会拒绝后续的请求同时如果桶满了，后续生成的令牌会直接丢弃

private static final String TOKEN_BUCKET_LUA = """

local key = KEYS[1]
local capacity = tonumber(ARGV[1])
local rate = tonumber(ARGV[2])
local now = redis.call('TIME')[1]
local last = redis.call('HGET', key, 'last')
local tokens = redis.call('HGET', key, 'tokens')
if not last then last = now; tokens = capacity end
local elapsed = tonumber(now) – tonumber(last)
local add = elapsed * rate
tokens = math.min(capacity, tonumber(tokens) + add)
if tokens < 1 then redis.call('HSET', key, 'last', now); redis.call('HSET', key, 'tokens', tokens); return 0 end
tokens = tokens – 1
redis.call('HSET', key, 'last', now)
redis.call('HSET', key, 'tokens', tokens)
redis.call('PEXPIRE', key, 60000)
return 1
""";

三、核心业务服务层设计

关注操作：lua令牌桶限流，并将时间写入outbox表

@Override
@Transactional
public boolean follow(long fromUserId, long toUserId) {
// Lua 脚本令牌桶限流–同一个用户1s内最多发起多少关注
Long ok = redis.execute(tokenScript, List.of("rl:follow:" + fromUserId), "100", "1");
if (ok == 0L) {
return false;
}

//调用线程安全的随机数生成工具，生成一个超大随机数，作为数据库主键ID
long id = ThreadLocalRandom.current().nextLong(Long.MAX_VALUE);
int inserted = mapper.insertFollowing(id, fromUserId, toUserId, 1);

if (inserted > 0) {
try {
Long outId = ThreadLocalRandom.current().nextLong(Long.MAX_VALUE);
String payload = objectMapper.writeValueAsString(new RelationEvent("FollowCreated", fromUserId, toUserId, id));
outboxMapper.insert(outId, "following", id, "FollowCreated", payload);
} catch (Exception ignored) {}

return true;
}
return false;
}

取关操作：写入outbox事件

/**
* 取消关注操作，并写入 Outbox 事件。
* @param fromUserId 发起取消关注的用户ID
* @param toUserId 被取消关注的用户ID
* @return 是否取消成功
*/
@Override
@Transactional
public boolean unfollow(long fromUserId, long toUserId) {
int updated = mapper.cancelFollowing(fromUserId, toUserId);
if (updated > 0) {
try {
Long outId = ThreadLocalRandom.current().nextLong(Long.MAX_VALUE);
String payload = objectMapper.writeValueAsString(new RelationEvent("FollowCanceled", fromUserId, toUserId, null));
outboxMapper.insert(outId, "following", null, "FollowCanceled", payload);
} catch (Exception ignored) {}
return true;
}
return false;
}

查看是否是第一次，防止消息重复执行业务

String dk = "dedup:rel:" + evt.type() + ":" + evt.fromUserId() + ":" + evt.toUserId() + ":" + (evt.id() == null ? "0" : String.valueOf(evt.id()));
Boolean first = redis.opsForValue().setIfAbsent(dk, "1", Duration.ofMinutes(10));

四、多级缓存架构：支撑高并发列表查询

this.flwsTopCache = Caffeine.newBuilder().maximumSize(1000).expireAfterWrite(Duration.ofMinutes(10)).build();
this.fansTopCache = Caffeine.newBuilder().maximumSize(1000).expireAfterWrite(Duration.ofMinutes(10)).build();

public void process(RelationEvent evt) {
String dk = "dedup:rel:" + evt.type() + ":" + evt.fromUserId() + ":" + evt.toUserId() + ":" + (evt.id() == null ? "0" : String.valueOf(evt.id()));
Boolean first = redis.opsForValue().setIfAbsent(dk, "1", Duration.ofMinutes(10));

// 非首次（存在去重键）直接返回，保证消息幂等，防止网络波动，消费者重启，重平衡，重试机制
if (first == null || !first) {
return;
}
if ("FollowCreated".equals(evt.type())) {
// 异步插入粉丝表
mapper.insertFollower(evt.id(), evt.toUserId(), evt.fromUserId(), 1);
long now = System.currentTimeMillis();

// 更新关注表与粉丝表缓存：ZSet 按时间分数维护最近项，设置短 TTL 减少陈旧数据
redis.opsForZSet().add("uf:flws:" + evt.fromUserId(), String.valueOf(evt.toUserId()), now);
redis.opsForZSet().add("uf:fans:" + evt.toUserId(), String.valueOf(evt.fromUserId()), now);
//使用redis.expire设置缓存过期时间
redis.expire("uf:flws:" + evt.fromUserId(), Duration.ofHours(2));
redis.expire("uf:fans:" + evt.toUserId(), Duration.ofHours(2));

// 更新关注数与粉丝数
userCounterService.incrementFollowings(evt.fromUserId(), 1);
userCounterService.incrementFollowers(evt.toUserId(), 1);
} else if ("FollowCanceled".equals(evt.type())) {
mapper.cancelFollower(evt.toUserId(), evt.fromUserId());

// 更新关注表与粉丝表缓存：移除 ZSet 项并刷新 TTL
redis.opsForZSet().remove("uf:flws:" + evt.fromUserId(), String.valueOf(evt.toUserId()));
redis.opsForZSet().remove("uf:fans:" + evt.toUserId(), String.valueOf(evt.fromUserId()));
redis.expire("uf:flws:" + evt.fromUserId(), Duration.ofHours(2));
redis.expire("uf:fans:" + evt.toUserId(), Duration.ofHours(2));

// 更新关注数与粉丝数
userCounterService.incrementFollowings(evt.fromUserId(), -1);
userCounterService.incrementFollowers(evt.toUserId(), -1);
}
}

五、数据持久层设计

六、异步事件通信核心链路

创建Canalkafka桥接器，拉去未处理消息

// 创建 Canal 单实例连接器并建立连接
connector = CanalConnectors.newSingleConnector(new InetSocketAddress(host, port), destination, username, password);
log.info("Canal connecting to {}:{} dest={} user={} filter={}", host, port, destination, username, filter);
connector.connect();
// 订阅过滤表达式，仅拉取关心的表（如 outbox）
connector.subscribe(filter);
// 回滚到上次确认位点，保证一致性处理
connector.rollback();
log.info("Canal connected and subscribed: host={} port={} dest={} filter={} batchSize={} intervalMs={}ms", host, port, destination, filter, batchSize, intervalMs);
while (running) {
// 拉取一批未确认消息（不自动 ack）
Message message = connector.getWithoutAck(batchSize);

try {
//把kafka消息转换成多行数据表
List<JsonNode> rows = OutboxMessageUtil.extractRows(objectMapper, message);
if (rows.isEmpty()) {
ack.acknowledge();
return;
}
for (JsonNode row : rows) {
JsonNode payloadNode = row.get("payload");
if (payloadNode == null) {
continue;
}
//将json转换成要读的字符串
RelationEvent evt = objectMapper.readValue(payloadNode.asText(), RelationEvent.class);
processor.process(evt);
}
ack.acknowledge();
} catch (Exception ignored) {}
}

写入Redis，写入follower表

七、用户计数精准保障体系

八、业务核心问题解决方案

九、架构优势总结