rabbitmq 延时队列

要使用 RabbitMQ Delayed Message Plugin 实现延时队列，首先需要确保插件已安装并启用。以下是实现延时队列的步骤和代码示例。

1. 安装 RabbitMQ Delayed Message Plugin

首先，确保你的 RabbitMQ 安装了 rabbitmq-delayed-message-exchange 插件。你可以通过以下命令安装和启用插件：

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

2. 创建交换机和队列

你需要创建一个 延时交换机（x-delayed-message）和一个普通队列。我们将在发送消息时指定延迟时间。

3. 发送延迟消息的代码示例

假设你已经在 RabbitMQ 中设置了延时交换机。以下是使用 Java 和 Spring AMQP 发送延迟消息的代码示例。

Maven 依赖

确保你的项目中已经添加了 Spring AMQP 相关依赖：

<dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-amqp</artifactId><version>2.4.6</version>  <!-- 适配你使用的版本 -->
</dependency>

配置延时交换机和队列

你需要配置一个 延时交换机 和 队列，并设置消息的延迟时间。

@Configuration
public class RabbitConfig {// 创建一个延时交换机@Beanpublic CustomExchange delayedExchange() {Map<String, Object> arguments = new HashMap<>();// 设定交换机类型为延时交换机arguments.put("x-delayed-type", "direct");return new CustomExchange("delayed-exchange", "x-delayed-message", true, false, arguments);}// 创建队列@Beanpublic Queue delayedQueue() {return new Queue("delayed-queue", true);}// 将队列绑定到延时交换机@Beanpublic Binding binding(Queue delayedQueue, CustomExchange delayedExchange) {return BindingBuilder.bind(delayedQueue).to(delayedExchange).with("delayed.routing.key").noargs();}
}

发送延迟消息

在消息发送时，你需要通过设置消息的属性来指定延迟时间。可以使用 AMQP.BasicProperties 来设置消息的 x-delay 属性，这个值表示延迟的时间（单位：毫秒）。

@Service
public class MessageProducer {@Autowiredprivate AmqpTemplate amqpTemplate;public void sendDelayedMessage(String message, int delayMilliseconds) {// 创建消息属性，并设置延迟时间MessageProperties messageProperties = new MessageProperties();messageProperties.setDelay(delayMilliseconds);  // 设置延迟时间（毫秒）Message messageObj = new Message(message.getBytes(), messageProperties);// 发送消息到延时交换机amqpTemplate.send("delayed-exchange", "delayed.routing.key", messageObj);System.out.println("Sent delayed message: " + message + " with delay: " + delayMilliseconds + " ms");}
}

在上面的代码中，setDelay(delayMilliseconds) 方法设置了延迟时间。这个时间会告诉 RabbitMQ 延迟多久后将消息投递到队列中。

监听消息

最后，你需要设置消费者来监听这个延时队列，并处理接收到的消息：

@Service
public class MessageConsumer {@RabbitListener(queues = "delayed-queue")public void consume(String message) {System.out.println("Received delayed message: " + message);}
}

4. 测试发送延迟消息

现在，你可以在业务逻辑中调用 sendDelayedMessage 方法发送延时消息。例如，发送一条延迟 10 秒的消息：

@Autowired
private MessageProducer messageProducer;public void testDelay() {// 发送一条延迟10秒的消息messageProducer.sendDelayedMessage("Hello, delayed world!", 10000);
}

5. 启动和测试

1. 启动你的 Spring Boot 应用。
2. 调用 testDelay 方法发送延迟消息。
3. 你将看到消息在队列中延迟指定的时间（例如，10秒）后被消费。

关键点：

• 通过 x-delayed-message 交换机，设置 x-delayed-type 为 direct 或 topic，根据需求选择交换机类型。
• 使用 setDelay 方法设置延迟时间，单位是毫秒。
• RabbitMQ 会在指定的时间到达后，将消息投递到目标队列。

总结

通过 RabbitMQ Delayed Message Plugin，你可以非常方便地实现延时队列。只需要创建一个支持延迟的交换机，并通过设置 x-delay 属性来指定消息的延迟时间。

配置并行消费

要启动多个消费者并并行处理 RabbitMQ 中的消息，通常可以通过 Spring AMQP 和 RabbitListener 实现。这将帮助你加快消费速度，提升系统的吞吐量。下面是如何启动多个消费者进行并行消费的代码修改步骤：

1. 配置多个消费者

Spring AMQP 支持使用 @RabbitListener 注解启动多个消费者实例。通过配置 并行消费者，Spring 会为每个消费者实例分配一个独立的线程来处理消息。

2. 增加消费者并发处理能力

为了实现并发消费，我们可以通过以下几种方式：

• 使用 @RabbitListener 启动多个消费者实例：每个 @RabbitListener 注解的消费者都会独立地消费队列中的消息。
• 配置 SimpleMessageListenerContainer 的并发设置：通过配置 SimpleMessageListenerContainer，你可以设置多个消费者同时监听队列，从而提高并发消费能力。

3. 代码修改示例

1) 创建并发消费者

