背景：消息消费的动态挑战

在分布式系统中，消息队列的消费速率直接影响系统的整体性能和稳定性。然而，实际生产环境中，消息流量往往是动态变化的：

• 流量低谷：消息量少，消费速度过快可能导致系统资源浪费
• 流量高峰：消息量突增，消费速度过慢可能导致队列积压
• 突发流量：短时间内大量消息涌入，需要快速响应
• 系统负载：不同时段系统负载不同，需要动态调整消费策略

传统的消息消费模式通常采用固定的批量大小和消费速率，无法适应这种动态变化的场景。当流量高峰来临时，固定的批量大小可能导致处理能力不足，队列积压严重；而在流量低谷时，又会造成系统资源的浪费。

本文将介绍如何使用 SpringBoot 实现消息消费速率自适应和动态批量处理，让系统能够根据实时流量自动调整批量大小，实现平滑处理。

核心概念

1. 消息消费速率

消息消费速率是指单位时间内消费的消息数量，通常以 QPS（Queries Per Second）来衡量。消费速率的大小直接影响消息处理的速度和系统资源的使用。

2. 动态批量处理

动态批量处理是指根据当前的系统状态和消息流量，自动调整每次消费的消息批量大小。在流量高峰时增加批量大小，提高处理能力；在流量低谷时减少批量大小，节省系统资源。

3. 自适应调整策略

4. 批量大小范围

技术实现

1. 核心实体类

// 消息实体
@Data
@Entity
@Table(name = "message")
public class Message {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
    
    @Column(name = "message_id", unique = true, nullable = false, length = 64)
    private String messageId; // 消息ID
    
    @Column(name = "topic", nullable = false, length = 128)
    private String topic; // 消息主题
    
    @Column(name = "content", columnDefinition = "TEXT", nullable = false)
    private String content; // 消息内容
    
    @Column(name = "status", nullable = false, length = 32)
    private String status = "PENDING"; // 状态：PENDING/PROCESSING/SUCCESS/FAILED
    
    @Column(name = "retry_count", nullable = false)
    private Integer retryCount = 0; // 重试次数
    
    @Column(name = "max_retry", nullable = false)
    private Integer maxRetry = 3; // 最大重试次数
    
    @Column(name = "error_message", length = 1024)
    private String errorMessage; // 错误信息
    
    @CreationTimestamp
    @Column(name = "create_time", nullable = false)
    private Date createTime;
    
    @UpdateTimestamp
    @Column(name = "update_time", nullable = false)
    private Date updateTime;
    
    @Column(name = "process_time")
    private Date processTime; // 处理时间
    
    @Column(name = "process_duration")
    private Long processDuration; // 处理耗时（毫秒）
}

// 消费速率统计
@Data
@Entity
@Table(name = "consumption_rate_statistics")
public class ConsumptionRateStatistics {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
    
    @Column(name = "timestamp", nullable = false)
    private Date timestamp; // 统计时间
    
    @Column(name = "messages_processed", nullable = false)
    private Integer messagesProcessed; // 处理消息数
    
    @Column(name = "batch_size", nullable = false)
    private Integer batchSize; // 批量大小
    
    @Column(name = "processing_time", nullable = false)
    private Long processingTime; // 处理时间（毫秒）
    
    @Column(name = "queue_length", nullable = false)
    private Long queueLength; // 队列长度
    
    @Column(name = "system_load", nullable = false)
    private Double systemLoad; // 系统负载
    
    @Column(name = "cpu_usage", nullable = false)
    private Double cpuUsage; // CPU 使用率
    
    @Column(name = "memory_usage", nullable = false)
    private Double memoryUsage; // 内存使用率
}

// 批量大小配置
@Data
@Entity
@Table(name = "batch_size_config")
public class BatchSizeConfig {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
    
    @Column(name = "min_batch_size", nullable = false)
    private Integer minBatchSize = 1; // 最小批量大小
    
    @Column(name = "max_batch_size", nullable = false)
    private Integer maxBatchSize = 100; // 最大批量大小
    
    @Column(name = "default_batch_size", nullable = false)
    private Integer defaultBatchSize = 10; // 默认批量大小
    
    @Column(name = "queue_threshold_low", nullable = false)
    private Long queueThresholdLow = 100L; // 队列长度低阈值
    
    @Column(name = "queue_threshold_medium", nullable = false)
    private Long queueThresholdMedium = 500L; // 队列长度中阈值
    
    @Column(name = "queue_threshold_high", nullable = false)
    private Long queueThresholdHigh = 1000L; // 队列长度高阈值
    
    @Column(name = "system_load_threshold", nullable = false)
    private Double systemLoadThreshold = 0.7; // 系统负载阈值
    
    @Column(name = "cpu_usage_threshold", nullable = false)
    private Double cpuUsageThreshold = 70.0; // CPU 使用率阈值
    
