// 示例：基于RocketMQ的缓冲队列设计
@Component
public class NotificationBufferQueue {
    
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    
    // 消息生产者
    public void produceNotification(NotificationMessage message) {
        // 设置消息优先级和延迟时间
        Message<NotificationMessage> mqMessage = MessageBuilder
            .withPayload(message)
            .setHeader("priority", message.getPriority())
            .setHeader("delayTime", calculateDelayTime(message))
            .build();
            
        // 发送到缓冲队列
        rocketMQTemplate.syncSend("notification_buffer_topic", mqMessage);
    }
    
    // 延迟时间计算
    private int calculateDelayTime(NotificationMessage message) {
        // 根据当前系统负载和消息优先级计算延迟时间
        int baseDelay = 1000; // 基础延迟1秒
        int loadFactor = getCurrentSystemLoad();
        int priorityFactor = message.getPriority().ordinal();
        
        return baseDelay + (loadFactor * priorityFactor);
    }
}

关键设计要点：

队列容量：合理设置队列容量避免内存溢出
消息优先级：支持消息优先级确保重要消息优先处理
延迟策略：根据系统负载动态调整消息处理延迟
监控告警：实时监控队列长度和处理速度

限流控制

// 示例：基于令牌桶算法的限流控制
@Component
public class NotificationRateLimiter {
    
    private final Map<String, RateLimiter> limiters = new ConcurrentHashMap<>();
    
    public boolean tryAcquire(String channelId) {
        RateLimiter limiter = limiters.computeIfAbsent(channelId, 
            key -> RateLimiter.create(getChannelRateLimit(key)));
        return limiter.tryAcquire();
    }
    
    private double getChannelRateLimit(String channelId) {
        // 根据通道类型和配置获取限流速率
        switch (channelId) {
            case "sms":
                return 1000.0; // 每秒1000条
            case "email":
                return 500.0;  // 每秒500条
            default:
                return 100.0;  // 默认每秒100条
        }
    }
}

关键设计要点：

通道限流：根据不同通道特性设置不同的限流策略
动态调整：根据实时监控数据动态调整限流参数
优先级保障：为高优先级消息保留处理能力
熔断机制：在通道异常时触发熔断保护

削峰效果评估

性能指标监控

关键监控指标：

队列长度：实时监控消息队列的长度变化
处理延迟：监控消息从入队到处理完成的延迟时间
吞吐量：监控系统的消息处理吞吐量
成功率：监控消息处理的成功率

容量规划

# 示例：消息队列容量规划配置
notification:
  queue:
    # 缓冲队列配置
    buffer:
      topic: notification_buffer_topic
      max_size: 1000000  # 最大队列长度
      retention: 7200000 # 消息保留时间(2小时)
    
    # 处理队列配置
    processing:
      topic: notification_processing_topic
      max_size: 500000   # 最大队列长度
      retention: 3600000 # 消息保留时间(1小时)
    
    # 死信队列配置
    dead_letter:
      topic: notification_dead_letter_topic
      max_size: 100000   # 最大队列长度
      retention: 86400000 # 消息保留时间(24小时)

关键规划要点：

容量评估：根据历史数据和业务预测评估队列容量
保留策略：合理设置消息保留时间和清理策略
扩展能力：确保队列具备水平扩展能力
灾备方案：制定队列故障时的应急处理方案

异步化处理机制

异步化处理是提升系统性能和用户体验的重要手段：

异步处理架构

生产者-消费者模式

// 示例：基于Kafka的异步处理架构
@Component
public class AsyncNotificationProcessor {
    
    @Autowired
    private KafkaTemplate<String, NotificationMessage> kafkaTemplate;
    
    @Autowired
    private NotificationService notificationService;
    
    // 消息生产者
    public void sendAsyncNotification(NotificationRequest request) {
        NotificationMessage message = buildNotificationMessage(request);
        
        // 异步发送到Kafka
        ListenableFuture<SendResult<String, NotificationMessage>> future = 
            kafkaTemplate.send("notification_topic", message.getId(), message);
        
        // 添加发送回调
        future.addCallback(new ListenableFutureCallback<SendResult<String, NotificationMessage>>() {
            @Override
            public void onSuccess(SendResult<String, NotificationMessage> result) {
                log.info("消息发送成功: {}", message.getId());
            }
            
            @Override
            public void onFailure(Throwable ex) {
                log.error("消息发送失败: {}", message.getId(), ex);
                // 失败处理逻辑
                handleSendFailure(message, ex);
            }
        });
    }
    
    // 消息消费者
    @KafkaListener(topics = "notification_topic", groupId = "notification_processor")
    public void processNotification(ConsumerRecord<String, NotificationMessage> record) {
        NotificationMessage message = record.value();
        try {
            // 处理通知发送
            notificationService.sendNotification(message);
            log.info("通知处理完成: {}", message.getId());
        } catch (Exception e) {
            log.error("通知处理失败: {}", message.getId(), e);
            // 失败重试或进入死信队列
            handleProcessFailure(message, e);
        }
    }
}

关键设计要点：

