【解决方案】Java 互联网项目中消息通知体系的计划与实现（下） ...

冬雨财经 · 2024-8-5 07:44:43

媒介

书接上回，消息通知体系（notification-system）作为一个独立的微服务，完备地负责了 App 端内所有消息通知相关的后端功能实现。该体系既必要与文章体系、订单体系、会员体系等相关联，也必要和别的业务体系相关联，是一个偏底层的通用服务体系。
App 端内的消息通知类型常见有这几项：评论通知、点赞通知、收藏通知、订单通知、活动通知、个人中央相关通知等。该体系在可拓展性、高性能、较高可用性、数据同等性等方面有较高要求，最终目的是提升用户粘性、增强 App 与用户的互动、支持焦点业务的发展。
文章的（上）篇将从需求分析、数据模型计划、关键流程计划这 3 部分来说明，（下）篇将从技能选型、后端接口计划、关键逻辑实现这 3 部分来进行说明。
四、技能选型

我将该体系必要利用到的关键技能选型做成表格，方便梳理：

说明：

可以用 Spirng Cloud 或者 Spirng Cloud Alibaba，哪个习惯用哪个，只要是能打包成一个可运行的微服务即可；
也可以用非关系型数据库如 MongoDB 来代替 MySQL，表与表之间的关系不密切的条件下，性能会更高；
Redis 拿来做缓存中央件去存储非布局化的一些数据黑白常合适的，许多场景下，突出的性能和便捷的 API 是它的上风；
MQ 其实是选用的，适合较为复杂的项目拿来异步/解耦，既可以 kafka 也可以 RabbitMQ，RocketMQ 是阿里亲生的，控制台用起来也方便；
别的开源依赖最好利用 apache 的顶级项目或者 Spring 官方的，像 hutool 这种第三方的包其实不太推荐，安全风险可能会比较高。

五、后端接口计划

作为一个偏底层的公共服务，基本上都会先由上游的业务体系进行调用，再服务于用户（即 App 端）。下面计划两个 Controller 分别针对业务端和 App 端，大家可以先参考一下接口规范，也写了总体的思路解释，关键逻辑会在下一节再睁开讲。
5.1业务体系接口

暴露给业务体系的有 3 个接口：

获取通知配置
发送通知
撤回通知

@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，已读消息时做算术更新：

Redis-Hash 布局说明：

Redis-key 是固定 String 常量 "sysName.notice.num.key"；
Hash-key 为 App 端用户唯一的 userUuid；
Hash-value 为该用户吸收的消息总数，新增 +1，撤回 -1。

如果大家对于 Redis 的基本布局还不太相识，参考下我的这篇博客：https://www.cnblogs.com/CodeBlogMan/p/17816699.html
下面是关键实现步骤的代码示例：

新增消息
1. //先入 MySQL
2. Notification notification = this.insertNotice(dto);
3. //再入 Redis
4. redisTemplate.opsForHash().increment(RedisKey, dto.getTargetUserUuid(), 1);
复制代码
撤回消息
1. //先更新 MySQL
2. this.updateById(notification);
3. //再更新 Redis
4. redisTemplate.opsForHash().increment(RedisKey, userUuid, -1);
复制代码

注意：
写操纵和更新操纵都是先操纵数据库，然后再同步入 Redis。原因：数据库里的数据是源头，且存的是布局化的持久性数据；Redis 只是作为缓存，发挥 Redis 读取速度快的优点，存储的是一些 size 不大的热点数据。
6.2已读消息处置处罚

已读和未读其实就是两种状态，Redis 里一开始存储的都是未读数，当用户点击查看列表时，前端会调用后端的消息列表接口，消息列表直接查数据库（记录了已读和未读状态），此时同步更新 Redis 里的未读消息数，那么此时：未读消息数 = Redis总数 - MySQL已读消息数。
下面的代码说得比较清楚了：

查询未读消息数
1. Integer num;
2. //先读 redis，没有再读数据库，最后再把数据库读出的放回 redis
3. num = (Integer) redisTemplate.opsForHash().get(RedisKey, userUuid);
4. //防止一开始新增通知的时候没放进 redis 里，null 表示什么都没有，而不是 0
5. if (Objects.nonNull(num)) {
6. msgNumVO.setMsgNum(num);
7. }else {
8. num = this.getNoticeNum(userUuid, queryDate);
9. log.info("缓存中没有未读消息总数，查数据库：{}", num);
10. msgNumVO.setMsgNum(num);
11. //放入缓存，取出什么放什么
12. redisTemplate.opsForHash().put(RedisKey, userUuid, num);
13. }
14. return num;
复制代码
查询消息列表
1. wrapper.eq(Notification::getTargetUserUuid, userUuid)
2. .eq(Notification::getSuperType, superType)
3. .eq(Notification::getMsgStatus, StatusEnum.TRUE.getType())
4. .le(Notification::getCreateTime, dateTime)
5. .orderByDesc(Notification::getCreateTime);
6. List<Notification> queryList = pageInfo.getResult();
7. //查询后即要同步去更新数据库中该类型下的消息为已读
8. this.updateListBySuperType(wrapper);
9. long isReadNum;
10. isReadNum = queryList.stream().filter(val -> NumberUtils.INTEGER_ZERO.equals(val.getIsRead())).count();
11. //关键的一步，同步更新 redis 里的未读消息数
12. Integer redisNum = (Integer) redisTemplate.opsForHash().get(RedisKey.INITIAL_NOTICE_NUM_PERFIX, userUuid);
13. //要先判断 redis 里是否为 null，和 0 不一样
14. int hv = Objects.isNull(redisNum) ? 0 : (int) (redisNum - isReadNum);
15. redisTemplate.opsForHash().put(RedisKey, userUuid, Math.max(hv, 0));
16. 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 开源堆栈应该已经建立好了。
文章如有错误和不敷，还望指正，同时也接待大家在评论区说出自己的想法！

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

		自动登录	找回密码
密码			立即注册

【解决方案】Java 互联网项目中消息通知体系的计划与实现（下） ...

本帖子中包含更多资源

0 个回复

快速回复

楼主热帖

标签云