RuoYi-Cloud微服务消息中心架构设计与优化实践

1. 项目背景与需求分析在RuoYi-Cloud微服务架构中消息通知模块最初采用SysNotice表结构实现这种设计存在明显的局限性。随着业务复杂度提升系统需要处理多种消息类型站内信、邮件、短信、企业微信等原有的单表结构难以满足以下需求消息类型扩展性差新增消息渠道需修改表结构和代码发送策略单一缺乏优先级、重试机制等高级功能状态追踪困难无法有效监控消息生命周期性能瓶颈高并发场景下单表压力过大2. 架构设计方案2.1 核心组件划分统一消息中心采用分层架构设计┌───────────────────────┐ │ API层 │ │ (消息发送/查询接口) │ └──────────┬────────────┘ │ ┌──────────▼────────────┐ │ 服务层 │ │ (消息路由、策略执行) │ └──────────┬────────────┘ │ ┌──────────▼────────────┐ │ 存储层 │ │ (多通道消息持久化) │ └───────────────────────┘2.2 数据库设计优化采用分表策略替代原SysNotice单表-- 消息主表记录通用信息 CREATE TABLE msg_message ( id BIGINT PRIMARY KEY, title VARCHAR(200), content TEXT, msg_type VARCHAR(20), priority INT, status VARCHAR(20), create_time DATETIME, update_time DATETIME ); -- 站内信扩展表 CREATE TABLE msg_website ( msg_id BIGINT PRIMARY KEY, receiver_id BIGINT, read_status BOOLEAN ); -- 邮件扩展表 CREATE TABLE msg_email ( msg_id BIGINT PRIMARY KEY, email_to VARCHAR(500), email_cc VARCHAR(500) );3. 核心功能实现3.1 消息发送流程改造// 统一发送接口示例 public interface MessageSender { SendResult send(MessageDTO message); } // 消息路由实现 Service public class MessageRouter { Autowired private MapString, MessageSender senderMap; public SendResult routeSend(MessageDTO message) { MessageSender sender senderMap.get(message.getChannel() Sender); if (sender null) { throw new IllegalArgumentException(Unsupported channel); } return sender.send(message); } }3.2 消息模板管理新增消息模板表支持变量替换// 模板处理逻辑 public class TemplateProcessor { public String process(String template, MapString, Object params) { String result template; for (Map.EntryString, Object entry : params.entrySet()) { result result.replace(${ entry.getKey() }, String.valueOf(entry.getValue())); } return result; } }4. 关键问题解决方案4.1 消息幂等性控制采用业务ID消息类型联合唯一索引ALTER TABLE msg_message ADD UNIQUE INDEX idx_biz_unique (biz_id, msg_type);配合Redis实现分布式锁public SendResult sendWithIdempotent(MessageDTO message) { String lockKey msg: message.getBizId() : message.getMsgType(); try { boolean locked redisLock.tryLock(lockKey, 10, TimeUnit.SECONDS); if (!locked) { throw new RuntimeException(Operation too frequent); } // 检查是否已存在 if (messageMapper.existsByBiz(message.getBizId(), message.getMsgType())) { return SendResult.alreadySent(); } return doSend(message); } finally { redisLock.unlock(lockKey); } }4.2 失败重试机制基于Spring Retry实现Retryable(value MessageException.class, maxAttempts 3, backoff Backoff(delay 1000, multiplier 2)) public SendResult retrySend(MessageDTO message) { // 发送实现 }5. 性能优化实践5.1 批量消息处理Transactional public BatchResult batchSend(ListMessageDTO messages) { // 批量入库 messageMapper.batchInsert(messages); // 分组并行发送 MapString, ListMessageDTO grouped messages.stream() .collect(Collectors.groupingBy(MessageDTO::getChannel)); ListCompletableFutureSendResult futures new ArrayList(); grouped.forEach((channel, list) - { futures.add(CompletableFuture.supplyAsync( () - senderMap.get(channel Sender).batchSend(list), channelThreadPools.get(channel) )); }); // 结果合并 return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])) .thenApply(v - futures.stream() .map(CompletableFuture::join) .collect(BatchResult.collector())) .join(); }5.2 读写分离实现配置多数据源# application.yml spring: datasource: write: url: jdbc:mysql://master:3306/msg_center read: url: jdbc:mysql://slave:3306/msg_center通过AOP实现自动路由Around(annotation(readOnly)) public Object route(ProceedingJoinPoint jp, ReadOnly readOnly) { String key readOnly.value() ? read : write; DynamicDataSourceHolder.setDataSourceKey(key); try { return jp.proceed(); } finally { DynamicDataSourceHolder.clear(); } }6. 监控与运维方案6.1 消息状态看板Scheduled(fixedRate 60000) public void collectMetrics() { MapString, Long stats messageMapper.countByStatus(); stats.forEach((status, count) - { metrics.gauge(message.status. status, count); }); }6.2 死信处理配置死信队列Bean public Queue deadLetterQueue() { return QueueBuilder.durable(msg.dlq) .withArgument(x-dead-letter-exchange, msg.retry.exchange) .withArgument(x-dead-letter-routing-key, msg.retry) .build(); }7. 迁移实施策略7.1 数据迁移方案采用双写过渡方案-- 迁移脚本示例 INSERT INTO msg_message (id, title, content, msg_type, create_time) SELECT id, title, content, SYSTEM, create_time FROM sys_notice; -- 应用层双写 Aspect public class NoticeMigrationAspect { AfterReturning(execution(* com.ruoyi.system.service.ISysNoticeService.*(..))) public void afterNoticeOperation(JoinPoint jp) { // 同步操作到新消息中心 } }7.2 灰度发布方案基于Nacos元数据配置# 新消息中心服务配置 spring: cloud: nacos: discovery: metadata: version: 2.0 env: gray通过网关路由控制Bean public RouteLocator customRouteLocator(RouteLocatorBuilder builder) { return builder.routes() .route(message-center, r - r .header(X-Gray-Version, 2.0) .uri(lb://message-center-v2)) .route(message-center-legacy, r - r .uri(lb://message-center-v1)) .build(); }8. 实际效果对比改造前后关键指标对比指标项原系统新系统吞吐量500 TPS3000 TPS平均延迟200ms50ms扩展新渠道周期2人日0.5人日消息可达率98.5%99.99%运维复杂度高中9. 典型问题排查记录9.1 消息重复消费现象部分消息被重复发送排查检查幂等控制日志发现biz_id生成规则有问题部分业务未正确设置biz_id解决// 自动生成biz_id的AOP实现 Before(annotation(messageApi)) public void generateBizId(JoinPoint jp) { Object arg jp.getArgs()[0]; if (arg instanceof MessageDTO) { MessageDTO message (MessageDTO) arg; if (StringUtils.isEmpty(message.getBizId())) { message.setBizId(SnowflakeId.nextId()); } } }9.2 通道阻塞问题现象邮件通道积压影响其他消息解决// 通道隔离线程池配置 Bean public Executor emailExecutor() { return new ThreadPoolExecutor(5, 5, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue(1000), new ThreadPoolExecutor.CallerRunsPolicy()); }10. 扩展设计思路10.1 插件化架构定义消息通道接口规范public interface MessagePlugin { String channel(); void init(MessageConfig config); SendResult send(Message message); HealthCheckResult healthCheck(); }通过SPI机制加载ServiceLoaderMessagePlugin plugins ServiceLoader.load(MessagePlugin.class); plugins.forEach(plugin - { plugin.init(config); pluginRegistry.register(plugin); });10.2 智能路由策略基于规则引擎实现public class SmartRouter { Autowired private RuleEngine ruleEngine; public String determineChannel(MessageDTO message) { RuleContext context new RuleContext(); context.put(message, message); context.put(user, userService.getById(message.getUserId())); ruleEngine.evaluate(channel-rule, context); return context.getResult(); } }