解耦设计：生产者和消费者完全解耦
异步处理：用户请求快速响应，后台异步处理
容错机制：完善的失败处理和重试机制
监控告警：实时监控处理状态和性能指标

批量处理优化

// 示例：批量处理优化
@Component
public class BatchNotificationProcessor {
    
    @Autowired
    private NotificationService notificationService;
    
    private final List<NotificationMessage> batchBuffer = new ArrayList<>();
    private final Object bufferLock = new Object();
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
    
    public BatchNotificationProcessor() {
        // 定时批量处理
        scheduler.scheduleAtFixedRate(this::processBatch, 0, 100, TimeUnit.MILLISECONDS);
    }
    
    public void addNotification(NotificationMessage message) {
        synchronized (bufferLock) {
            batchBuffer.add(message);
            // 达到批量大小立即处理
            if (batchBuffer.size() >= 100) {
                processBatch();
            }
        }
    }
    
    private void processBatch() {
        List<NotificationMessage> batch;
        synchronized (bufferLock) {
            if (batchBuffer.isEmpty()) {
                return;
            }
            batch = new ArrayList<>(batchBuffer);
            batchBuffer.clear();
        }
        
        try {
            // 批量处理通知
            notificationService.sendNotifications(batch);
            log.info("批量处理完成，处理数量: {}", batch.size());
        } catch (Exception e) {
            log.error("批量处理失败，处理数量: {}", batch.size(), e);
            // 失败处理
            handleBatchFailure(batch, e);
        }
    }
}

关键优化要点：

批量聚合：将多个小请求聚合成批量处理
定时触发：定时检查并处理批量数据
大小控制：合理控制批量处理的大小
失败隔离：单个失败不影响整个批次

异步处理优势

性能提升

关键性能优势：

响应时间：用户请求响应时间显著缩短
吞吐量：系统整体吞吐量大幅提升
资源利用：系统资源利用效率提高
并发处理：支持更高的并发处理能力

用户体验

关键体验优势：

快速响应：用户操作得到快速响应
系统稳定：系统在高负载下保持稳定
错误隔离：单个请求失败不影响其他请求
进度追踪：支持异步处理进度追踪

最终一致性保障

在分布式系统中，最终一致性是平衡性能和数据一致性的有效方案：

一致性模型选择

事件驱动架构

// 示例：基于事件驱动的最终一致性实现
@Component
public class EventDrivenConsistencyManager {
    
    @Autowired
    private ApplicationEventPublisher eventPublisher;
    
    // 发送通知并发布事件
    public NotificationResult sendNotificationWithEvent(NotificationRequest request) {
        try {
            // 发送通知
            NotificationResult result = notificationService.send(request);
            
            // 发布通知发送事件
            NotificationSentEvent event = new NotificationSentEvent(
                request.getId(), 
                request.getReceiver(), 
                result.getStatus(),
                System.currentTimeMillis()
            );
            eventPublisher.publishEvent(event);
            
            return result;
        } catch (Exception e) {
            // 发布通知失败事件
            NotificationFailedEvent failedEvent = new NotificationFailedEvent(
                request.getId(),
                request.getReceiver(),
                e.getMessage(),
                System.currentTimeMillis()
            );
            eventPublisher.publishEvent(failedEvent);
            
            throw new NotificationException("通知发送失败", e);
        }
    }
    
    // 事件监听器
    @EventListener
    public void handleNotificationSent(NotificationSentEvent event) {
        // 更新业务系统状态
        businessService.updateNotificationStatus(event.getMessageId(), event.getStatus());
        
        // 记录审计日志
        auditService.recordNotificationEvent(event);
        
        // 触发后续业务逻辑
        triggerNextBusinessProcess(event);
    }
}

关键设计要点：

事件发布：在关键操作后发布业务事件
事件监听：其他服务监听并处理相关事件
状态同步：通过事件实现跨服务状态同步
补偿机制：提供事件处理失败的补偿机制

Saga模式实现

// 示例：基于Saga模式的分布式事务实现
@Component
public class NotificationSagaOrchestrator {
    
    @Autowired
    private SagaExecutionEngine sagaEngine;
    
    public SagaInstance startNotificationSaga(NotificationRequest request) {
        SagaDefinition<NotificationSagaData> sagaDefinition = step()
            .invokeLocal("validate", this::validateRequest)
            .step()
            .invokeParticipant("send", notificationService::send, notificationService::compensate)
            .step()
            .invokeParticipant("record", auditService::record, auditService::compensate)
            .step()
            .invokeLocal("complete", this::completeSaga)
            .build();
            
        NotificationSagaData sagaData = new NotificationSagaData(request);
        return sagaEngine.create(sagaDefinition, sagaData);
    }
    
    private void validateRequest(NotificationSagaData data) {
        // 验证请求参数
        if (!validator.validate(data.getRequest())) {
            throw new ValidationException("请求参数验证失败");
        }
    }
    
    private void completeSaga(NotificationSagaData data) {
        // 完成Saga，更新最终状态
        data.setCompleted(true);
        log.info("通知Saga完成: {}", data.getRequest().getId());
    }
}

关键实现要点：

