当热心用户或者老师给学生回答了问题以后,所有学员可以给自己心仪的回答点赞,点赞越高,排名也越靠前。
1.1.业务需求
首先我们来分析整理一下点赞业务的需求,一个通用点赞系统需要满足下列特性:
1.2.实现思路
要保证安全,避免重复点赞,我们就必须保存每一次点赞记录。只有这样在下次用户点赞时我们才能查询数据,判断是否是重复点赞。同时,因为业务方经常需要根据点赞数量排序,因此每个业务的点赞数量也需要记录下来。
综上,点赞的基本思路如下:
点赞系统可以在点赞数变更时,通过MQ通知业务方,这样业务方就可以更新自己的点赞数量了。并且还避免了点赞系统与业务方的耦合。
于是,实现思路变成了这样:
2.数据结构
点赞的数据结构分两部分,一是点赞记录,二是与业务关联的点赞数。
点赞数自然是与具体业务表关联在一起记录,比如互动问答的点赞,自然是在问答表中记录点赞数。学员笔记点赞,自然是在笔记表中记录点赞数。
在之前实现互动问答的时候,我们已经给回答表设计了点赞数字段了:
因此,本节我们只需要实现点赞记录的表结构设计即可。
2.1.ER图
点赞记录本质就是记录谁给什么内容点了赞,所以核心属性包括:
- 点赞目标id
- 点赞人id
不过点赞的内容多种多样,为了加以区分,我们还需要把点赞内的类型记录下来: - 点赞对象类型(为了通用性)
当然还有点赞时间,综上对应的数据库ER图如下:
2.2.表结构
由于点赞系统是独立于其它业务的,这里我们需要创建一个新的数据库:tj_remark
CREATE DATABASE tj_remark CHARACTER SET 'utf8mb4';
然后在ER图基础上,加上一些通用属性,点赞记录表结构如下:
CREATE TABLE IF NOT EXISTS `liked_record` (
`id` bigint NOT NULL AUTO_INCREMENT COMMENT '主键id',
`user_id` bigint NOT NULL COMMENT '用户id',
`biz_id` bigint NOT NULL COMMENT '点赞的业务id',
`biz_type` VARCHAR(16) NOT NULL COMMENT '点赞的业务类型',
`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 `idx_biz_user` (`biz_id`,`user_id`)
) ENGINE=InnoDB AUTO_INCREMENT=8 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='点赞记录表';
3.实现点赞功能
从表面来看,点赞功能要实现的接口就是一个点赞接口。不过仔细观察所有的点赞页面,你会发现点赞按钮有灰色和点亮两种状态。
也就是说我们还需要实现查询用户点赞状态的接口,这样前端才能根据点赞状态渲染不同效果。因此我们要实现的接口包括:
- 点赞/取消点赞
- 根据多个业务id批量查询用户是否点赞多个业务
3.1.点赞或取消点赞
3.1.1.接口信息
当用户点击点赞按钮的时候,第一次点击是点赞,按钮会高亮;第二次点击是取消,点赞按钮变灰:
从后台实现来看,点赞就是新增一条点赞记录,取消就是删除这条记录。为了方便前端交互,这两个合并为一个接口即可。
因此,请求参数首先要包含点赞有关的数据,并且要标记是点赞还是取消:
- 点赞的目标业务id:bizId
- 谁在点赞(就是登陆用户,可以不用提交)
- 点赞还是取消
除此以外,我们之前说过,在问答、笔记等功能中都会出现点赞功能,所以点赞必须具备通用性。因此还需要在提交一个参数标记点赞的类型:
- 点赞目标的类型
返回值有两种设计:
- 方案一:无返回值,200就是成功,页面直接把点赞数+1展示给用户即可
- 方案二:返回点赞数量,页面渲染
这里推荐使用方案一,因为每次统计点赞数量也有很大的性能消耗。
综上,按照Restful风格设计,接口信息如下:
3.1.2.实体
请求参数需要定义一个DTO实体类来接收
@Data
@ApiModel(description = "点赞记录表单实体")
public class LikeRecordFormDTO {
@ApiModelProperty("点赞业务id")
@NotNull(message = "业务id不能为空")
private Long bizId;
@ApiModelProperty("点赞业务类型")
@NotNull(message = "业务类型不能为空")
private String bizType;
@ApiModelProperty("是否点赞,true:点赞;false:取消点赞")
@NotNull(message = "是否点赞不能为空")
private Boolean liked;
}
3.1.3.代码实现
首先是tj-remark的com.tianji.remark.controller.LikedRecordController:
/**
* <p>
* 点赞记录表 控制器
* </p>
*/
@RestController
@RequiredArgsConstructor
@RequestMapping("/likes")
@Api(tags = "点赞业务相关接口")
public class LikedRecordController {
private final ILikedRecordService likedRecordService;
@PostMapping
@ApiOperation("点赞或取消点赞")
public void addLikeRecord(@Valid @RequestBody LikeRecordFormDTO recordDTO) {
likedRecordService.addLikeRecord(recordDTO);
}
}
然后是tj-remark的com.tianji.remark.service.ILikedRecordService:
public interface ILikedRecordService extends IService<LikedRecord> {
void addLikeRecord(LikeRecordFormDTO recordFormDTO);
}
实现类:
@Service
public class LikedRecordServiceImpl extends ServiceImpl<LikedRecordMapper, LikedRecord> implements ILikedRecordService {
@Override
public void addLikeRecord(LikeRecordFormDTO recordFormDTO) {
// TODO 实现点赞或取消点赞
}
}
3.1.4.业务流程
我们先梳理一下点赞业务的几点需求:
- 点赞就新增一条点赞记录,取消点赞就删除记录
- 用户不能重复点赞
- 点赞数由具体的业务方保存,需要通知业务方更新点赞数
由于业务方的类型很多,比如互动问答、笔记、课程等。所以通知方式必须是低耦合的,这里建议使用MQ来实现。
当点赞或取消点赞后,点赞数发生变化,我们就发送MQ通知。整体业务流程如图:
需要注意的是,由于每次点赞的业务类型不同,所以没有必要通知到所有业务方,而是仅仅通知与当前点赞业务关联的业务方即可。
在RabbitMQ中,利用TOPIC类型的交换机**,结合不同的RoutingKey(路由key),可以实现通知对象的变化**。我们需要让不同的业务方监听不同的RoutingKey,然后发送通知时根据点赞类型不同,发送不同RoutingKey:
在tj-common中,我们已经定义了MQ的常量:
interface Exchange{
/*点赞记录有关的交换机*/
....
String LIKE_RECORD_EXCHANGE = "like.record.topic";
....
}
并且定义了点赞有关的Exchange和RoutingKey常量:
interface Key{
.....
/*点赞的RoutingKey*/
String LIKED_TIMES_KEY_TEMPLATE = "{}.times.changed";
/*问答*/
String QA_LIKED_TIMES_KEY = "QA.times.changed";
/*笔记*/
String NOTE_LIKED_TIMES_KEY = "NOTE.times.changed";
......
}
其中的RoutingKey只是一个模板,其中{}部分是占位符,不同业务类型就填写不同的具体值。
3.1.5.实现完整业务
首先我们需要定义一个MQ通知的消息体,由于这个消息体会在各个相关微服务中使用,需要定义到公用的模块中,这里我们定义到tj-api模块:
@Data
@NoArgsConstructor
@AllArgsConstructor
public class LikedTimesDTO {
/**
* 点赞的业务id
*/
private Long bizId;
/**
* 总的点赞次数 (不同的微服务业务 对应自己的点赞次数)
*/
private Integer likedTimes;
}
完整实现类:
@RequiredArgsConstructor
public class LikedRecordServiceImpl extends ServiceImpl<LikedRecordMapper, LikedRecord> implements ILikedRecordService {
private final RabbitMqHelper mqHelper;
@Override
public void addLikeRecord(LikeRecordFormDTO recordDTO) {
// 1.基于前端的参数,判断是执行点赞还是取消点赞(true:点赞;false:取消点赞)
boolean success = recordDTO.getLiked() ?
like(recordDTO) : unlike(recordDTO);
//true执行like点赞方法,否则执行unlike取消点赞
// 2.判断是否执行成功,如果失败,则直接结束
if (!success) {
return;
}
// 3.如果执行成功,统计点赞总数(查询)
Integer likedTimes = lambdaQuery()
//点赞业务id
.eq(LikedRecord::getBizId, recordDTO.getBizId())
.count();
// 4.发送MQ通知 (通知消息:业务对应的点赞总数)
mqHelper.send(
//点赞记录有关的交换机
LIKE_RECORD_EXCHANGE,
//点赞的RoutingKey
StringUtils.format(LIKED_TIMES_KEY_TEMPLATE, recordDTO.getBizType()),
//消息:点赞业务id和点赞总数
LikedTimesDTO.of(recordDTO.getBizId(), likedTimes));
}
private boolean unlike(LikeRecordFormDTO recordDTO) {
return remove(new QueryWrapper<LikedRecord>().lambda()
//查询得到一个LikedRecord实体类对象 然后调用remove删除
.eq(LikedRecord::getUserId,
UserContext.getUser())//用户id
.eq(LikedRecord::getBizId,
recordDTO.getBizId()));//点赞业务id
}
private boolean like(LikeRecordFormDTO recordDTO) {
Long userId = UserContext.getUser();
// 1.查询点赞记录
Integer count = lambdaQuery()
.eq(LikedRecord::getUserId, userId)//当前用户id
.eq(LikedRecord::getBizId, recordDTO.getBizId())//点赞业务id
.count();
// 2.判断是否存在,如果已经存在,直接结束(因为这个是点赞方法)
if (count > 0) {
return false;
}
// 3.如果不存在,直接新增点赞
LikedRecord r = new LikedRecord();
r.setUserId(userId);
r.setBizId(recordDTO.getBizId());//点赞的业务id
r.setBizType(recordDTO.getBizType());//点赞的业务类型
save(r);
return true;
}
}
3.2.批量查询点赞状态
由于这个接口是供其它微服务调用,实现完成接口后,还需要定义对应的FeignClient
3.2.1.接口信息
这里是查询多个业务的点赞状态,因此请求参数自然是业务id的集合。由于是查询当前用户的点赞状态,因此无需传递用户信息。
经过筛选判断后,我们把点赞过的业务id集合返回即可。
3.3.2.代码
@GetMapping("list")
@ApiOperation("查询指定业务id的点赞状态")
public Set<Long> isBizLiked(@RequestParam("bizIds") List<Long> bizIds){
return likedRecordService.isBizLiked(bizIds);
}
service接口:
/**
* <p>
* 点赞记录表 服务类
* </p>
*/
public interface ILikedRecordService extends IService<LikedRecord> {
void addLikeRecord(LikeRecordFormDTO recordDTO);
Set<Long> isBizLiked(List<Long> bizIds);
}
实现类:注意接口说明(了解需求是什么)
public Set<Long> isBizLiked(List<Long> bizIds) {
// 1.获取登录用户id
Long userId = UserContext.getUser();
// 2.查询点赞状态
List<LikedRecord> list = lambdaQuery()
//业务id集合
.in(LikedRecord::getBizId, bizIds)
//当前用户id
.eq(LikedRecord::getUserId, userId)
.list(); //当前用户给哪些业务点了赞
// 3.返回结果
return list.stream().map(LikedRecord::getBizId).collect(Collectors.toSet());
}
3.3.3.暴露Feign接口
由于该接口是给其它微服务调用的,所以必须暴露出Feign客户端,并且定义好fallback降级处理:
这里用了sentinel做降级(引入相关依赖)
我们在tj-api模块中定义一个客户端:
其中RemarkClient如下:
@FeignClient(value = "remark-service", fallbackFactory = RemarkClientFallback.class)
public interface RemarkClient {
//对用被调用者的方法(路径方法名参数返回值都要一样)
@GetMapping("/likes/list")
Set<Long> isBizLiked(@RequestParam("bizIds") Iterable<Long> bizIds);
}
对应的fallback逻辑:
实现接口: FallbackFactory<feign接口>
@Slf4j
public class RemarkClientFallback implements FallbackFactory<RemarkClient> {
@Override
public RemarkClient create(Throwable cause) {
log.error("查询remark-service服务异常", cause);
return new RemarkClient() {
@Override
public Set<Long> isBizLiked(Iterable<Long> bizIds) {
return CollUtils.emptySet();
}
};
}
}
由于RemarkClientFallback是定义在tj-api的com.tianji.api包,由于每个微服务扫描包不一致。因此其它引用tj-api的微服务是无法通过扫描包加载到这个类的。
我们需要通过SpringBoot的自动加载机制来加载这些fallback类:
/spring.factories:通过这个配置文件发现config配置类,不然引用api的服务还是要添加扫描注解,扫描配置类。
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
com.tianji.api.config.RequestIdRelayConfiguration, \
com.tianji.api.config.RoleCacheConfig, \
com.tianji.api.config.FallbackConfig, \
com.tianji.api.config.CategoryCacheConfig
由于SpringBoot会在启动时读取/META-INF/spring.factories文件,我们只需要在该文件中指定了要加载
FallbackConig类:
@Configuration
public class FallbackConfig {
@Bean
public LearningClientFallback learningClientFallback(){
return new LearningClientFallback();
}
@Bean
public TradeClientFallback tradeClientFallback(){
return new TradeClientFallback();
}
@Bean
public RemarkClientFallback remarkClientFallback(){
return new RemarkClientFallback();
}
}
这样所有在其中定义的fallback类都会被加载了。
3.4.监听点赞变更的消息
既然点赞后会发送MQ消息通知业务服务,那么每一个有关的业务服务都应该监听点赞数变更的消息,更新本地的点赞数量。
例如互动问答,我们需要在tj-learning服务中定义MQ监听器:
4.点赞功能改进
虽然我们初步实现了点赞功能,不过有一个非常严重的问题,点赞业务包含多次数据库读写操作:
更重要的是,点赞操作波动较大,有可能会在短时间内访问量激增。例如有人非常频繁的点赞、取消点赞。这样就会给数据库带来非常大的压力。
怎么办呢?
4.1.改进思路分析
其实在实现提交学习记录的时候,我们就给大家分析过高并发问题的处理方案。点赞业务与提交播放记录类似,都是高并发写操作。
按照之前我们讲的,高并发写操作常见的优化手段有:
- 优化SQL和代码
- 变同步写为异步写
- 合并写请求
有同学可能会说,我们更新业务方点赞数量的时候,不就是利用MQ异步写来实现的吗?
没错,确实如此,虽然异步写减少了业务执行时间,降低了数据库写频率。不过此处更重要的是利用MQ来解耦。而且数据库的写次数没有减少,压力依然很大。
所以,我们应该像之前播放记录业务一样,采用合并写请求的方案。当然,现在的异步处理也保留,这样就兼顾了异步写、合并写的优势。
需要注意的是,合并写是有使用场景的,必须是对中间的N次写操作不敏感的情况下。点赞业务是否符合这一需求呢?
无论用户中间执行点赞、取消、再点赞、再取消多少次,点赞次数发生了多少次变化,业务方只关注最终的点赞结果即可:
- 用户是否点赞了
- 业务的总点赞次数
因此,点赞功能可以使用合并写方案。最终我们的点赞业务流程变成这样:
合并写请求有两个关键点要考虑:
- 数据如何缓存
- 缓存何时写入数据库
4.1.1.点赞数据缓存
点赞记录中最两个关键信息:
- 用户是否点赞
- 某业务的点赞总次数
这两个信息需要分别记录,也就是说我们需要在Redis中设计两种数据结构分别存储。
4.1.1.1.用户是否点赞
要知道某个用户是否点赞某个业务,就必须记录业务id以及给业务点赞的所有用户id . 由于一个业务可以被很多用户点赞,显然是需要一个集合来记录。而Redis中的集合类型包含四种:
- List
- Set
- SortedSet
- Hash
而要判断用户是否点赞,就是判断存在且唯一。显然,Set集合是最合适的。我们可以用业务id为Key,创建Set集合,将点赞的所有用户保存其中,格式如下:
可以使用Set集合的下列命令完成点赞功能:
# 判断用户是否点赞
SISMEMBER bizId userId
# 点赞,如果返回1则代表点赞成功,返回0则代表点赞失败
SADD bizId userId
# 取消点赞,就是删除一个元素
SREM bizId userId
# 统计点赞总数
SCARD bizId
由于Redis本身具备持久化机制,AOF提供的数据可靠性已经能够满足点赞业务的安全需求,因此我们完全可以用Redis存储来代替数据库的点赞记录。
也就是说,用户的一切点赞行为,以及将来查询点赞状态我们可以都走Redis,不再使用数据库查询。
如果点赞数据非常庞大,达到数百亿,那么该怎办呢?
大多数企业根本达不到这样的规模,如果真的达到也没有关系。这个时候我们可以将Redis与数据库结合。
- 先利用Redis来记录点赞状态,并设置缓存过期时间
- 并且定期的将Redis中的点赞状态持久化到数据库
- 当某个记录点赞时,优先去Redis查询并判断,如果Redis中不存在,再去查询数据库数据并缓存到Redis。
4.1.1.2.点赞次数
由于点赞次数需要在业务方持久化存储到数据库,因此Redis只起到缓存作用即可。
由于需要记录业务id、业务类型、点赞数三个信息:
- 一个业务类型下包含多个业务id
- 每个业务id对应一个点赞数。
因此,我们可以把每一个业务类型作为一组,使用Redis的一个key,然后业务id作为键,点赞数作为值。这样的键值对集合,有两种结构都可以满足: - Hash:传统键值对集合,无序
- SortedSet:基于Hash结构,并且增加了跳表。因此可排序,但更占用内存
如果是从节省内存角度来考虑,Hash结构无疑是最佳的选择;但是考虑到将来我们要从Redis读取点赞数,然后移除(避免重复处理)。为了保证线程安全,查询、移除操作必须具备原子性。而SortedSet则提供了几个移除并获取的功能,天生具备原子性。**并且我们每隔一段时间就会将数据从Redis移除,并不会占用太多内存。**因此,这里我们计划使用SortedSet结构。
格式如下:
当用户对某个业务点赞时,我们统计点赞总数,并将其缓存在Redis中。这样一来在一段时间内,不管有多少用户对该业务点赞(热点业务数据,比如某个微博大V),都只在Redis中修改点赞总数,无需修改数据库。
4.1.2.点赞数据入库
点赞数据写入缓存了,但是这里有一个新的问题:
何时把缓存的点赞数,通过MQ通知到业务方,持久化到业务方的数据库呢?
在之前的提交播放记录业务中,由于播放记录是定期每隔15秒发送一次请求,频率固定。因此我们可以通过接收到播放记录后延迟20秒检测数据变更来确定是否有新数据到达。
但是点赞则不然,用户何时点赞、点赞频率如何完全不确定。因此无法采用延迟检测这样的手段。怎么办?
事实上这也是大多数合并写请求业务面临的问题,而多数情况下,我们只能通过定时任务,定期将缓存的数据持久化到数据库中。
4.1.3.流程图
综上所述,基于Redis做写缓存后,点赞流程如下:
4.2.改造点赞逻辑
需要改造的内容包括:
- tj-remark中所有点赞有关接口
- 点赞接口
- 查询单个点赞状态
- 批量查询点赞状态
- tj-remark处理点赞数据持久化的定时任务
- tj-learning监听点赞数变更消息的业务
由于需要访问Redis,我们提前定义一个常量类,把Redis相关的Key定义为常量:
public interface RedisConstants {
/*给业务点赞的用户集合的KEY前缀,后缀是业务id*/
String LIKE_BIZ_KEY_PREFIX = "likes:set:biz:";
/*业务点赞数统计的KEY前缀,后缀是业务类型*/
String LIKE_COUNT_KEY_PREFIX = "likes:times:type:";
}
4.2.1.点赞接口
接下来,我们定义一个新的点赞业务实现类:
代码如下:
/**
* <p>
* 点赞记录表 服务实现类
* </p>
*/
@Service
@RequiredArgsConstructor
public class LikedRecordServiceRedisImpl extends ServiceImpl<LikedRecordMapper, LikedRecord> implements ILikedRecordService {
private final RabbitMqHelper mqHelper;
private final StringRedisTemplate redisTemplate;
@Override
public void addLikeRecord(LikeRecordFormDTO recordDTO) {
// LikeRecordFormDTO 包括点赞业务id 、点赞业务类型 、以及是点赞还是取消赞
// 1.基于前端的参数,判断是执行点赞还是取消点赞
boolean success = recordDTO.getLiked() ? like(recordDTO) : unlike(recordDTO);
// 2.判断是否执行成功,如果失败,则直接return结束
if (!success) {
return;
}
// 3.如果执行成功,统计该业务的点赞总数
Long likedTimes = redisTemplate.opsForSet()
.size(RedisConstants.LIKES_BIZ_KEY_PREFIX + recordDTO.getBizId());
if (likedTimes == null) {
return;
}
// 4.缓存点总数到Redis
redisTemplate.opsForZSet().add(
RedisConstants.LIKES_TIMES_KEY_PREFIX + recordDTO.getBizType(),//业务类型
recordDTO.getBizId().toString(),//业务id
likedTimes//该业务的点赞总数(zset的Score)
);
}
private boolean unlike(LikeRecordFormDTO recordDTO) {
// 1.获取用户id
Long userId = UserContext.getUser();
// 2.获取Key
String key = RedisConstants.LIKES_BIZ_KEY_PREFIX + recordDTO.getBizId();
// 3.执行SREM命令
Long result = redisTemplate.opsForSet().remove(key, userId.toString());
return result != null && result > 0;
}
private boolean like(LikeRecordFormDTO recordDTO) {
// 1.获取用户id
Long userId = UserContext.getUser();
// 2.获取Key
String key = RedisConstants.LIKES_BIZ_KEY_PREFIX + recordDTO.getBizId();
// 3.执行SADD命令
Long result = redisTemplate.opsForSet().add(key, userId.toString());
return result != null && result > 0;
}
}
4.2.2.批量查询点赞状态统计
目前我们的Redis点赞记录数据结构如下:
点赞记录数据结构是set,一个业务id对应许多用户id(用户id唯一)
//判断用户是否点赞
SISMEMBER bizId userId
需要注意的是,这个命令只能判断一个用户对某一个业务的点赞状态。而我们的接口是要查询当前用户对多个业务的点赞状态。
因此,我们就需要多次调用SISMEMBER命令,也就需要向Redis多次发起网络请求,给网络带宽带来非常大的压力,影响业务性能。
那么,有没有办法能够一个命令完成多个业务点赞状态判断呢?
非常遗憾,答案是没有!只能多次执行SISMEMBER命令来判断。
不过,Redis中提供了一个功能,可以在一次请求中执行多个命令,实现批处理效果。这个功能就是Pipeline(管道)
不要在一次批处理中传输太多命令,否则单次命令占用带宽过多,会导致网络阻塞
Spring提供的RedisTemplate也具备pipeline功能,最终批量查询点赞状态功能实现如下:
@Override
public Set<Long> isBizLiked(List<Long> bizIds) {//传过来业务id集合
// 1.获取登录用户id
Long userId = UserContext.getUser();
// 2.查询点赞状态
List<Object> objects = redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
StringRedisConnection src = (StringRedisConnection) connection;
for (Long bizId : bizIds) {//遍历业务id集合
String key = RedisConstants.LIKES_BIZ_KEY_PREFIX + bizId;
src.sIsMember(key, userId.toString());//管道技术 批量处理 判断当前业务(key)对应的set中有无当前用户id
}
return null;
});
// 3.返回结果 set集合(点赞的业务id集合)
return IntStream.range(0, objects.size()) // 创建从0到集合size的流
.filter(i -> (boolean) objects.get(i)) // 遍历每个元素,保留结果为true的角标i
.mapToObj(bizIds::get)// 用角标i取bizIds中的对应数据,就是点赞过的id
.collect(Collectors.toSet());// 收集
}
4.2.3.定时任务
点赞成功后,会更新点赞总数并写入Redis中。而我们需要定时读取这些点赞总数的变更数据,通过MQ发送给业务方。这就需要定时任务来实现了。
定时任务的实现方案有很多,简单的例如:
- SpringTask
- Quartz
还有一些依赖第三方服务的分布式任务框架: - Elastic-Job
- XXL-Job
此处我们先使用简单的SpringTask来实现并测试效果。
首先,在tj-remark模块的RemarkApplication启动类上添加注解:
其作用就是启用Spring的定时任务功能。然后,定义一个定时任务处理器类:
代码如下:
@Component
@RequiredArgsConstructor
public class LikedTimesCheckTask {
private static final List<String> BIZ_TYPES = List.of("QA", "NOTE");
private static final int MAX_BIZ_SIZE = 30;
private final ILikedRecordService recordService;
@Scheduled(fixedDelay = 20000)//20秒
public void checkLikedTimes(){
for (String bizType : BIZ_TYPES) {
recordService.readLikedTimesAndSendMessage(bizType, MAX_BIZ_SIZE);
//由于可能存在多个业务类型,不能厚此薄彼只处理部分业务。
// 所以我们会遍历多种业务类型,分别处理。同时为了避免一次处理的业务过多,
//这里设定了每次处理的业务数量为30(MAX_BIZ_SIZE),当然这些都是可以调整的。
}
}
}
这是一个定时任务,真正处理业务的逻辑封装到了ILikedRecordService中:
public interface ILikedRecordService extends IService<LikedRecord> {
// ... 略
void readLikedTimesAndSendMessage(String bizType, int maxBizSize);
}
其实现类:
@Override //业务类型 , 每次处理的业务数量
public void readLikedTimesAndSendMessage(String bizType, int maxBizSize) {
// 1.读取并移除Redis中缓存的点赞总数
String key = RedisConstants.LIKES_TIMES_KEY_PREFIX + bizType;//zset的key (业务类型)
//使用Spring Data Redis库来操作Redis中的有序集合(ZSet)数据结构的代码
Set<ZSetOperations.TypedTuple<String>> tuples =
redisTemplate.opsForZSet().popMin(key, maxBizSize);
// popMin 表示从有序集合中移除并返回分数(score)最小的成员。key 是要操作的有序集合的键名,
//maxBizSize(设置的是30)是要移除并返回的成员数量。
//综合起来,这行代码的意思是从Redis中的指定有序集合中,移除并返回分数最小的前30个成员。
//每个返回的成员都会以 ZSetOperations.TypedTuple<String> 类型的对象形式存储在变量 tuples 中,
//其中包括成员的值(String类型)和对应的分数。这在处理排行榜、取出前N个元素等情况下非常有用。
if (CollUtils.isEmpty(tuples)) {
return;//没数据 结束return
}
}
// 2.数据转换
List<LikedTimesDTO> list = new ArrayList<>(tuples.size());
for (ZSetOperations.TypedTuple<String> tuple : tuples) {
String bizId = tuple.getValue();
Double likedTimes = tuple.getScore();
if (bizId == null || likedTimes == null) {
continue;
}
//加入到list集合,LikedTimesDTO实体类包含业务id和点赞总数
list.add( LikedTimesDTO.of(Long.valueOf(bizId), likedTimes.intValue()) );
}
// 3.发送MQ消息
mqHelper.send(
LIKE_RECORD_EXCHANGE,
StringUtils.format(LIKED_TIMES_KEY_TEMPLATE, bizType),//"{bizType}.times.changed"
list);
}
4.2.4.监听点赞数变更
需要注意的是,由于在定时任务中一次最多处理30条数据,这些数据就需要通过MQ一次发送到业务方.
也就是说MQ的消息体变成了一个集合:即上述代码中的
List<LikedTimesDTO> list = new ArrayList<>(tuples.size());
因此,作为业务方,在监听MQ消息的时候也必须接收集合格式。
我们修改tj-learning中的类com.tianji.learning.mq.LikeTimesChangeListener:
tj-learning中的数据库表中有点赞总数的字段(回答或评论),需要更新。
@Slf4j
@Component
@RequiredArgsConstructor
public class LikeTimesChangeListener {
//互动问题的回答或评论 服务接口对象
private final IInteractionReplyService replyService;
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "qa.liked.times.queue", durable = "true"),
exchange = @Exchange(name = LIKE_RECORD_EXCHANGE, type = ExchangeTypes.TOPIC),
key = QA_LIKED_TIMES_KEY
))
//监听方法 参数是消息通知传来的LikedTimesDTO实体类 包含业务id和点赞总数
public void listenReplyLikedTimesChange(List<LikedTimesDTO> likedTimesDTOs){
log.debug("监听到回答或评论的点赞数变更");
//InteractionReply回答系统实体类
List<InteractionReply> list = new ArrayList<>(likedTimesDTOs.size());
for (LikedTimesDTO dto : likedTimesDTOs) {
InteractionReply r = new InteractionReply();
r.setId(dto.getBizId());//互动问题的回答id (对应业务id)
r.setLikedTimes(dto.getLikedTimes());//点赞数量
list.add(r);
}
//批量更新(持久化)
replyService.updateBatchById(list);
}
}