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. 啟動和測試

啟動你的 Spring Boot 應用。
調用 testDelay 方法發送延遲消息。
你將看到消息在隊列中延遲指定的時間（例如，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);
}