步骤分解：将复杂操作分解为多个步骤
补偿操作：为每个步骤定义补偿操作
状态管理：管理Saga的执行状态
异常处理：在异常时执行补偿操作

一致性保障机制

幂等性设计

// 示例：幂等性设计实现
@Component
public class IdempotentNotificationService {
    
    @Autowired
    private IdempotentRepository idempotentRepository;
    
    public NotificationResult sendNotification(NotificationRequest request) {
        // 生成幂等键
        String idempotentKey = generateIdempotentKey(request);
        
        // 检查是否已处理
        Optional<NotificationResult> existingResult = 
            idempotentRepository.findResultByIdempotentKey(idempotentKey);
            
        if (existingResult.isPresent()) {
            // 已处理，直接返回结果
            return existingResult.get();
        }
        
        // 未处理，执行发送逻辑
        NotificationResult result = doSendNotification(request);
        
        // 保存幂等记录
        IdempotentRecord record = new IdempotentRecord(
            idempotentKey, 
            request.getId(), 
            result, 
            System.currentTimeMillis()
        );
        idempotentRepository.save(record);
        
        return result;
    }
    
    private String generateIdempotentKey(NotificationRequest request) {
        // 根据业务逻辑生成幂等键
        return DigestUtils.md5DigestAsHex(
            (request.getReceiver().toString() + 
             request.getTemplateId() + 
             request.getVariables().toString()).getBytes()
        );
    }
}

关键设计要点：

幂等键生成：根据业务逻辑生成唯一幂等键
结果缓存：缓存已处理请求的结果
重复检查：处理前检查是否已处理
数据一致性：确保幂等记录与业务数据一致性

对账机制

// 示例：对账机制实现
@Component
public class NotificationReconciliationService {
    
    @Scheduled(cron = "0 0 2 * * ?") // 每天凌晨2点执行
    public void performDailyReconciliation() {
        LocalDate yesterday = LocalDate.now().minusDays(1);
        
        // 获取昨日发送记录
        List<NotificationRecord> sentRecords = 
            notificationRepository.findSentRecords(yesterday);
            
        // 获取昨日回调记录
        List<CallbackRecord> callbackRecords = 
            callbackRepository.findCallbackRecords(yesterday);
            
        // 执行对账
        ReconciliationResult result = reconcileRecords(sentRecords, callbackRecords);
        
        // 处理差异
        handleReconciliationDifferences(result.getDifferences());
        
        // 记录对账结果
        reconciliationRepository.save(result);
        
        log.info("对账完成，总计: {}, 差异: {}", 
            result.getTotalCount(), result.getDifferenceCount());
    }
    
    private ReconciliationResult reconcileRecords(
        List<NotificationRecord> sentRecords, 
        List<CallbackRecord> callbackRecords) {
        
        Map<String, NotificationRecord> sentMap = sentRecords.stream()
            .collect(Collectors.toMap(NotificationRecord::getMessageId, record -> record));
            
        Map<String, CallbackRecord> callbackMap = callbackRecords.stream()
            .collect(Collectors.toMap(CallbackRecord::getMessageId, record -> record));
            
        List<ReconciliationDifference> differences = new ArrayList<>();
        
        // 检查发送记录与回调记录的差异
        for (NotificationRecord sentRecord : sentRecords) {
            CallbackRecord callbackRecord = callbackMap.get(sentRecord.getMessageId());
            if (callbackRecord == null || 
                !sentRecord.getStatus().equals(callbackRecord.getStatus())) {
                differences.add(new ReconciliationDifference(
                    sentRecord.getMessageId(),
                    sentRecord.getStatus(),
                    callbackRecord != null ? callbackRecord.getStatus() : null
                ));
            }
        }
        
        return new ReconciliationResult(sentRecords.size(), differences);
    }
}

关键实现要点：

定时执行：定期执行对账任务
数据比对：比对不同数据源的数据一致性
差异处理：自动处理或人工干预差异数据
结果记录：记录对账结果用于分析和审计

消息队列选型与配置

合理选型和配置消息队列是保障系统稳定性的关键：

消息队列选型

Kafka vs RocketMQ vs RabbitMQ

特性	Kafka	RocketMQ	RabbitMQ
吞吐量	极高	高	中等
延迟	低	低	中等
可靠性	高	高	高
功能丰富度	中等	高	高
运维复杂度	高	中等	低
社区活跃度	高	中等	高

选型建议：

Kafka：适用于大数据处理和日志收集场景
RocketMQ：适用于金融级高可靠场景
RabbitMQ：适用于功能丰富的企业级应用

配置优化

# 示例：RocketMQ配置优化
rocketmq:
  name-server: 192.168.1.100:9876;192.168.1.101:9876
  producer:
    group: notification_producer_group
    send-message-timeout: 3000
    compress-message-body-threshold: 4096
    retry-times-when-send-failed: 2
    retry-times-when-send-async-failed: 2
  consumer:
    group: notification_consumer_group
    consume-thread-min: 20
    consume-thread-max: 64
    consume-message-batch-max-size: 32
    pull-batch-size: 32
    pull-interval: 1000