    @Column(name = "memory_usage_threshold", nullable = false)
    private Double memoryUsageThreshold = 80.0; // 内存使用率阈值
    
    @Column(name = "adjustment_factor", nullable = false)
    private Double adjustmentFactor = 1.5; // 调整因子
    
    @UpdateTimestamp
    @Column(name = "update_time", nullable = false)
    private Date updateTime;
}

2. 消息消费速率自适应服务

@Service
@Slf4j
public class AdaptiveConsumptionRateService {
    
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    @Autowired
    private ConsumptionRateStatisticsRepository statisticsRepository;
    
    @Autowired
    private BatchSizeConfigRepository batchSizeConfigRepository;
    
    @Value("${message.queue.name}")
    private String messageQueueName;
    
    @Value("${message.consumption.rate.window:60000}")
    private Long rateWindow; // 速率统计窗口（毫秒）
    
    @Value("${message.consumption.rate.history:10}")
    private Integer rateHistory; // 历史记录数量
    
    private final AtomicInteger currentBatchSize = new AtomicInteger(10);
    private final AtomicLong lastAdjustmentTime = new AtomicLong(0);
    private final AtomicInteger messagesProcessed = new AtomicInteger(0);
    private final AtomicLong processingTime = new AtomicLong(0);
    
    private final List<ConsumptionRate> rateHistoryList = new CopyOnWriteArrayList<>();
    
    /**
     * 获取当前批量大小
     */
    public int getCurrentBatchSize() {
        adjustBatchSizeIfNeeded();
        return currentBatchSize.get();
    }
    
    /**
     * 调整批量大小
     */
    private void adjustBatchSizeIfNeeded() {
        long currentTime = System.currentTimeMillis();
        if (currentTime - lastAdjustmentTime.get() < 5000) { // 5秒内不重复调整
            return;
        }
        
        BatchSizeConfig config = getBatchSizeConfig();
        if (config == null) {
            return;
        }
        
        // 获取当前队列长度
        long queueLength = getQueueLength();
        
        // 获取系统负载
        SystemMetrics metrics = getSystemMetrics();
        
        // 计算推荐批量大小
        int recommendedBatchSize = calculateRecommendedBatchSize(queueLength, metrics, config);
        
        // 调整批量大小
        if (recommendedBatchSize != currentBatchSize.get()) {
            currentBatchSize.set(recommendedBatchSize);
            lastAdjustmentTime.set(currentTime);
            log.info("Adjusted batch size to: {}, queue length: {}, system load: {}, CPU: {}%, memory: {}%",
                    recommendedBatchSize, queueLength, metrics.getSystemLoad(), metrics.getCpuUsage(), metrics.getMemoryUsage());
        }
    }
    
    /**
     * 计算推荐批量大小
     */
    private int calculateRecommendedBatchSize(long queueLength, SystemMetrics metrics, BatchSizeConfig config) {
        int baseBatchSize = config.getDefaultBatchSize();
        
        // 根据队列长度调整
        if (queueLength > config.getQueueThresholdHigh()) {
            baseBatchSize = config.getMaxBatchSize();
        } else if (queueLength > config.getQueueThresholdMedium()) {
            baseBatchSize = (config.getDefaultBatchSize() + config.getMaxBatchSize()) / 2;
        } else if (queueLength > config.getQueueThresholdLow()) {
            baseBatchSize = config.getDefaultBatchSize();
        } else {
            baseBatchSize = config.getMinBatchSize();
        }
        
        // 根据系统负载调整
        if (metrics.getSystemLoad() > config.getSystemLoadThreshold() ||
            metrics.getCpuUsage() > config.getCpuUsageThreshold() ||
            metrics.getMemoryUsage() > config.getMemoryUsageThreshold()) {
            // 系统负载高，减小批量大小
            baseBatchSize = Math.max(config.getMinBatchSize(), (int)(baseBatchSize / config.getAdjustmentFactor()));
        }
        
        // 检查历史消费速率
        if (!rateHistoryList.isEmpty()) {
            ConsumptionRate recentRate = rateHistoryList.get(rateHistoryList.size() - 1);
            if (recentRate.getMessagesPerSecond() > 100) { // 消费速率高
                baseBatchSize = Math.min(config.getMaxBatchSize(), (int)(baseBatchSize * config.getAdjustmentFactor()));
            } else if (recentRate.getMessagesPerSecond() < 10) { // 消费速率低
                baseBatchSize = Math.max(config.getMinBatchSize(), (int)(baseBatchSize / config.getAdjustmentFactor()));
            }
        }
        
        // 确保在合理范围内
        return Math.max(config.getMinBatchSize(), Math.min(config.getMaxBatchSize(), baseBatchSize));
    }
    
