bm-site

Appearance

智能家居监控系统数据收集积压优化 ​

假设我们正在开发一个智能家居监控系统。该系统从数百万个智能设备（如温度传感器、安全摄像头、烟雾探测器等）收集数据，并通过 RocketMQ 将这些数据传输到后端进行处理和分析。

在某些情况下，比如突发事件或系统升级时，可能会导致消息处理速度跟不上消息生产速度，从而造成消息积压。

亮点：RocketMQ 消息大量积压问题的解决 ​

要解决这个问题，我们可以采取以下策略：

  1. 增加消费者数量
  2. 提高单个消费者的处理能力
  3. 实现动态扩缩容
  4. 消息优先级处理
  5. 临时存储和批量处理

下面是具体的实现方案和代码示例：

  1. 消费者配置
java
@Configuration  
public class RocketMQConsumerConfig {  

    @Value("${rocketmq.name-server}")  
    private String nameServer;  

    @Value("${rocketmq.consumer.group}")  
    private String consumerGroup;  

    @Bean  
    public DefaultMQPushConsumer deviceDataConsumer() throws MQClientException {  
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);  
        consumer.setNamesrvAddr(nameServer);  
        consumer.subscribe("DEVICE_DATA_TOPIC", "*");  
        consumer.setConsumeThreadMin(20);  
        consumer.setConsumeThreadMax(64);  
        consumer.setConsumeMessageBatchMaxSize(1);  
        consumer.registerMessageListener(new MessageListenerConcurrently() {  
            @Override  
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {  
                for (MessageExt msg : msgs) {  
                    processMessage(msg);  
                }  
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;  
            }  
        });  
        return consumer;  
    }  

    private void processMessage(MessageExt msg) {  
        // 处理消息的逻辑  
    }  
}
  1. 动态扩缩容服务
java
@Service  
public class ConsumerScalingService {  

    @Autowired  
    private DefaultMQPushConsumer deviceDataConsumer;  

    public void scaleConsumers(int threadCount) {  
        deviceDataConsumer.setConsumeThreadMin(threadCount);  
        deviceDataConsumer.setConsumeThreadMax(threadCount);  
    }  
}
  1. 消息优先级处理
java
@Service  
public class PriorityMessageProcessor {  

    @Autowired  
    private DeviceDataRepository deviceDataRepository;  

    public void processMessage(MessageExt msg) {  
        DeviceData data = parseMessage(msg);  
        if (isHighPriority(data)) {  
            processHighPriorityData(data);  
        } else {  
            deviceDataRepository.save(data);  
        }  
    }  

    private boolean isHighPriority(DeviceData data) {  
        // 判断是否为高优先级数据，如安全警报  
        return data.getType().equals(DeviceDataType.SECURITY_ALERT);  
    }  

    private void processHighPriorityData(DeviceData data) {  
        // 立即处理高优先级数据  
    }  
}
  1. 临时存储和批量处理服务
java
@Service  
public class BatchProcessingService {  

    @Autowired  
    private DeviceDataRepository deviceDataRepository;  

    @Autowired  
    private ElasticsearchTemplate elasticsearchTemplate;  

    @Scheduled(fixedRate = 300000) // 每5分钟执行一次  
    public void batchProcessStoredData() {  
        List<DeviceData> storedData = deviceDataRepository.findAll();  
        BulkRequest bulkRequest = new BulkRequest();  
        for (DeviceData data : storedData) {  
            bulkRequest.add(new IndexRequest("device_data")  
                            .id(data.getId().toString())  
                            .source(convertToMap(data)));  
        }  
        elasticsearchTemplate.bulk(bulkRequest);  
        deviceDataRepository.deleteAll(storedData);  
    }  

    private Map<String, Object> convertToMap(DeviceData data) {  
        // 将 DeviceData 对象转换为 Map  
        return null;  
    }  
}
  1. 监控和告警服务
java
@Service  
public class MessageAccumulationMonitor {  

    @Autowired  
    private DefaultMQPushConsumer deviceDataConsumer;  

    @Autowired  
    private ConsumerScalingService consumerScalingService;  

    @Autowired  
    private AlertService alertService;  

    @Scheduled(fixedRate = 60000) // 每分钟检查一次  
    public void monitorMessageAccumulation() {  
        long accumulatedMsgs = deviceDataConsumer.getDefaultMQPushConsumerImpl()  
        .getRebalanceImpl().getmQClientFactory()  
        .getMQAdminImpl().fetchConsumeStats(deviceDataConsumer.getConsumerGroup())  
        .getTotalDiff();  

        if (accumulatedMsgs > 100000) { // 如果积压超过10万条  
            consumerScalingService.scaleConsumers(100); // 将消费者线程扩展到100  
            alertService.sendAlert("Message accumulation detected: " + accumulatedMsgs);  
        } else if (accumulatedMsgs < 10000) { // 如果积压小于1万条  
            consumerScalingService.scaleConsumers(20); // 将消费者线程缩减到20  
        }  
    }  
}

解决方案说明： ​

  1. 增加消费者数量：通过 ConsumerScalingService 动态调整消费者线程数。
  2. 提高单个消费者的处理能力：在 RocketMQConsumerConfig 中配置了较大的并发消费线程数。
  3. 实现动态扩缩容：MessageAccumulationMonitor 服务监控消息积压情况，并根据需要动态调整消费者数量。
  4. 消息优先级处理：PriorityMessageProcessor 服务对高优先级消息（如安全警报）进行优先处理。
  5. 临时存储和批量处理：对于无法及时处理的消息，先存储到本地数据库，然后通过 BatchProcessingService 定期批量处理。
  6. 监控和告警：MessageAccumulationMonitor 服务监控消息积压情况，当积压严重时发送告警。

通过以上方案，我们能够有效地处理 RocketMQ 消息积压问题，确保智能家居监控系统能够及时处理大量设备数据，特别是在数据突增的情况下。这个方案不仅提高了系统的吞吐量，还保证了关键数据的及时处理，同时通过动态扩缩容和批量处理来优化资源使用。

更新: 2024-08-21 17:12:15
原文: https://www.yuque.com/tulingzhouyu/db22bv/fndpbpml9b19czom