SpringBoot整合RabbitMQ幂等性、优先级队列、惰性队列_springboot rabbitmq 优先级队列

在Spring Boot整合RabbitMQ时，实现幂等性、优先级队列和惰性队列是提高消息处理可靠性和性能的重要手段。下面是如何配置和使用这些特性：

幂等性（Idempotency）

实现思路：

• 消费者端：确保消息的消费操作具有幂等性，即无论同一消息被消费多少次，系统状态都保持一致。
• 唯一标识：为每条消息生成一个唯一的标识符（如业务ID或UUID），并在数据库或其他存储中记录已处理的消息标识符。
• 预处理检查：消费消息前先检查该消息是否已被处理过，若已处理则直接跳过。

@Service
public class IdempotentMessageConsumer {

    private Set<String> processedIds = new HashSet<>();

    @RabbitListener(queues = "myQueue")
    public void consume(String message, MessageHeaders headers) {
        String messageId = (String) headers.get("messageId");
        if (!processedIds.contains(messageId)) {
            // 模拟业务处理
            processMessage(message);
            processedIds.add(messageId);
        } else {
            log.info("Message with ID {} is already processed, skipping", messageId);
        }
    }

    private void processMessage(String message) {
        // 你的业务逻辑处理代码...
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

优先级队列（Priority Queue）

配置方式：

• 创建多个普通队列，并通过一个优先级交换器（x-match: all 或 x-match: any）进行绑定，每个队列设置不同的优先级（x-max-priority 参数）。

@Configuration
public class RabbitConfig {

    @Bean
    public FanoutExchange priorityExchange() {
        return new FanoutExchange("priorityExchange", true, false);
    }

    @Bean
    public Queue highPriorityQueue() {
        Map<String, Object> args = new HashMap<>();
        args.put("x-max-priority", 10); // 假设最高优先级为10
        return new Queue("highPriorityQueue", true, false, false, args);
    }

    @Bean
    public Queue mediumPriorityQueue() {
        Map<String, Object> args = new HashMap<>();
        args.put("x-max-priority", 5);
        return new Queue("mediumPriorityQueue", true, false, false, args);
    }

    // 绑定队列到优先级交换器并指定路由键（在这种模式下，路由键可以相同）
    @Bean
    public Binding highPriorityBinding(FanoutExchange priorityExchange, Queue highPriorityQueue) {
        return BindingBuilder.bind(highPriorityQueue).to(priorityExchange);
    }

    @Bean
    public Binding mediumPriorityBinding(FanoutExchange priorityExchange, Queue mediumPriorityQueue) {
        return BindingBuilder.bind(mediumPriorityQueue).to(priorityExchange);
    }
}

// 在发送消息时，根据业务需求设置优先级字段
rabbitTemplate.convertAndSend(exchange, "", message, new MessagePostProcessor() {
    @Override
    public Message postProcessMessage(Message message) throws AmqpException {
        message.getMessageProperties().setHeader("x-priority", 8); // 设置优先级值
        return message;
    }
});
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

惰性队列（Lazy Queue）

配置方式：

• 在创建队列时启用惰性队列属性（x-queue-mode 设置为 lazy）。

@Bean
public Queue lazyQueue() {
    Map<String, Object> args = new HashMap<>();
    args.put("x-queue-mode", "lazy"); // 启用惰性队列
    return new Queue("lazyQueue", true, false, false, args);
}
1
2
3
4
5
6

惰性队列在没有消费者连接时不会将消息从磁盘读取到内存，只有当有消费者开始监听这个队列时才会逐渐加载消息，有助于减少资源消耗，特别是在大容量队列和长时间不活跃的场景中。

以上示例仅提供了基本的配置方法，实际应用中可能还需要结合具体的业务场景进一步优化和调整。