记录一次发送千万级别数量消息的定时任务优化

后端

2023-11-07 16:44:39

优化 Kafka 架构，提升消息队列性能

在面临庞大的用户数据处理任务时，选择合适的技术架构至关重要。本文将探讨如何通过优化消息队列架构，显著提升系统吞吐量和降低延迟。

优化前的系统架构

最初的系统架构依赖于 Kafka 消息队列，但存在性能问题。发送一个消息需要大约 10 分钟，消费一个消息则需要 20 分钟。

优化后的系统架构

为了解决性能瓶颈，我们进行了以下优化：

消息队列： 将 Kafka 替换为 Redis。Redis 具有更低的延迟，更适合高频、低延迟的消息处理。
定时任务： 将 Spring Boot 的定时任务替换为 RocketMQ 的定时任务。RocketMQ 的定时任务具有更高的可靠性和准确性。
消费者： 将 Spring Boot 开发的消费者程序替换为 RocketMQ 的消费者程序。RocketMQ 的消费者程序具有更高的吞吐量。

优化效果

经过优化后，系统性能大幅提升：

发送消息： 1 分钟左右
消费消息： 5 分钟左右

性能提升的原因

优化后，系统性能提升的主要原因包括：

Redis 的低延迟： Redis 的内存缓存特性使得消息发送和消费更加迅速。
RocketMQ 的可靠性和准确性： RocketMQ 保证了定时任务的可靠性和准确性，确保消息及时发送和消费。
RocketMQ 的高吞吐量： RocketMQ 的高吞吐量消费者程序可以更快地处理消息。

代码示例

发送消息：

RedisTemplate<String, String> redisTemplate = new RedisTemplate<>();
redisTemplate.opsForValue().set("CUSTOMER_DAILY", "daily-data");

消费消息：

RocketMQMessageListenerContainer container = new RocketMQMessageListenerContainer();
container.setConsumerGroup("group-name");
container.setTopic("CUSTOMER_DAILY");
container.setMessageListener(new MessageListener() {
    @Override
    public void onMessage(Message message) {
        // 处理消息
    }
});
container.start();