    /**
     * 获取队列长度
     */
    private long getQueueLength() {
        return redisTemplate.opsForList().size(messageQueueName);
    }
    
    /**
     * 获取系统指标
     */
    private SystemMetrics getSystemMetrics() {
        SystemMetrics metrics = new SystemMetrics();
        
        // 获取系统负载
        OperatingSystemMXBean osBean = ManagementFactory.getOperatingSystemMXBean();
        metrics.setSystemLoad(osBean.getSystemLoadAverage());
        
        // 获取CPU使用率
        com.sun.management.OperatingSystemMXBean sunOsBean = (com.sun.management.OperatingSystemMXBean) osBean;
        metrics.setCpuUsage(sunOsBean.getProcessCpuLoad() * 100);
        
        // 获取内存使用率
        MemoryMXBean memoryBean = ManagementFactory.getMemoryMXBean();
        long totalMemory = memoryBean.getHeapMemoryUsage().getTotal();
        long usedMemory = memoryBean.getHeapMemoryUsage().getUsed();
        metrics.setMemoryUsage((double) usedMemory / totalMemory * 100);
        
        return metrics;
    }
    
    /**
     * 记录消费统计
     */
    public void recordConsumption(int batchSize, int messagesProcessed, long processingTime) {
        this.messagesProcessed.addAndGet(messagesProcessed);
        this.processingTime.addAndGet(processingTime);
        
        // 定期统计
        long currentTime = System.currentTimeMillis();
        if (currentTime - lastAdjustmentTime.get() > rateWindow) {
            saveConsumptionStatistics(batchSize, currentTime);
            updateRateHistory();
            resetCounters();
        }
    }
    
    /**
     * 保存消费统计
     */
    private void saveConsumptionStatistics(int batchSize, long timestamp) {
        ConsumptionRateStatistics statistics = new ConsumptionRateStatistics();
        statistics.setTimestamp(new Date(timestamp));
        statistics.setMessagesProcessed(messagesProcessed.get());
        statistics.setBatchSize(batchSize);
        statistics.setProcessingTime(processingTime.get());
        statistics.setQueueLength(getQueueLength());
        
        SystemMetrics metrics = getSystemMetrics();
        statistics.setSystemLoad(metrics.getSystemLoad());
        statistics.setCpuUsage(metrics.getCpuUsage());
        statistics.setMemoryUsage(metrics.getMemoryUsage());
        
        statisticsRepository.save(statistics);
    }
    
    /**
     * 更新速率历史
     */
    private void updateRateHistory() {
        double messagesPerSecond = messagesProcessed.get() * 1000.0 / processingTime.get();
        ConsumptionRate rate = new ConsumptionRate(System.currentTimeMillis(), messagesPerSecond);
        
        rateHistoryList.add(rate);
        if (rateHistoryList.size() > rateHistory) {
            rateHistoryList.remove(0);
        }
    }
    
    /**
     * 重置计数器
     */
    private void resetCounters() {
        messagesProcessed.set(0);
        processingTime.set(0);
    }
    
    /**
     * 获取批量大小配置
     */
    private BatchSizeConfig getBatchSizeConfig() {
        List<BatchSizeConfig> configs = batchSizeConfigRepository.findAll();
        if (configs.isEmpty()) {
            // 创建默认配置
            BatchSizeConfig config = new BatchSizeConfig();
            return batchSizeConfigRepository.save(config);
        }
        return configs.get(0);
    }
    
    /**
     * 系统指标
     */
    @Data
    private static class SystemMetrics {
        private double systemLoad;
        private double cpuUsage;
        private double memoryUsage;
    }
    
    /**
     * 消费速率
     */
    @Data
    private static class ConsumptionRate {
        private long timestamp;
        private double messagesPerSecond;
        
        public ConsumptionRate(long timestamp, double messagesPerSecond) {
            this.timestamp = timestamp;
            this.messagesPerSecond = messagesPerSecond;
        }
    }
}

3. 动态批量处理服务

@Service
@Slf4j
public class DynamicBatchProcessingService {
    
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    @Autowired
    private AdaptiveConsumptionRateService consumptionRateService;
    
    @Autowired
    private MessageRepository messageRepository;
    
    @Value("${message.queue.name}")
    private String messageQueueName;
    
    @Value("${message.batch.processing.timeout:5000}")
    private Long batchProcessingTimeout; // 批量处理超时时间（毫秒）
    
    /**
     * 批量消费消息
     */
    public List<Message> consumeBatchMessages() {
        int batchSize = consumptionRateService.getCurrentBatchSize();
        return consumeBatchMessages(batchSize);
    }
    
