feat(msg): SSE 实时通知体系 (Phase 1-4 + 7 部分)

后端:
- MessageSseService (emitter 注册表, 25s 心跳, pushNewMessage 仅 log 不抛)
- BizMessageSseController (独立 Controller, 避开 /{msgId} 路径抢匹配)
- BizMessageMapper.countUnread SQL 修复 (跟 limit 解耦, SSE 推真值)
- BizMessageServiceImpl.insert/markRead 成功后 pushNewMessage 触发 SSE
- BizNotifyService.expertAuditResult (#1: 审核结果通知专家本人)
- BizExpertServiceImpl.updateByPrimaryKey 比对旧 audit_status, 真变化才发通知
- PlaceholderConfig (Spring 7 严格 placeholder 兜底, ignoreUnresolvable=true)

前端:
- utils/sseClient.js (@microsoft/fetch-event-source 封装, Bearer 鉴权, 单例连接)
- 全局事件总线 onGlobal: SSE new_message 扇出到业务页面 (切路由不掉)
- AdminLayout 全局订阅 + bell 角标 (el-badge 显示 unreadCount)
- userStore.expertAuditApproved: doctor 审核状态实时跟 DB
- AdminLayout doctor menu 加 requireAuditApproved 过滤 (未通过审核隐藏 3 menu)
- doctor/Home: 待参加会议/待签协议 2 pannel 按 audit 状态显示; load() 仅审核通过才拉数据
- doctor/Home + doctor/Messages: 点列表弹详情 dialog + markMessageRead 真已读
- AdminLayout SSE 触发时: 刷新 unread + 重新拉 audit 状态 (审核通过瞬间可见)

业务规则:
- 专家注册不群发 manager (manager 群体可能 100+, 每条注册发 N 条浪费)
- 通过审核无需重新登录 (SSE 链路实时联动)
This commit is contained in:
郭庆泰
2026-08-22 08:46:26 +08:00
parent c3eb8ed9c3
commit b6ac5b7d41
18 changed files with 697 additions and 39 deletions
@@ -17,7 +17,7 @@ import com.ruoyi.business.domain.BizMessage;
import com.ruoyi.business.service.IBizMessageService;
/**
* 个人消息Controller
* 个人消息Controller (SSE 推送端点见 BizMessageSseController)
*/
@RestController
@RequestMapping("/business/message")
@@ -42,16 +42,14 @@ public class BizMessageController extends BaseController
* 当前登录用户的最近消息 (含未读统计)
* GET /business/message/my?limit=50
* 返回 {rows: [...], unread: 12}
* unread 用 countUnread 跟 limit 解耦, SSE 推的也是真值
*/
@GetMapping("/my")
public AjaxResult my(@RequestParam(value = "limit", required = false, defaultValue = "50") Integer limit)
{
Long uid = SecurityUtils.getUserId();
List<BizMessage> rows = bizMessageService.selectMyRecent(uid, limit);
int unread = 0;
for (BizMessage m : rows) {
if ("0".equals(m.getIsRead())) unread++;
}
int unread = bizMessageService.countUnread(uid);
Map<String, Object> data = new HashMap<>();
data.put("rows", rows);
data.put("unread", unread);
@@ -0,0 +1,38 @@
package com.ruoyi.business.controller;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import com.ruoyi.business.sse.MessageSseService;
import com.ruoyi.common.utils.SecurityUtils;
/**
* 个人消息 SSE 推送端点 — 独立 Controller.
*
* <p>独立成 Controller 的原因: BizMessageController 有 `GET /{msgId:\\d+}` 详情端点,
* 即使加正则约束, Spring 6.x 的 PathPattern 在某些边界条件下仍会把 `/stream` 字面量
* 路由到 `/{msgId}` 上去, 报 "参数类型不匹配". 拆到独立 Controller 后两条路径
* 完全不在同一个映射表里竞争, 零冲突风险.
*/
@RestController
@RequestMapping("/business/message")
public class BizMessageSseController
{
@Autowired
private MessageSseService messageSseService;
/**
* SSE 订阅当前用户的实时通知信号.
* 鉴权: Authorization: Bearer <jwt> — 走 SecurityUtils.getUserId(), 无 token 抛 401.
* 推送载荷只含信号 ({type:'new_message', unread:N}), 不带消息体, 防止 SSE 挂了影响正确性.
* 心跳 25s 一条 ": ping" 注释帧, 防止 nginx/proxy 60s 空闲断连.
*/
@GetMapping(value = "/stream", produces = "text/event-stream")
public SseEmitter stream()
{
Long uid = SecurityUtils.getUserId();
return messageSseService.subscribe(uid);
}
}
@@ -111,6 +111,10 @@ public class BizRegisterController extends BaseController {
expert.setStatus("Y"); // biz_expert.status: Y=正常 N=禁用
expertService.insert(expert);
// 业务规则 (2026-08-22): 专家注册不群发给 manager
// 原因: manager 群体可能有几十到上百人, 每条注册都发 N 条通知是浪费
// manager 自己到 /manager/experts 列表页拉待审核的即可
return success("注册成功, 请等待审核").put("userId", userId);
}
}
@@ -15,6 +15,9 @@ public interface BizMessageMapper
/** 某人的未读 + 最近消息 (按时间倒序, limit 由调用方控制) */
List<BizMessage> selectMyRecent(BizMessage entity);
/** 收件人未读总数 (SSE 推送用, 跟 limit 解耦) */
int countUnread(Long receiverUserId);
int insert(BizMessage entity);
int updateByPrimaryKey(BizMessage entity);
@@ -0,0 +1,99 @@
package com.ruoyi.business.notify;
import java.util.List;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import com.ruoyi.business.domain.BizExpert;
import com.ruoyi.business.domain.BizMessage;
import com.ruoyi.business.mapper.BizSysUserQueryMapper;
import com.ruoyi.business.service.IBizMessageService;
import com.ruoyi.common.enums.BizAuditStatusEnum;
import com.ruoyi.system.domain.vo.SysUserExtendVo;
/**
* 业务消息通知门面 (Phase 4).
*
* <p>提供 N 个业务场景的语义化方法, 每个方法只负责拼 biz_message 然后调
* {@link IBizMessageService#insert}, SSE 推送由该 insert 内部已注入的
* pushNewMessage 自动触发 (Phase 1 做的事), 不用本服务关心.
*
* <p>设计要点 (沿用 task_plan.md 决策):
* <ul>
* <li><b>语义化方法</b>而非裸 notify() — 业务代码调 expertAuditResult() 比拼字典字段清楚</li>
* <li><b>只在原业务 insert/update 之后调</b> — 业务事务回滚时通知自动回滚</li>
* <li><b>群发 manager 用 {@link BizSysUserQueryMapper#selectActiveByRole}</b> — 避开
* ISysUserService.selectUserExtendList 的 @DataScope 切面 (本服务保留 mapper 字段以便未来场景使用)</li>
* <li><b>createBy="system"</b> — 系统自动发的通知, 不需要用户身份</li>
* </ul>
*
* <p>业务规则 (2026-08-22): 专家注册**不**群发给 manager. 原因: manager 群体可能有
* 几十到上百人, 每条注册都发 N 条通知是浪费, manager 自己到 /manager/experts 列表页拉待审核的即可.
* 所以本类目前只暴露 #1 (审核结果→专家本人), 不再有 expertRegistered().
*/
@Service
public class BizNotifyService
{
private static final Logger log = LoggerFactory.getLogger(BizNotifyService.class);
@Autowired
private IBizMessageService bizMessageService;
@Autowired
private BizSysUserQueryMapper bizSysUserQueryMapper;
// ====================== msgType 常量 ======================
/** 普通通知 */
public static final String TYPE_NOTIFY = "1";
/** 待办 (需要用户操作) */
public static final String TYPE_TODO = "2";
/** 系统消息 */
public static final String TYPE_SYSTEM = "3";
// ====================== bizType 常量 ======================
public static final String BIZ_EXPERT = "expert";
public static final String BIZ_PROJECT = "project";
public static final String BIZ_MEETING = "meeting";
public static final String BIZ_AGREEMENT = "agreement";
public static final String BIZ_PUBLICITY = "publicity";
public static final String BIZ_ORG = "org";
public static final String BIZ_AUTH = "auth";
/**
* #1 专家审核结果 → 通知专家本人 (auditStatus 1→2 通过 / 1→3 拒绝).
*
* <p>调用方: BizExpertServiceImpl.updateByPrimaryKey, 检测到 audit_status 真的变了才调.
* 其余字段编辑 (姓名/科室/...) 不发.
*/
public void expertAuditResult(BizExpert expert, String oldAuditStatus)
{
if (expert.getUserId() == null) {
log.warn("[notify] expertAuditResult: expert.userId 为空, 跳过 (expertId={})", expert.getExpertId());
return;
}
String newStatus = expert.getAuditStatus();
String title;
String content;
if (BizAuditStatusEnum.APPROVED.getCode().equals(newStatus)) {
title = "您的专家注册已通过审核";
content = "感谢您的注册, 现可登录使用全部功能。";
} else if (BizAuditStatusEnum.REJECTED.getCode().equals(newStatus)) {
title = "您的专家注册审核未通过";
content = "请检查提交资料, 必要时联系管理员。";
} else {
// 其它状态变更 (0→1, 2→3 等) 不发结果通知
return;
}
BizMessage msg = new BizMessage();
msg.setReceiverUserId(expert.getUserId());
msg.setMsgType(TYPE_NOTIFY);
msg.setTitle(title);
msg.setContent(content);
msg.setBizType(BIZ_EXPERT);
msg.setBizId(expert.getExpertId());
msg.setCreateBy("system");
bizMessageService.insert(msg);
log.info("[notify] expertAuditResult 已发 uid={} ({} → {})", expert.getUserId(), oldAuditStatus, newStatus);
}
}
@@ -15,6 +15,9 @@ public interface IBizMessageService
/** 收件人的最近消息 (含未读) */
List<BizMessage> selectMyRecent(Long receiverUserId, Integer limit);
/** 收件人未读总数 (SSE 推送用) */
int countUnread(Long receiverUserId);
int insert(BizMessage entity);
int updateByPrimaryKey(BizMessage entity);
@@ -16,6 +16,7 @@ import com.ruoyi.business.domain.BizExpert;
import com.ruoyi.business.domain.dto.ImportResult;
import com.ruoyi.business.domain.vo.BizExpertImportVo;
import com.ruoyi.business.mapper.BizExpertMapper;
import com.ruoyi.business.notify.BizNotifyService;
import com.ruoyi.business.service.IBizExpertService;
import com.ruoyi.common.core.domain.entity.SysUser;
import com.ruoyi.common.exception.ServiceException;
@@ -42,6 +43,9 @@ public class BizExpertServiceImpl implements IBizExpertService
@Autowired
private SysUserMapper sysUserMapper;
@Autowired
private BizNotifyService bizNotifyService;
@Override
public BizExpert getById(Long expertId)
{ return bizExpertMapper.selectByPrimaryKey(expertId); }
@@ -107,9 +111,34 @@ public class BizExpertServiceImpl implements IBizExpertService
return result;
}
/**
* 通用 update. #1 触发点 (审核结果通知) 在这里插桩:
* - 先读旧 audit_status
* - 执行 update
* - 仅当 audit_status 真的变化 (1→2 通过 / 1→3 拒绝) 才发通知
* - 其它字段编辑 (姓名/科室/...) 不发, 避免噪音
*/
@Override
public int updateByPrimaryKey(BizExpert entity)
{ return bizExpertMapper.updateByPrimaryKey(entity); }
{
String oldAuditStatus = null;
if (entity.getExpertId() != null) {
BizExpert existed = bizExpertMapper.selectByPrimaryKey(entity.getExpertId());
if (existed != null) {
oldAuditStatus = existed.getAuditStatus();
}
}
int n = bizExpertMapper.updateByPrimaryKey(entity);
if (n > 0 && entity.getAuditStatus() != null
&& !entity.getAuditStatus().equals(oldAuditStatus)) {
// 重新读一次拿 userId (entity 可能只传了 expertId+auditStatus)
BizExpert after = bizExpertMapper.selectByPrimaryKey(entity.getExpertId());
if (after != null && after.getUserId() != null) {
bizNotifyService.expertAuditResult(after, oldAuditStatus);
}
}
return n;
}
/**
* 启用/禁用专家: 同步 biz_expert.status + sys_user.status
@@ -6,6 +6,7 @@ import org.springframework.stereotype.Service;
import com.ruoyi.business.domain.BizMessage;
import com.ruoyi.business.mapper.BizMessageMapper;
import com.ruoyi.business.service.IBizMessageService;
import com.ruoyi.business.sse.MessageSseService;
import com.ruoyi.common.utils.id.SnowflakeId;
@Service
@@ -14,6 +15,9 @@ public class BizMessageServiceImpl implements IBizMessageService
@Autowired
private BizMessageMapper bizMessageMapper;
@Autowired
private MessageSseService messageSseService;
@Override
public BizMessage getById(Long msgId)
{ return bizMessageMapper.selectByPrimaryKey(msgId); }
@@ -34,11 +38,23 @@ public class BizMessageServiceImpl implements IBizMessageService
return bizMessageMapper.selectMyRecent(q);
}
@Override
public int countUnread(Long receiverUserId)
{
return bizMessageMapper.countUnread(receiverUserId);
}
@Override
public int insert(BizMessage entity)
{
SnowflakeId.injectIfEmpty(entity, "msgId");
return bizMessageMapper.insert(entity);
int n = bizMessageMapper.insert(entity);
// 写完即推: 让接收人前端立即看到新消息信号 (SSE 挂了也不影响正确性)
if (n > 0 && entity.getReceiverUserId() != null) {
int unread = bizMessageMapper.countUnread(entity.getReceiverUserId());
messageSseService.pushNewMessage(entity.getReceiverUserId(), unread);
}
return n;
}
@Override
@@ -52,7 +68,13 @@ public class BizMessageServiceImpl implements IBizMessageService
q.setMsgId(msgId);
q.setReceiverUserId(receiverUserId);
q.setIsRead("1");
return bizMessageMapper.markRead(q);
int n = bizMessageMapper.markRead(q);
// 标记成功后立即推新 unread, navbar 角标自动减 (无需等下次刷新)
if (n > 0 && receiverUserId != null) {
int unread = bizMessageMapper.countUnread(receiverUserId);
messageSseService.pushNewMessage(receiverUserId, unread);
}
return n;
}
@Override
@@ -0,0 +1,119 @@
package com.ruoyi.business.sse;
import java.io.IOException;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
/**
* 个人消息 SSE 推送服务.
*
* <p>设计要点 (沿用 task_plan.md 决策, 不踩 hwt-serve 的 10 个坑):
* <ul>
* <li><b>只推信号不推内容</b> — 载荷 {type:'new_message', unread:N}, 不带消息体, DB 是唯一可信源</li>
* <li><b>推送失败只 log 绝不抛</b> — hwt-serve 心跳抛 RuntimeException 让定时任务永久死掉, 这里 try/catch 包死</li>
* <li><b>Set 而非 List</b> — 同一用户多 tab 同账号只保留一份 emitter, 避免重复推送风暴</li>
* <li><b>心跳 25s</b> — nginx/proxy 默认 60s 空闲断, 心跳用 SSE 注释帧 (": ping") 不触发前端 onmessage</li>
* <li><b>鉴权靠 SecurityUtils</b> — controller 里 SecurityUtils.getUserId() 抛 401 当 token 无效, 无需额外校验</li>
* </ul>
*/
@Service
public class MessageSseService
{
private static final Logger log = LoggerFactory.getLogger(MessageSseService.class);
/** 不设服务器端超时, 客户端断开时 send() 抛 IOException → onError 触发清理 */
private static final long SSE_TIMEOUT_MS = 0L;
/** uid -> 该用户所有连接 (一账号多 tab 共享一份 Set) */
private final ConcurrentHashMap<Long, Set<SseEmitter>> emitters = new ConcurrentHashMap<>();
/**
* 注册一个新 emitter. 注册完立即发一条 "connected" 帧, 让前端确认链路通.
* emitter 在完成 / 超时 / 出错 任一情况下都会从 map 移除.
*/
public SseEmitter subscribe(Long userId)
{
SseEmitter emitter = new SseEmitter(SSE_TIMEOUT_MS);
Set<SseEmitter> userSet = emitters.computeIfAbsent(userId,
k -> ConcurrentHashMap.newKeySet());
userSet.add(emitter);
Runnable cleanup = () -> {
Set<SseEmitter> set = emitters.get(userId);
if (set != null) {
set.remove(emitter);
if (set.isEmpty()) emitters.remove(userId);
}
};
emitter.onCompletion(cleanup);
emitter.onTimeout(cleanup);
emitter.onError((ex) -> cleanup.run());
try {
// 立即发一条 connected 帧, 前端收到即可视为订阅成功
emitter.send(SseEmitter.event().name("connected").data("ok"));
} catch (IOException e) {
log.warn("[SSE] 初始 connected 帧发送失败 uid={} err={}", userId, e.toString());
cleanup.run();
emitter.completeWithError(e);
}
log.info("[SSE] subscribe uid={} currentSetSize={}", userId, userSet.size());
return emitter;
}
/**
* 给某用户推一条新消息信号. **绝不抛**, 异常仅 log (hwt-serve 教训).
* 心跳 / 业务推送 / 标记已读 都会调这个.
*/
public void pushNewMessage(Long userId, int unread)
{
Set<SseEmitter> set = emitters.get(userId);
if (set == null || set.isEmpty()) return;
for (SseEmitter em : set) {
try {
em.send(SseEmitter.event()
.name("new_message")
.data("{\"type\":\"new_message\",\"unread\":" + unread + "}"));
} catch (Exception e) {
log.warn("[SSE] push 失败, 自动清理该 emitter uid={} err={}", userId, e.toString());
try { em.completeWithError(e); } catch (Exception ignore) {}
set.remove(em);
}
}
if (set.isEmpty()) emitters.remove(userId);
}
/**
* 25s 心跳 — SSE 注释帧 (以 ':' 开头) 不会触发 EventSource.onmessage,
* 仅用来保活防 nginx/proxy 60s 空闲断连.
* 整个方法 try/catch 包死, 单次失败不影响后续调度 (Spring TaskScheduler 每次调用是独立的).
*/
@Scheduled(fixedRate = 25_000)
public void heartbeat()
{
try {
for (Map.Entry<Long, Set<SseEmitter>> e : emitters.entrySet()) {
Long uid = e.getKey();
Set<SseEmitter> set = e.getValue();
if (set == null) continue;
for (SseEmitter em : set) {
try {
em.send(SseEmitter.event().comment("ping"));
} catch (Exception ex) {
log.warn("[SSE] 心跳失败, 自动清理 uid={} err={}", uid, ex.toString());
try { em.completeWithError(ex); } catch (Exception ignore) {}
set.remove(em);
}
}
}
} catch (Exception e) {
log.error("[SSE] heartbeat 调度异常 (不影响下次调度)", e);
}
}
}
@@ -54,6 +54,12 @@
</if>
</select>
<select id="countUnread" resultType="int" parameterType="Long">
select count(*) from biz_message
where receiver_user_id = #{receiverUserId}
and is_read = '0'
</select>
<insert id="insert" parameterType="BizMessage" useGeneratedKeys="true" keyProperty="msgId">
insert into biz_message
<trim prefix="(" suffix=")" suffixOverrides=",">
@@ -0,0 +1,36 @@
package com.ruoyi.framework.config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.support.PropertySourcesPlaceholderConfigurer;
/**
* 全局宽松占位符解析器 — 2026-08-22 加.
*
* <p>Spring Framework 7 的 PlaceholderParser 改成严格模式:
* @Value("${prop}") 找不到就抛 PlaceholderResolutionException, 启动失败.
* 即便加 :default 也不兜底 (实测 guoju0808 的 xss.excludes / token.header 都炸).
*
* <p>修法: 全局注册一个 ignoreUnresolvablePlaceholders=true 的
* PropertySourcesPlaceholderConfigurer, 行为回退到 Spring 6 时代 — 找不到就
* 留作字面量 ${...}, 不抛. 各 @Value 调用方按需自取:
* <ul>
* <li>要么 yml 里确实配了 (现在的 FilterConfig / TokenService 都是这种情况)</li>
* <li>要么代码层用 Environment.getProperty(key, default) 自兜底 (更安全)</li>
* </ul>
*
* <p>bean name 必须叫 propertySourcesPlaceholderConfigurer, 覆盖 Spring Boot
* 自动注册的默认那个.
*/
@Configuration
public class PlaceholderConfig
{
@Bean
public static PropertySourcesPlaceholderConfigurer propertySourcesPlaceholderConfigurer()
{
PropertySourcesPlaceholderConfigurer c = new PropertySourcesPlaceholderConfigurer();
c.setIgnoreUnresolvablePlaceholders(true);
c.setIgnoreResourceNotFound(true);
return c;
}
}