首先，创建一个通用的消息监听器，并将 @RabbitListener 注解应用于多个消费者实例上。你可以通过 @RabbitListener 注解中的 concurrency 属性来设置消费者的并发数量。

@Service
public class ConcurrentMessageConsumer {// 使用 @RabbitListener 注解配置多个并发消费者，默认启动2个消费者@RabbitListener(queues = "delayed-queue", concurrency = "3-5")  // 设置并发消费者数目 3-5 个消费者public void consume(String message) {System.out.println("Thread: " + Thread.currentThread().getName() + " - Received message: " + message);}
}

在上面的代码中，concurrency = "3-5" 表示 Spring 会启动 3 到 5 个消费者实例来并行处理队列中的消息。消费者数目是动态的，具体数量由 Spring 的消息监听容器控制。

• "3-5" 表示最低启动 3 个消费者，最多启动 5 个消费者来并行处理消息。
• 如果消息量很大，Spring 会动态调整消费者的数量，以适应系统的负载。

2) 配置并发消费者的线程池（可选）

为了更好地控制消费者的线程池和消息消费的并发度，你可以通过配置 SimpleMessageListenerContainer 来定义更具体的并发设置。例如，你可以在 Spring 配置类中手动定义消费者容器。

@Configuration
public class RabbitConfig {@Beanpublic SimpleMessageListenerContainer messageListenerContainer(ConnectionFactory connectionFactory,MessageListener messageListener) {SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();container.setConnectionFactory(connectionFactory);container.setQueueNames("delayed-queue");container.setMessageListener(messageListener);// 设置并发消费的最小值和最大值container.setConcurrentConsumers(3);  // 最小3个消费者container.setMaxConcurrentConsumers(10);  // 最大10个消费者return container;}
}

• setConcurrentConsumers(3)：设置最小消费者数量。
• setMaxConcurrentConsumers(10)：设置最大消费者数量，Spring 会根据消息的积压情况动态调整消费者的数量。

3) 控制消费者的负载和流量

如果你希望更精细地控制消息消费的负载，可以使用 @RabbitListener 注解中的 acknowledgeMode 设置来调整消息确认模式，确保消息被正确地处理和确认。例如，使用 MANUAL 手动确认消费：

@RabbitListener(queues = "delayed-queue", ackMode = "MANUAL")
public void consumeWithAck(Message message, Channel channel) throws IOException {try {// 消费消息System.out.println("Consumed message: " + new String(message.getBody()));// 手动确认消息channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);} catch (Exception e) {// 处理异常，手动拒绝消息channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);}
}

通过手动确认，你可以更好地控制消息的确认和失败重试机制，防止在消费者挂掉的情况下丢失消息。

4. 测试并发消费

你可以通过调用 testDelay 方法或者其他方式，发送延时消息来验证并发消费是否生效。发送的消息会被多个消费者并行处理，输出的日志中会显示哪个线程消费了哪个消息，从而验证消费者的并发能力。

@Autowired
private MessageProducer messageProducer;public void testDelay() {// 发送一条延迟10秒的消息messageProducer.sendDelayedMessage("Hello, delayed world!", 10000);
}

5. 总结

通过配置多个并发消费者来加速消息消费，有以下几个要点：

• 使用 @RabbitListener(concurrency = "3-5") 注解来启动多个并发消费者。
• 配置 SimpleMessageListenerContainer 来更灵活地管理消费者线程池。
• 使用手动确认模式（ackMode = "MANUAL"）可以更精细地控制消息确认和失败重试。

通过这些配置，你可以根据消息量的大小和系统负载动态调整消费者数量，以达到加快消费速度的目的。