    /**
     * 批量消费消息
     */
    public List<Message> consumeBatchMessages(int batchSize) {
        List<Message> messages = new ArrayList<>();
        long startTime = System.currentTimeMillis();
        
        try {
            // 从队列中批量获取消息
            List<Object> messageJsons = redisTemplate.opsForList().rightPop(messageQueueName, batchSize);
            if (messageJsons == null || messageJsons.isEmpty()) {
                return messages;
            }
            
            // 解析消息
            for (Object messageJson : messageJsons) {
                try {
                    Message message = JSON.parseObject(messageJson.toString(), Message.class);
                    if (message != null) {
                        messages.add(message);
                    }
                } catch (Exception e) {
                    log.error("Failed to parse message: {}", messageJson, e);
                }
            }
            
            // 处理消息
            if (!messages.isEmpty()) {
                processMessages(messages);
                
                // 记录消费统计
                long processingTime = System.currentTimeMillis() - startTime;
                consumptionRateService.recordConsumption(batchSize, messages.size(), processingTime);
            }
        } catch (Exception e) {
            log.error("Failed to consume batch messages", e);
        }
        
        return messages;
    }
    
    /**
     * 处理消息
     */
    private void processMessages(List<Message> messages) {
        CompletableFuture[] futures = messages.stream()
                .map(message -> CompletableFuture.runAsync(() -> processMessage(message)))
                .toArray(CompletableFuture[]::new);
        
        CompletableFuture.allOf(futures).join();
    }
    
    /**
     * 处理单个消息
     */
    private void processMessage(Message message) {
        try {
            // 更新消息状态为处理中
            message.setStatus("PROCESSING");
            message.setProcessTime(new Date());
            messageRepository.save(message);
            
            // 处理消息逻辑
            long processStartTime = System.currentTimeMillis();
            doProcessMessage(message);
            long processDuration = System.currentTimeMillis() - processStartTime;
            
            // 更新消息状态为成功
            message.setStatus("SUCCESS");
            message.setProcessDuration(processDuration);
        } catch (Exception e) {
            log.error("Failed to process message: {}", message.getMessageId(), e);
            
            // 更新消息状态为失败
            message.setStatus("FAILED");
            message.setErrorMessage(e.getMessage());
            message.setRetryCount(message.getRetryCount() + 1);
            
            // 重试逻辑
            if (message.getRetryCount() < message.getMaxRetry()) {
                // 重新发送到队列
                message.setStatus("PENDING");
                String messageJson = JSON.toJSONString(message);
                redisTemplate.opsForList().leftPush(messageQueueName, messageJson);
            }
        } finally {
            messageRepository.save(message);
        }
    }
    
    /**
     * 实际处理消息
     */
    private void doProcessMessage(Message message) {
        // 这里实现具体的消息处理逻辑
        // 示例：根据消息主题调用不同的处理方法
        switch (message.getTopic()) {
            case "payment":
                processPaymentMessage(message);
                break;
            case "order":
                processOrderMessage(message);
                break;
            case "notification":
                processNotificationMessage(message);
                break;
            default:
                processDefaultMessage(message);
                break;
        }
    }
    
    /**
     * 处理支付消息
     */
    private void processPaymentMessage(Message message) {
        log.info("Processing payment message: {}", message.getMessageId());
        // 模拟处理时间
        simulateProcessingTime(100);
    }
    
    /**
     * 处理订单消息
     */
    private void processOrderMessage(Message message) {
        log.info("Processing order message: {}", message.getMessageId());
        // 模拟处理时间
        simulateProcessingTime(200);
    }
    
    /**
     * 处理通知消息
     */
    private void processNotificationMessage(Message message) {
        log.info("Processing notification message: {}", message.getMessageId());
        // 模拟处理时间
        simulateProcessingTime(50);
    }
    
    /**
     * 处理默认消息
     */
    private void processDefaultMessage(Message message) {
        log.info("Processing default message: {}", message.getMessageId());
        // 模拟处理时间
        simulateProcessingTime(100);
    }
    
    /**
     * 模拟处理时间
     */
    private void simulateProcessingTime(long millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
    
    /**
     * 发送消息
     */
    public void sendMessage(Message message) {
        // 保存消息到数据库
        messageRepository.save(message);
        
        // 发送到队列
        String messageJson = JSON.toJSONString(message);
        redisTemplate.opsForList().leftPush(messageQueueName, messageJson);
        
        log.info("Message sent: {}", message.getMessageId());
    }
    
    /**
     * 批量发送消息
     */
    public void sendMessages(List<Message> messages) {
        for (Message message : messages) {
            sendMessage(message);
        }
    }
    
    /**
     * 获取队列长度
     */
    public long getQueueLength() {
        return redisTemplate.opsForList().size(messageQueueName);
    }
    
    /**
     * 清空队列
     */
    public void clearQueue() {
        redisTemplate.delete(messageQueueName);
        log.info("Queue cleared");
    }
}

4. 流量监控和调整服务

@Service
@Slf4j
public class TrafficMonitoringService {
    
