前言今天开始整理项目中的用户关系系统。用户关系模块主要负责关注用户取消关注查询是否关注、是否被关注、是否互关查询关注列表和粉丝列表维护关注数、粉丝数异步更新缓存和粉丝表这部分没有简单地在一个接口里同时更新所有数据而是采用了“一主多从 事件驱动”的模型。其中following 表主事实表 follower 表粉丝视角的伪从表 用户计数 SDS计数伪从 Redis ZSet列表缓存伪从关注动作发生时只要求following主表和outbox事件表在同一个 MySQL 事务中成功。后面的 follower 表、计数、缓存都通过事件异步更新。这一篇先看最核心的写入链路关注 / 取关接口如何落到 following 主表并在同一事务中写入 Outbox 事件。一、用户关系模块整体结构用户关系相关代码主要在src/main/java/com/tongji/relation ├── api │ └── RelationController.java ├── service │ ├── RelationService.java │ └── impl/RelationServiceImpl.java ├── mapper │ └── RelationMapper.java ├── event │ └── RelationEvent.java └── outbox └── OutboxMapper.java本篇关注这几个文件RelationController.java RelationServiceImpl.java RelationMapper.xml OutboxMapper.xml RelationEvent.java整体写入流程如下用户点击关注 ↓ Controller 获取当前登录用户 ID ↓ Service 执行关注限流 ↓ 写 following 主表 ↓ 同一事务写 outbox 事件 ↓ 返回关注结果这里最重要的是关注主事实和事件投递意图在一个事务中完成。二、数据库表设计1. following 主表following表表示“我关注了谁”。-- db/schema.sqlCREATETABLEIFNOTEXISTSfollowing(idBIGINTUNSIGNEDNOTNULL,from_user_idBIGINTUNSIGNEDNOTNULL,to_user_idBIGINTUNSIGNEDNOTNULL,rel_statusTINYINTNOTNULLDEFAULT1,created_atDATETIME(3)NOTNULL,updated_atDATETIME(3)NOTNULL,PRIMARYKEY(id),UNIQUEKEYuk_from_to(from_user_id,to_user_id),KEYidx_from_created(from_user_id,created_at,to_user_id,rel_status),KEYidx_to(to_user_id,from_user_id,rel_status));字段说明字段含义from_user_id发起关注的人to_user_id被关注的人rel_status关系状态1表示关注中0表示已取消created_at创建时间updated_at更新时间这里有一个唯一索引UNIQUEKEYuk_from_to(from_user_id,to_user_id)它保证同一对用户之间不会插入多条关注关系。2. outbox 事件表Outbox 表用于保存领域事件。CREATETABLEIFNOTEXISTSoutbox(idBIGINTUNSIGNEDNOTNULL,aggregate_typeVARCHAR(64)NOTNULL,aggregate_idBIGINTUNSIGNEDNULL,typeVARCHAR(64)NOTNULL,payload JSONNOTNULL,created_atTIMESTAMP(3)NOTNULLDEFAULTCURRENT_TIMESTAMP(3),PRIMARYKEY(id),KEYix_outbox_agg(aggregate_type,aggregate_id),KEYix_outbox_ct(created_at));关注成功时会写入一条FollowCreated事件取消关注时会写入一条FollowCanceled事件。三、关注接口 Controller 层// src/main/java/com/tongji/relation/api/RelationController.javaRestControllerRequestMapping(/api/v1/relation)publicclassRelationController{privatefinalRelationServicerelationService;privatefinalJwtServicejwtService;PostMapping(/follow)publicbooleanfollow(RequestParam(toUserId)longtoUserId,AuthenticationPrincipalJwtjwt){longuidjwtService.extractUserId(jwt);returnrelationService.follow(uid,toUserId);}PostMapping(/unfollow)publicbooleanunfollow(RequestParam(toUserId)longtoUserId,AuthenticationPrincipalJwtjwt){longuidjwtService.extractUserId(jwt);returnrelationService.unfollow(uid,toUserId);}}接口路径POST /api/v1/relation/follow?toUserId123 POST /api/v1/relation/unfollow?toUserId123Controller 层主要做两件事通过AuthenticationPrincipal Jwt获取当前登录用户。把当前用户 ID 和目标用户 ID 交给 Service 层处理。这里的fromUserId不由前端传而是从 JWT 中解析这样可以避免用户伪造关注发起者。四、关注 Service 层实现1. 关注流程// src/main/java/com/tongji/relation/service/impl/RelationServiceImpl.javaOverrideTransactionalpublicbooleanfollow(longfromUserId,longtoUserId){Longokredis.execute(tokenScript,List.of(rl:follow:fromUserId),100,1);if(ok0L){returnfalse;}longidThreadLocalRandom.current().nextLong(Long.MAX_VALUE);intinsertedmapper.insertFollowing(id,fromUserId,toUserId,1);if(inserted0){try{LongoutIdThreadLocalRandom.current().nextLong(Long.MAX_VALUE);StringpayloadobjectMapper.writeValueAsString(newRelationEvent(FollowCreated,fromUserId,toUserId,id));outboxMapper.insert(outId,following,id,FollowCreated,payload);}catch(Exceptionignored){}returntrue;}returnfalse;}这段代码可以拆成三步看。第一步关注限流redis.execute(tokenScript,List.of(rl:follow:fromUserId),100,1);每个用户都有自己的限流 Keyrl:follow:{fromUserId}第二步写入following主表mapper.insertFollowing(id,fromUserId,toUserId,1);第三步写入 Outbox 事件outboxMapper.insert(outId,following,id,FollowCreated,payload);由于方法上有Transactional所以 following 主表和 outbox 表处在同一个事务里。这就保证了关注关系写成功事件一定会落库 关注关系失败事件也不会落库。2. 取消关注流程OverrideTransactionalpublicbooleanunfollow(longfromUserId,longtoUserId){intupdatedmapper.cancelFollowing(fromUserId,toUserId);if(updated0){try{LongoutIdThreadLocalRandom.current().nextLong(Long.MAX_VALUE);StringpayloadobjectMapper.writeValueAsString(newRelationEvent(FollowCanceled,fromUserId,toUserId,null));outboxMapper.insert(outId,following,null,FollowCanceled,payload);}catch(Exceptionignored){}returntrue;}returnfalse;}取消关注也是同样的思路更新 following 主表 rel_status0 ↓ 写入 FollowCanceled 事件这里没有直接删除关系而是逻辑取消。这样做的好处是可以保留关系变更痕迹也方便后续重新关注时复用唯一键。五、令牌桶限流 Lua关注操作前会先执行 Redis Lua 令牌桶。-- src/main/java/com/tongji/relation/service/impl/RelationServiceImpl.javalocalkeyKEYS[1]localcapacitytonumber(ARGV[1])localratetonumber(ARGV[2])localnowredis.call(TIME)[1]locallastredis.call(HGET,key,last)localtokensredis.call(HGET,key,tokens)ifnotlastthenlastnow tokenscapacityendlocalelapsedtonumber(now)-tonumber(last)localaddelapsed*rate tokensmath.min(capacity,tonumber(tokens)add)iftokens1thenredis.call(HSET,key,last,now)redis.call(HSET,key,tokens,tokens)return0endtokenstokens-1redis.call(HSET,key,last,now)redis.call(HSET,key,tokens,tokens)redis.call(PEXPIRE,key,60000)return1调用参数是List.of(rl:follow:fromUserId),100,1也就是说桶容量100补充速率1 个 token / 秒每次关注消耗 1 个 token这样可以防止短时间内恶意批量关注。六、Mapper 层实现1. 插入关注关系!-- src/main/resources/mapper/RelationMapper.xml --insertidinsertFollowingINSERT INTO following ( id, from_user_id, to_user_id, rel_status, created_at, updated_at ) VALUES ( #{id}, #{fromUserId}, #{toUserId}, #{relStatus}, NOW(3), NOW(3) ) ON DUPLICATE KEY UPDATE rel_status VALUES(rel_status), updated_at VALUES(updated_at)/insert这里用了ONDUPLICATEKEYUPDATE因为(from_user_id, to_user_id)有唯一索引。如果之前已经关注过又取消了再次关注时不会插入重复数据而是把rel_status更新回1。2. 取消关注updateidcancelFollowingUPDATE following SET rel_status 0, updated_at NOW(3) WHERE from_user_id #{fromUserId} AND to_user_id #{toUserId}/update取消关注只修改状态不删除记录。七、Outbox 事件结构关系事件定义如下// src/main/java/com/tongji/relation/event/RelationEvent.javapublicrecordRelationEvent(Stringtype,LongfromUserId,LongtoUserId,Longid){}关注成功事件示例{type:FollowCreated,fromUserId:100,toUserId:200,id:123456}取消关注事件示例{type:FollowCanceled,fromUserId:100,toUserId:200,id:null}八、Outbox Mapper 实现!-- src/main/resources/mapper/OutboxMapper.xml --insertidinsertINSERT INTO outbox ( id, aggregate_type, aggregate_id, type, payload, created_at ) VALUES ( #{id}, #{aggregateType}, #{aggregateId}, #{type}, #{payload}, NOW(3) )/insert在关注系统中Outbox 的aggregate_type是following事件类型是FollowCreated FollowCanceledpayload 中保存具体事件内容。九、为什么要引入 Outbox如果关注接口中直接这样写写 following 表 发送 Kafka 消息就会遇到一个经典问题数据库写成功了但消息发送失败怎么办 消息发送成功了但数据库事务回滚怎么办Outbox 模式解决的就是这个问题。它把“发送消息”变成“写一条事件记录”写 following 主表 写 outbox 事件表 提交同一个 MySQL 事务后续再由 Canal 订阅 outbox 表 binlog把事件异步推送到 Kafka。这样可以保证主事实和事件意图一致。十、知识点总结1. following 为什么是主表因为它表示“我关注了谁”是关注行为的直接事实来源。后面的 follower 表、缓存、计数都是可以重建的派生数据。2. 为什么取消关注不直接删除逻辑取消可以保留关系记录也方便后续重新关注时使用唯一键做幂等更新。3. Outbox 模式解决什么问题解决数据库事务和消息发送之间的一致性问题。主表和事件表同事务成功后续再通过 Canal/Kafka 异步分发。4. 令牌桶有什么作用关注属于容易被滥用的行为令牌桶可以按用户维度限制短时间高频关注。总结这一篇主要整理了用户关系系统的主写入流程。关注和取关并不是一次性同步更新所有数据而是只保证following主表和outbox表在同一事务内完成。这样following成为唯一主事实Outbox 事件成为后续异步同步的入口。这就是“一主多从 事件驱动”模型的第一步主表强一致派生视图最终一致。
【知识获取与分享社区项目 | 项目日记第 7 天】关注取关实现：following 主表 + Outbox 同事务
前言今天开始整理项目中的用户关系系统。用户关系模块主要负责关注用户取消关注查询是否关注、是否被关注、是否互关查询关注列表和粉丝列表维护关注数、粉丝数异步更新缓存和粉丝表这部分没有简单地在一个接口里同时更新所有数据而是采用了“一主多从 事件驱动”的模型。其中following 表主事实表 follower 表粉丝视角的伪从表 用户计数 SDS计数伪从 Redis ZSet列表缓存伪从关注动作发生时只要求following主表和outbox事件表在同一个 MySQL 事务中成功。后面的 follower 表、计数、缓存都通过事件异步更新。这一篇先看最核心的写入链路关注 / 取关接口如何落到 following 主表并在同一事务中写入 Outbox 事件。一、用户关系模块整体结构用户关系相关代码主要在src/main/java/com/tongji/relation ├── api │ └── RelationController.java ├── service │ ├── RelationService.java │ └── impl/RelationServiceImpl.java ├── mapper │ └── RelationMapper.java ├── event │ └── RelationEvent.java └── outbox └── OutboxMapper.java本篇关注这几个文件RelationController.java RelationServiceImpl.java RelationMapper.xml OutboxMapper.xml RelationEvent.java整体写入流程如下用户点击关注 ↓ Controller 获取当前登录用户 ID ↓ Service 执行关注限流 ↓ 写 following 主表 ↓ 同一事务写 outbox 事件 ↓ 返回关注结果这里最重要的是关注主事实和事件投递意图在一个事务中完成。二、数据库表设计1. following 主表following表表示“我关注了谁”。-- db/schema.sqlCREATETABLEIFNOTEXISTSfollowing(idBIGINTUNSIGNEDNOTNULL,from_user_idBIGINTUNSIGNEDNOTNULL,to_user_idBIGINTUNSIGNEDNOTNULL,rel_statusTINYINTNOTNULLDEFAULT1,created_atDATETIME(3)NOTNULL,updated_atDATETIME(3)NOTNULL,PRIMARYKEY(id),UNIQUEKEYuk_from_to(from_user_id,to_user_id),KEYidx_from_created(from_user_id,created_at,to_user_id,rel_status),KEYidx_to(to_user_id,from_user_id,rel_status));字段说明字段含义from_user_id发起关注的人to_user_id被关注的人rel_status关系状态1表示关注中0表示已取消created_at创建时间updated_at更新时间这里有一个唯一索引UNIQUEKEYuk_from_to(from_user_id,to_user_id)它保证同一对用户之间不会插入多条关注关系。2. outbox 事件表Outbox 表用于保存领域事件。CREATETABLEIFNOTEXISTSoutbox(idBIGINTUNSIGNEDNOTNULL,aggregate_typeVARCHAR(64)NOTNULL,aggregate_idBIGINTUNSIGNEDNULL,typeVARCHAR(64)NOTNULL,payload JSONNOTNULL,created_atTIMESTAMP(3)NOTNULLDEFAULTCURRENT_TIMESTAMP(3),PRIMARYKEY(id),KEYix_outbox_agg(aggregate_type,aggregate_id),KEYix_outbox_ct(created_at));关注成功时会写入一条FollowCreated事件取消关注时会写入一条FollowCanceled事件。三、关注接口 Controller 层// src/main/java/com/tongji/relation/api/RelationController.javaRestControllerRequestMapping(/api/v1/relation)publicclassRelationController{privatefinalRelationServicerelationService;privatefinalJwtServicejwtService;PostMapping(/follow)publicbooleanfollow(RequestParam(toUserId)longtoUserId,AuthenticationPrincipalJwtjwt){longuidjwtService.extractUserId(jwt);returnrelationService.follow(uid,toUserId);}PostMapping(/unfollow)publicbooleanunfollow(RequestParam(toUserId)longtoUserId,AuthenticationPrincipalJwtjwt){longuidjwtService.extractUserId(jwt);returnrelationService.unfollow(uid,toUserId);}}接口路径POST /api/v1/relation/follow?toUserId123 POST /api/v1/relation/unfollow?toUserId123Controller 层主要做两件事通过AuthenticationPrincipal Jwt获取当前登录用户。把当前用户 ID 和目标用户 ID 交给 Service 层处理。这里的fromUserId不由前端传而是从 JWT 中解析这样可以避免用户伪造关注发起者。四、关注 Service 层实现1. 关注流程// src/main/java/com/tongji/relation/service/impl/RelationServiceImpl.javaOverrideTransactionalpublicbooleanfollow(longfromUserId,longtoUserId){Longokredis.execute(tokenScript,List.of(rl:follow:fromUserId),100,1);if(ok0L){returnfalse;}longidThreadLocalRandom.current().nextLong(Long.MAX_VALUE);intinsertedmapper.insertFollowing(id,fromUserId,toUserId,1);if(inserted0){try{LongoutIdThreadLocalRandom.current().nextLong(Long.MAX_VALUE);StringpayloadobjectMapper.writeValueAsString(newRelationEvent(FollowCreated,fromUserId,toUserId,id));outboxMapper.insert(outId,following,id,FollowCreated,payload);}catch(Exceptionignored){}returntrue;}returnfalse;}这段代码可以拆成三步看。第一步关注限流redis.execute(tokenScript,List.of(rl:follow:fromUserId),100,1);每个用户都有自己的限流 Keyrl:follow:{fromUserId}第二步写入following主表mapper.insertFollowing(id,fromUserId,toUserId,1);第三步写入 Outbox 事件outboxMapper.insert(outId,following,id,FollowCreated,payload);由于方法上有Transactional所以 following 主表和 outbox 表处在同一个事务里。这就保证了关注关系写成功事件一定会落库 关注关系失败事件也不会落库。2. 取消关注流程OverrideTransactionalpublicbooleanunfollow(longfromUserId,longtoUserId){intupdatedmapper.cancelFollowing(fromUserId,toUserId);if(updated0){try{LongoutIdThreadLocalRandom.current().nextLong(Long.MAX_VALUE);StringpayloadobjectMapper.writeValueAsString(newRelationEvent(FollowCanceled,fromUserId,toUserId,null));outboxMapper.insert(outId,following,null,FollowCanceled,payload);}catch(Exceptionignored){}returntrue;}returnfalse;}取消关注也是同样的思路更新 following 主表 rel_status0 ↓ 写入 FollowCanceled 事件这里没有直接删除关系而是逻辑取消。这样做的好处是可以保留关系变更痕迹也方便后续重新关注时复用唯一键。五、令牌桶限流 Lua关注操作前会先执行 Redis Lua 令牌桶。-- src/main/java/com/tongji/relation/service/impl/RelationServiceImpl.javalocalkeyKEYS[1]localcapacitytonumber(ARGV[1])localratetonumber(ARGV[2])localnowredis.call(TIME)[1]locallastredis.call(HGET,key,last)localtokensredis.call(HGET,key,tokens)ifnotlastthenlastnow tokenscapacityendlocalelapsedtonumber(now)-tonumber(last)localaddelapsed*rate tokensmath.min(capacity,tonumber(tokens)add)iftokens1thenredis.call(HSET,key,last,now)redis.call(HSET,key,tokens,tokens)return0endtokenstokens-1redis.call(HSET,key,last,now)redis.call(HSET,key,tokens,tokens)redis.call(PEXPIRE,key,60000)return1调用参数是List.of(rl:follow:fromUserId),100,1也就是说桶容量100补充速率1 个 token / 秒每次关注消耗 1 个 token这样可以防止短时间内恶意批量关注。六、Mapper 层实现1. 插入关注关系!-- src/main/resources/mapper/RelationMapper.xml --insertidinsertFollowingINSERT INTO following ( id, from_user_id, to_user_id, rel_status, created_at, updated_at ) VALUES ( #{id}, #{fromUserId}, #{toUserId}, #{relStatus}, NOW(3), NOW(3) ) ON DUPLICATE KEY UPDATE rel_status VALUES(rel_status), updated_at VALUES(updated_at)/insert这里用了ONDUPLICATEKEYUPDATE因为(from_user_id, to_user_id)有唯一索引。如果之前已经关注过又取消了再次关注时不会插入重复数据而是把rel_status更新回1。2. 取消关注updateidcancelFollowingUPDATE following SET rel_status 0, updated_at NOW(3) WHERE from_user_id #{fromUserId} AND to_user_id #{toUserId}/update取消关注只修改状态不删除记录。七、Outbox 事件结构关系事件定义如下// src/main/java/com/tongji/relation/event/RelationEvent.javapublicrecordRelationEvent(Stringtype,LongfromUserId,LongtoUserId,Longid){}关注成功事件示例{type:FollowCreated,fromUserId:100,toUserId:200,id:123456}取消关注事件示例{type:FollowCanceled,fromUserId:100,toUserId:200,id:null}八、Outbox Mapper 实现!-- src/main/resources/mapper/OutboxMapper.xml --insertidinsertINSERT INTO outbox ( id, aggregate_type, aggregate_id, type, payload, created_at ) VALUES ( #{id}, #{aggregateType}, #{aggregateId}, #{type}, #{payload}, NOW(3) )/insert在关注系统中Outbox 的aggregate_type是following事件类型是FollowCreated FollowCanceledpayload 中保存具体事件内容。九、为什么要引入 Outbox如果关注接口中直接这样写写 following 表 发送 Kafka 消息就会遇到一个经典问题数据库写成功了但消息发送失败怎么办 消息发送成功了但数据库事务回滚怎么办Outbox 模式解决的就是这个问题。它把“发送消息”变成“写一条事件记录”写 following 主表 写 outbox 事件表 提交同一个 MySQL 事务后续再由 Canal 订阅 outbox 表 binlog把事件异步推送到 Kafka。这样可以保证主事实和事件意图一致。十、知识点总结1. following 为什么是主表因为它表示“我关注了谁”是关注行为的直接事实来源。后面的 follower 表、缓存、计数都是可以重建的派生数据。2. 为什么取消关注不直接删除逻辑取消可以保留关系记录也方便后续重新关注时使用唯一键做幂等更新。3. Outbox 模式解决什么问题解决数据库事务和消息发送之间的一致性问题。主表和事件表同事务成功后续再通过 Canal/Kafka 异步分发。4. 令牌桶有什么作用关注属于容易被滥用的行为令牌桶可以按用户维度限制短时间高频关注。总结这一篇主要整理了用户关系系统的主写入流程。关注和取关并不是一次性同步更新所有数据而是只保证following主表和outbox表在同一事务内完成。这样following成为唯一主事实Outbox 事件成为后续异步同步的入口。这就是“一主多从 事件驱动”模型的第一步主表强一致派生视图最终一致。
