企业官网建设流程全解析

1. 项目背景与需求分析

在RuoYi-Cloud微服务架构中，消息通知模块最初采用SysNotice表结构实现，这种设计存在明显的局限性。随着业务复杂度提升，系统需要处理多种消息类型（站内信、邮件、短信、企业微信等），原有的单表结构难以满足以下需求：

1. 消息类型扩展性差：新增消息渠道需修改表结构和代码
2. 发送策略单一：缺乏优先级、重试机制等高级功能
3. 状态追踪困难：无法有效监控消息生命周期
4. 性能瓶颈：高并发场景下单表压力过大

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 Map<String, 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, Map<String, Object> params) { String result = template; for (Map.Entry<String, 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(List<MessageDTO> messages) { // 批量入库 messageMapper.batchInsert(messages); // 分组并行发送 Map<String, List<MessageDTO>> grouped = messages.stream() .collect(Collectors.groupingBy(MessageDTO::getChannel)); List<CompletableFuture<SendResult>> 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() { Map<String, 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. 实际效果对比

改造前后关键指标对比：

9. 典型问题排查记录

9.1 消息重复消费

现象：部分消息被重复发送排查：

1. 检查幂等控制日志发现biz_id生成规则有问题
2. 部分业务未正确设置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机制加载：

ServiceLoader<MessagePlugin> 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(); } }