    @Autowired
    private DynamicBatchProcessingService batchProcessingService;
    
    @Autowired
    private AdaptiveConsumptionRateService consumptionRateService;
    
    @Autowired
    private ConsumptionRateStatisticsRepository statisticsRepository;
    
    @Value("${message.monitoring.interval:30000}")
    private Long monitoringInterval; // 监控间隔（毫秒）
    
    @Value("${message.monitoring.alert.threshold:1000}")
    private Long alertThreshold; // 告警阈值
    
    @Scheduled(fixedRateString = "${message.monitoring.interval:30000}")
    public void monitorTraffic() {
        // 获取队列长度
        long queueLength = batchProcessingService.getQueueLength();
        
        // 获取当前批量大小
        int currentBatchSize = consumptionRateService.getCurrentBatchSize();
        
        // 记录监控数据
        log.info("Traffic monitoring - Queue length: {}, Current batch size: {}", queueLength, currentBatchSize);
        
        // 检查是否需要告警
        if (queueLength > alertThreshold) {
            sendAlert(queueLength, currentBatchSize);
        }
        
        // 定期清理历史统计数据
        cleanupHistoricalData();
    }
    
    /**
     * 发送告警
     */
    private void sendAlert(long queueLength, int batchSize) {
        log.warn("Alert: Queue length ({}) exceeds threshold ({}), current batch size: {}", 
                queueLength, alertThreshold, batchSize);
        
        // 这里可以实现具体的告警逻辑，如发送邮件、短信等
        // TODO: 实现告警通知
    }
    
    /**
     * 清理历史数据
     */
    private void cleanupHistoricalData() {
        // 删除7天前的统计数据
        Calendar calendar = Calendar.getInstance();
        calendar.add(Calendar.DAY_OF_MONTH, -7);
        Date cutoffDate = calendar.getTime();
        
        List<ConsumptionRateStatistics> oldStatistics = statisticsRepository.findByTimestampBefore(cutoffDate);
        if (!oldStatistics.isEmpty()) {
            statisticsRepository.deleteAll(oldStatistics);
            log.info("Cleaned up {} old statistics records", oldStatistics.size());
        }
    }
    
    /**
     * 获取消费速率趋势
     */
    public List<ConsumptionRateTrend> getConsumptionRateTrends() {
        List<ConsumptionRateStatistics> statistics = statisticsRepository.findAll();
        List<ConsumptionRateTrend> trends = new ArrayList<>();
        
        for (ConsumptionRateStatistics stat : statistics) {
            ConsumptionRateTrend trend = new ConsumptionRateTrend();
            trend.setTimestamp(stat.getTimestamp());
            trend.setMessagesProcessed(stat.getMessagesProcessed());
            trend.setBatchSize(stat.getBatchSize());
            trend.setProcessingTime(stat.getProcessingTime());
            trend.setQueueLength(stat.getQueueLength());
            trend.setSystemLoad(stat.getSystemLoad());
            trend.setCpuUsage(stat.getCpuUsage());
            trend.setMemoryUsage(stat.getMemoryUsage());
            
            // 计算消息处理速率
            if (stat.getProcessingTime() > 0) {
                double rate = stat.getMessagesProcessed() * 1000.0 / stat.getProcessingTime();
                trend.setMessagesPerSecond(rate);
            }
            
            trends.add(trend);
        }
        
        return trends;
    }
    
    /**
     * 消费速率趋势
     */
    @Data
    public static class ConsumptionRateTrend {
        private Date timestamp;
        private Integer messagesProcessed;
        private Integer batchSize;
        private Long processingTime;
        private Long queueLength;
        private Double systemLoad;
        private Double cpuUsage;
        private Double memoryUsage;
        private Double messagesPerSecond;
    }
}

5. 消息管理控制器

@RestController
@RequestMapping("/api/message")
@Slf4j
public class MessageController {
    
    @Autowired
    private DynamicBatchProcessingService batchProcessingService;
    
    @Autowired
    private AdaptiveConsumptionRateService consumptionRateService;
    
    @Autowired
    private TrafficMonitoringService trafficMonitoringService;
    
    @Autowired
    private MessageRepository messageRepository;
    
    @Autowired
    private BatchSizeConfigRepository batchSizeConfigRepository;
    
    /**
     * 发送消息
     */
    @PostMapping("/send")
    public Result<String> sendMessage(@RequestBody MessageRequest request) {
        Message message = createMessage(request);
        batchProcessingService.sendMessage(message);
        return Result.success(message.getMessageId());
    }
    
