【解决方案】Java 互联网项目中消息通知系统的设计与实现(下)
cnblogs 2024-08-05 09:09:00 阅读 60
书接上回,消息通知系统(notification-system)作为一个独立的微服务,完整地负责了 App 端内所有消息通知相关的后端功能实现。该系统既需要与文章系统、订单系统、会员系统等相关联,也需要和其它业务系统相关联,是一个偏底层的通用服务系统。
目录
- <li>前言
- 四、技术选型
- 五、后端接口设计
- 5.1业务系统接口
- 5.2App 端接口
- 六、关键逻辑实现
- 6.1Redis存储结构
- 6.2已读消息处理
- 6.3缓存定时清除
- 本篇小结
前言
书接上回,消息通知系统(notification-system)作为一个独立的微服务,完整地负责了 App 端内所有消息通知相关的后端功能实现。该系统既需要与文章系统、订单系统、会员系统等相关联,也需要和其它业务系统相关联,是一个偏底层的通用服务系统。
App 端内的消息通知类型常见有这几项:评论通知、点赞通知、收藏通知、订单通知、活动通知、个人中心相关通知等。该系统在可拓展性、高性能、较高可用性、数据一致性等方面有较高要求,最终目的是提升用户粘性、加强 App 与用户的互动、支撑核心业务的发展。
文章的(上)篇将从需求分析、数据模型设计、关键流程设计这 3 部分来说明,(下)篇将从技术选型、后端接口设计、关键逻辑实现这 3 部分来进行说明。
四、技术选型
我将该系统需要使用到的关键技术选型做成表格,方便梳理:
说明:
- <li>可以用 Spirng Cloud 或者 Spirng Cloud Alibaba,哪个习惯用哪个,只要是能打包成一个可运行的微服务即可;
- 也可以用非关系型数据库如 MongoDB 来代替 MySQL,表与表之间的关系不密切的前提下,性能会更高;
- Redis 拿来做缓存中间件去存储非结构化的一些数据是非常合适的,很多场景下,突出的性能和便捷的 API 是它的优势;
- MQ 其实是选用的,适合较为复杂的项目拿来异步/解耦,既可以 kafka 也可以 RabbitMQ,RocketMQ 是阿里亲生的,控制台用起来也方便;
- 其它开源依赖最好使用 apache 的顶级项目或者 Spring 官方的,像 hutool 这种第三方的包其实不太推荐,安全风险可能会比较高。
五、后端接口设计
作为一个偏底层的公共服务,基本上都会先由上游的业务系统进行调用,再服务于用户(即 App 端)。下面设计两个 Controller 分别针对业务端和 App 端,大家可以先参考一下接口规范,也写了总体的思路注释,关键逻辑会在下一节再展开讲。
5.1业务系统接口
暴露给业务系统的有 3 个接口:
- 获取通知配置
- 发送通知
- 撤回通知
<code>@RestController
@RequestMapping("notice/api")
public class NoticeApiController {
@Resource
private NotificationService notificationService;
/**
* 新增通知,业务系统用
* @param dto
* @return 消息系统唯一 id
*/
@PostMapping("/add")
public Response<Long> addNotice(@Valid @RequestBody AddNoticeDTO dto){
//业务方调用该接口前需要先根据 sourceId 确认来源,实现就是先入数据库,再入 Redis
return ResponseBuilder.buildSuccess(this.notificationService.addNotice(dto));
}
/**
* 撤回通知(同批量撤回),业务系统用
* @param idList,需要撤回的消息主键 id 集合
* @return 是否成功:true-成功,false-失败
*/
@PostMapping("/recall")
public Response<Boolean> recallNotice(@RequestBody List<Long> idList){
//撤回只需要考虑先更新数据库,后更新 Redis
return ResponseBuilder.buildSuccess(this.notificationService.recallNotice(idList));
}
/**
* 获取通知配置
* @param sourceId 业务系统标识
* @return 配置详情信息
*/
@GetMapping("/getNoticeConfig")
public Response<NotificationConfig> getNoticeConfig(@RequestParam(value = "noticeId") String sourceId){
//每个业务系统调用前需要校验通知配置,以防非法调用
return ResponseBuilder.buildSuccess(this.notificationService.getNoticeConfig(sourceId));
}
}
5.2App 端接口
开放给 App 端使用的有 2 个接口:
- 获取用户未读消息总数
- 获取用户消息列表
@RestController
@RequestMapping("notice/app")
public class NoticeAppController {
@Resource
private NotificationService notificationService;
/**
* 获取用户未读消息总数
*/
@Auth
@GetMapping("/num")
public Response<NoticeNumVO> getMsgNum() {
//App 端的用户唯一 uuid
String userUuid = "";
return ResponseBuilder.buildSuccess(this.notificationService.getMsgNum(userUuid));
}
/**
* 获取用户消息列表
*
* @param queryDate:查询时间 queryDate
* @param pageIndex:页码,1开始
* @param pageSize:每页大小
* @param superType:消息父类型,1-评论、点赞、系统消息,2-通知,3-私信,4-客服消息
*/
@Auth
@GetMapping("/list/{queryDate}/{pageIndex}/{pageSize}/{superType}")
public Response<List<Notification>> getNoticeList(@PathVariable String queryDate, @PathVariable Integer pageIndex,
@PathVariable Integer pageSize, @PathVariable Integer superType) throws ParseException {
//App 端的用户唯一 uuid
String userUuid = "";
Date dateStr = DateUtils.parseDate(queryDate, new String[]{"yyyyMMddHHmmss"});
return ResponseBuilder.buildSuccess(this.notificationService.getNoticeList(userUuid, dateStr, pageIndex, pageSize, superType));
}
}
六、关键逻辑实现
本小节会针对 APP 端的两个接口进行详细讲解,未读消息数和消息列表的实现需要 Redis + MySQL 的紧密配合。
6.1Redis存储结构
下面先着重介绍一下本系统的 Redis 缓存结构设计,全局只使用 Hash 结构,新增消息时+1,撤回消息时-1,已读消息时做算术更新:
说明:
- <li>
Hash-key 为 App 端用户唯一的 userUuid;
Hash-value 为该用户接收的消息总数,新增 +1,撤回 -1。
Redis-key 是固定 String 常量 "sysName.notice.num.key";
如果大家对于 Redis 的基本结构还不太了解,参考下我的这篇博客:https://www.cnblogs.com/CodeBlogMan/p/17816699.html
下面是关键实现步骤的代码示例:
新增消息
<code> //先入 MySQL
Notification notification = this.insertNotice(dto);
//再入 Redis
li>redisTemplate.opsForHash().increment(RedisKey, dto.getTargetUserUuid(), 1);
撤回消息
//先更新 MySQL
this.updateById(notification);
//再更新 Redis
redisTemplate.opsForHash().increment(RedisKey, userUuid, -1);
注意:
写操作和更新操作都是先操作数据库,然后再同步入 Redis。原因:数据库里的数据是源头,且存的是结构化的持久性数据;Redis 只是作为缓存,发挥 Redis 读取速度快的优点,存储的是一些 size 不大的热点数据。
6.2已读消息处理
已读和未读其实就是两种状态,Redis 里一开始存储的都是未读数,当用户点击查看列表时,前端会调用后端的消息列表接口,消息列表直接查数据库(记录了已读和未读状态),此时同步更新 Redis 里的未读消息数,那么此时:未读消息数 = Redis总数 - MySQL已读消息数。
下面的代码说得比较清楚了:
查询未读消息数
Integer num;
//先读 redis,没有再读数据库,最后再把数据库读出的放回 redis
num = (Integer) redisTemplate.opsForHash().get(RedisKey, userUuid);
//防止一开始新增通知的时候没放进 redis 里,null 表示什么都没有,而不是 0
if (Objects.nonNull(num)) {
msgNumVO.setMsgNum(num);
}else {
num = this.getNoticeNum(userUuid, queryDate);
log.info("缓存中没有未读消息总数,查数据库:{}", num);
msgNumVO.setMsgNum(num);
//放入缓存,取出什么放什么
redisTemplate.opsForHash().put(RedisKey, userUuid, num);
}
return num;
查询消息列表
wrapper.eq(Notification::getTargetUserUuid, userUuid)
.eq(Notification::getSuperType, superType)
.eq(Notification::getMsgStatus, StatusEnum.TRUE.getType())
.le(Notification::getCreateTime, dateTime)
.orderByDesc(Notification::getCreateTime);
List<Notification> queryList = pageInfo.getResult();
//查询后即要同步去更新数据库中该类型下的消息为已读
this.updateListBySuperType(wrapper);
long isReadNum;
isReadNum = queryList.stream().filter(val -> NumberUtils.INTEGER_ZERO.equals(val.getIsRead())).count();
//关键的一步,同步更新 redis 里的未读消息数
Integer redisNum = (Integer) redisTemplate.opsForHash().get(RedisKey.INITIAL_NOTICE_NUM_PERFIX, userUuid);
//要先判断 redis 里是否为 null,和 0 不一样
int hv = Objects.isNull(redisNum) ? 0 : (int) (redisNum - isReadNum);
redisTemplate.opsForHash().put(RedisKey, userUuid, Math.max(hv, 0));
return queryList;
6.3缓存定时清除
由于在上述的 redis-hash 结构中并没有加入 expire 过期时间,那么显而易见的是这个结构随着时间增加会越来越大,最终导致形成一个大key,给 redis 的读/写性能带来影响。
所以这里需要给出一个方案来解决这个问题,我的核心思路是:
- 每当写redis计数的时候同时用另一个 key 记操作时间,每10分钟执行一次定时任务;
- 逐一将时间 key 的 value (即操作时间)根据 uuid 拿出来,如果当前系统时间 - 该uuid的操作时间>3600ms(即一个小时)那么就将该 uuid 的数据删除;
- 下次调接口先读数据库,再写进 redis 里面,具体看代码。
@Component
@Slf4j
public class HandleNoticeCache {
private static final Long FLAG_TIME = 3600L;
@Resource
private RedisTemplate redisTemplate;
@Scheduled(cron = " * 0/10 * * * ? ")
public void deleteNoticeCache(){
HashOperations<String, String, Integer> hashOperations = redisTemplate.opsForHash();
//通知操作的全部 uuid,数据量一大可能导致 OOM
Set<String> uuidList = hashOperations.keys(RedisKey.NOTICE_NUM_TIME);
if (CollectionUtils.isNotEmpty(uuidList)){
uuidList.forEach(val -> {
Integer operateTime = hashOperations.get(RedisKey.NOTICE_NUM_TIME, val);
if (Objects.nonNull(operateTime)){
//当前系统时间-操作的记录时间
long resultTime = System.currentTimeMillis() - operateTime;
if (resultTime > FLAG_TIME){
hashOperations.delete(RedisKey.NOTICE_NUM_PERFIX, val);
log.info("删除通知的 uuid 为:{}", val);
hashOperations.delete(RedisKey.COMMENT_NUM_PERFIX, val);
log.info("删除评论通知的 uuid 为:{}", val);
}
}
});
}
}
}
本篇小结
到这里关于互联网消息通知系统的设计与实现就分享完了,至于源码我看在周末或者假期有没有时间发出来,之后自己的个人 git 开源仓库应该已经建设好了。
文章如有错误和不足,还望指正,同时也欢迎大家在评论区说出自己的想法!
声明
本文内容仅代表作者观点,或转载于其他网站,本站不以此文作为商业用途
如有涉及侵权,请联系本站进行删除
转载本站原创文章,请注明来源及作者。