    /**
     * 批量发送消息
     */
    @PostMapping("/send/batch")
    public Result<String> sendMessagesBatch(@RequestBody List<MessageRequest> requests) {
        List<Message> messages = new ArrayList<>();
        for (MessageRequest request : requests) {
            Message message = createMessage(request);
            messages.add(message);
        }
        batchProcessingService.sendMessages(messages);
        return Result.success("批量消息发送成功");
    }
    
    /**
     * 消费消息（自动批量大小）
     */
    @PostMapping("/consume")
    public Result<String> consumeMessage() {
        List<Message> messages = batchProcessingService.consumeBatchMessages();
        return Result.success("消费了 " + messages.size() + " 条消息");
    }
    
    /**
     * 消费消息（指定批量大小）
     */
    @PostMapping("/consume/batch")
    public Result<String> consumeMessagesBatch(@RequestParam int batchSize) {
        List<Message> messages = batchProcessingService.consumeBatchMessages(batchSize);
        return Result.success("消费了 " + messages.size() + " 条消息");
    }
    
    /**
     * 获取队列状态
     */
    @GetMapping("/queue/status")
    public Result<QueueStatus> getQueueStatus() {
        long queueLength = batchProcessingService.getQueueLength();
        int currentBatchSize = consumptionRateService.getCurrentBatchSize();
        
        QueueStatus status = new QueueStatus();
        status.setQueueLength(queueLength);
        status.setCurrentBatchSize(currentBatchSize);
        
        return Result.success(status);
    }
    
    /**
     * 获取批量大小配置
     */
    @GetMapping("/batch/config")
    public Result<BatchSizeConfig> getBatchSizeConfig() {
        List<BatchSizeConfig> configs = batchSizeConfigRepository.findAll();
        if (configs.isEmpty()) {
            return Result.error("配置不存在");
        }
        return Result.success(configs.get(0));
    }
    
    /**
     * 更新批量大小配置
     */
    @PutMapping("/batch/config")
    public Result<String> updateBatchSizeConfig(@RequestBody BatchSizeConfig config) {
        batchSizeConfigRepository.save(config);
        return Result.success("配置更新成功");
    }
    
    /**
     * 获取消费速率趋势
     */
    @GetMapping("/trends")
    public Result<List<TrafficMonitoringService.ConsumptionRateTrend>> getConsumptionTrends() {
        List<TrafficMonitoringService.ConsumptionRateTrend> trends = trafficMonitoringService.getConsumptionRateTrends();
        return Result.success(trends);
    }
    
    /**
     * 获取消息列表
     */
    @GetMapping("/list")
    public Result<List<Message>> getMessages() {
        List<Message> messages = messageRepository.findAll();
        return Result.success(messages);
    }
    
    /**
     * 获取消息详情
     */
    @GetMapping("/{messageId}")
    public Result<Message> getMessage(@PathVariable String messageId) {
        Optional<Message> messageOptional = messageRepository.findByMessageId(messageId);
        if (!messageOptional.isPresent()) {
            return Result.error("消息不存在");
        }
        return Result.success(messageOptional.get());
    }
    
    /**
     * 清空队列
     */
    @PostMapping("/queue/clear")
    public Result<String> clearQueue() {
        batchProcessingService.clearQueue();
        return Result.success("队列已清空");
    }
    
    /**
     * 创建消息
     */
    private Message createMessage(MessageRequest request) {
        Message message = new Message();
        message.setMessageId(UUID.randomUUID().toString());
        message.setTopic(request.getTopic());
        message.setContent(request.getContent());
        message.setStatus("PENDING");
        message.setRetryCount(0);
        message.setMaxRetry(3);
        return message;
    }
    
    /**
     * 消息请求
     */
    @Data
    public static class MessageRequest {
        private String topic;
        private String content;
    }
    
    /**
     * 队列状态
     */
    @Data
    public static class QueueStatus {
        private long queueLength;
        private int currentBatchSize;
    }
}

核心流程

1. 消息发送流程

1. 接收发送请求：客户端发送消息请求
2. 创建消息：根据请求创建消息对象
3. 保存消息：将消息保存到数据库
4. 发送到队列：将消息发送到 Redis 队列
5. 返回结果：返回消息发送结果

2. 消息消费流程

1. 获取批量大小：根据自适应策略获取当前推荐的批量大小
2. 批量获取消息：从 Redis 队列中批量获取消息
3. 解析消息：将消息 JSON 解析为消息对象
4. 并行处理：使用 CompletableFuture 并行处理消息
5. 更新状态：根据处理结果更新消息状态
6. 记录统计：记录消费统计数据
7. 调整批量：根据统计数据调整批量大小

3. 自适应调整流程

1. 监控指标：定期监控队列长度、系统负载等指标
2. 计算批量：根据监控指标计算推荐的批量大小
3. 调整批量：根据计算结果调整批量大小
4. 记录历史：记录消费速率历史，用于后续调整
5. 告警处理：当队列积压严重时发送告警

技术要点

1. 自适应批量大小算法

• 基于队列长度：队列长度越大，批量大小越大
• 基于系统负载：系统负载越高，批量大小越小
• 基于历史速率：消费速率越高，批量大小越大
• 动态调整：根据实时情况动态调整，避免频繁变动

2. 并行处理优化

• CompletableFuture：使用 CompletableFuture 并行处理消息
• 线程池：合理配置线程池大小，提高处理效率
• 超时控制：设置批量处理超时时间，避免长时间阻塞

3. 监控和告警

• 定期监控：定期监控队列长度、消费速率等指标
• 阈值告警：当队列积压超过阈值时发送告警
• 历史数据：记录历史统计数据，用于趋势分析
• 数据清理：定期清理历史数据，避免数据库膨胀

4. 性能优化

• 批量操作：批量获取和处理消息，减少网络开销
• 异步处理：异步处理消息，提高并发能力
• 缓存优化：合理使用缓存，减少数据库操作
• 资源管理：根据系统资源动态调整处理能力

最佳实践

1. 批量大小配置

• 最小批量：设置为 1-5，确保系统负载高时能够稳定处理
• 最大批量：设置为 50-100，避免批量过大导致内存占用过高
• 默认批量：设置为 10-20，适用于正常流量场景
• 调整因子：设置为 1.5-2.0，确保调整幅度合理

2. 监控配置

• 监控间隔：设置为 30-60 秒，既保证实时性又避免监控开销过大
• 告警阈值：根据系统处理能力设置合理的队列长度阈值
• 历史数据：保留 7-30 天的历史数据，用于趋势分析

3. 系统调优

• 线程池配置：根据 CPU 核心数设置合理的线程池大小
• Redis 配置：优化 Redis 配置，提高队列操作性能
• 数据库配置：优化数据库连接池，提高数据存取性能
• JVM 配置：根据系统内存设置合理的 JVM 参数

4. 异常处理

• 重试机制：对处理失败的消息进行重试，避免消息丢失
• 死信队列：对多次重试失败的消息，放入死信队列进行人工处理
• 熔断机制：当系统负载过高时，暂停消费，避免系统崩溃
• 降级处理：当系统异常时，降级为最小批量处理，确保系统稳定

常见问题

1. 批量大小调整过于频繁

问题：批量大小频繁变动，导致系统不稳定

解决方案：

• 增加调整间隔，避免短时间内重复调整
• 增加调整阈值，只有当指标变化超过一定阈值时才调整
• 平滑调整，每次调整的幅度不要过大

2. 系统负载过高

问题：批量处理导致系统负载过高

解决方案：

• 动态减小批量大小
• 实现熔断机制，当系统负载超过阈值时暂停消费
• 优化消息处理逻辑，减少资源消耗
• 增加系统资源，如 CPU、内存等

3. 队列积压严重

问题：消息队列积压严重，处理速度跟不上发送速度

解决方案：

• 动态增大批量大小
• 增加消费者实例，分布式处理
• 优化消息处理逻辑，提高处理速度
• 对非核心消息进行限流，优先处理核心消息

4. 消息处理超时

问题：批量处理超时，影响系统性能

解决方案：

• 合理设置批量处理超时时间
• 减小批量大小，避免单次处理时间过长
• 优化消息处理逻辑，减少处理时间
• 实现超时中断机制，避免长时间阻塞

代码优化建议

1. 自适应算法优化

/**
 * 优化的自适应批量大小算法
 */
private int calculateRecommendedBatchSize(long queueLength, SystemMetrics metrics, BatchSizeConfig config) {
    // 基础批量大小
    int baseBatchSize = config.getDefaultBatchSize();
    
    // 队列长度权重
    double queueWeight = 0.6;
    // 系统负载权重
    double systemWeight = 0.3;
    // 历史速率权重
    double historyWeight = 0.1;
    
    // 队列长度得分
    double queueScore = calculateQueueScore(queueLength, config);
    // 系统负载得分
    double systemScore = calculateSystemScore(metrics, config);
    // 历史速率得分
    double historyScore = calculateHistoryScore();
    
    // 综合得分
    double totalScore = queueWeight * queueScore + systemWeight * systemScore + historyWeight * historyScore;
    
    // 根据得分计算批量大小
    int calculatedBatchSize = (int) (config.getMinBatchSize() + (config.getMaxBatchSize() - config.getMinBatchSize()) * totalScore);
    
    // 确保在合理范围内
    return Math.max(config.getMinBatchSize(), Math.min(config.getMaxBatchSize(), calculatedBatchSize));
}

/**
 * 计算队列长度得分
 */
private double calculateQueueScore(long queueLength, BatchSizeConfig config) {
    if (queueLength <= config.getQueueThresholdLow()) {
        return 0.1;
    } else if (queueLength <= config.getQueueThresholdMedium()) {
        return 0.3;
    } else if (queueLength <= config.getQueueThresholdHigh()) {
        return 0.7;
    } else {
        return 1.0;
    }
}

/**
 * 计算系统负载得分
 */
private double calculateSystemScore(SystemMetrics metrics, BatchSizeConfig config) {
    double loadScore = Math.min(metrics.getSystemLoad() / config.getSystemLoadThreshold(), 1.0);
    double cpuScore = Math.min(metrics.getCpuUsage() / config.getCpuUsageThreshold(), 1.0);
    double memoryScore = Math.min(metrics.getMemoryUsage() / config.getMemoryUsageThreshold(), 1.0);
    
    double avgScore = (loadScore + cpuScore + memoryScore) / 3;
    // 系统负载越高，得分越低
    return 1.0 - avgScore;
}

/**
 * 计算历史速率得分
 */
private double calculateHistoryScore() {
    if (rateHistoryList.isEmpty()) {
        return 0.5;
    }
    
    double avgRate = rateHistoryList.stream()
            .mapToDouble(ConsumptionRate::getMessagesPerSecond)
            .average()
            .orElse(0.0);
    
    // 速率越高，得分越高
    return Math.min(avgRate / 100.0, 1.0);
}

2. 并行处理优化

/**
 * 优化的并行处理
 */
private void processMessages(List<Message> messages) {
    // 根据系统 CPU 核心数设置线程池大小
    int corePoolSize = Runtime.getRuntime().availableProcessors();
    ExecutorService executorService = Executors.newFixedThreadPool(corePoolSize);
    
    List<CompletableFuture<Void>> futures = new ArrayList<>();
    
    for (Message message : messages) {
        CompletableFuture<Void> future = CompletableFuture.runAsync(() -> processMessage(message), executorService);
        futures.add(future);
    }
    
    // 等待所有处理完成，设置超时
    CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
            .orTimeout(batchProcessingTimeout, TimeUnit.MILLISECONDS)
            .exceptionally(ex -> {
                log.error("Batch processing timed out", ex);
                return null;
            })
            .join();
    
    executorService.shutdown();
}

3. 监控优化

/**
 * 优化的监控逻辑
 */
@Scheduled(fixedRateString = "${message.monitoring.interval:30000}")
public void monitorTraffic() {
    // 获取队列长度
    long queueLength = batchProcessingService.getQueueLength();
    
    // 获取当前批量大小
    int currentBatchSize = consumptionRateService.getCurrentBatchSize();
    
    // 获取系统指标
    SystemMetrics metrics = consumptionRateService.getSystemMetrics();
    
    // 记录监控数据
    log.info("Traffic monitoring - Queue length: {}, Current batch size: {}, System load: {}, CPU: {}%, Memory: {}%",
            queueLength, currentBatchSize, metrics.getSystemLoad(), metrics.getCpuUsage(), metrics.getMemoryUsage());
    
    // 检查是否需要告警
    if (queueLength > alertThreshold) {
        sendAlert(queueLength, currentBatchSize, metrics);
    }
    
    // 定期清理历史统计数据
    cleanupHistoricalData();
    
    // 生成监控报告
    generateMonitoringReport();
}

/**
 * 生成监控报告
 */
private void generateMonitoringReport() {
    List<TrafficMonitoringService.ConsumptionRateTrend> trends = getConsumptionRateTrends();
    if (trends.isEmpty()) {
        return;
    }
    
    // 计算统计信息
    double avgMessagesPerSecond = trends.stream()
            .mapToDouble(TrafficMonitoringService.ConsumptionRateTrend::getMessagesPerSecond)
            .average()
            .orElse(0.0);
    
    double avgBatchSize = trends.stream()
            .mapToInt(TrafficMonitoringService.ConsumptionRateTrend::getBatchSize)
            .average()
            .orElse(0.0);
    
    long maxQueueLength = trends.stream()
            .mapToLong(TrafficMonitoringService.ConsumptionRateTrend::getQueueLength)
            .max()
            .orElse(0);
    
    log.info("Monitoring report - Avg messages per second: {:.2f}, Avg batch size: {:.1f}, Max queue length: {}",
            avgMessagesPerSecond, avgBatchSize, maxQueueLength);
}

互动话题

1. 你在实际项目中遇到过哪些消息消费的性能问题？是如何解决的？
2. 对于消息消费的批量大小，你认为最佳实践是什么？
3. 在分布式环境下，如何协调多个消费者的批量大小调整？
4. 你认为消息消费速率自适应还有哪些可以改进的地方？

欢迎在评论区交流讨论！

公众号：服务端技术精选，关注最新技术动态，分享最佳实践。