三分鐘學會在 RabbitMQ 中實現發布訂閱模式
在這個充滿挑戰和收獲的60天學習之旅中，你將迅速提升成為一名全棧工程師。專注于Spring Boot框架，我們將深入研究高級特性，從項目初始化到微服務架構，再到性能優化和持續集成部署。無論你是初學者還是有一定經驗的開發者，這個專題都將帶你穿越從零到全面掌握Spring Boot的學習曲線。
Day 32 ~ Springboot3.1.x|3分鐘學會在 RabbitMQ 中實現發布訂閱模式
實現發布與訂閱消息模式
發布-訂閱模式是一種消息傳遞方式，其中發送者（發布者）不會將消息直接發送到特定的接收者（訂閱者）。發布者類別定義了哪些訂閱者因為訂閱者匹配了發布者的類別而接收消息。
以下是使用RabbitMQ實現發布-訂閱模式的一種例子，我們將使用RabbitMQ的Fanout Exchange。
Producer
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.nio.charset.StandardCharsets;
public class EmitLog {
private static final String EXCHANGE_NAME = "logs";
public static void main(String[] argv) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
try (Connection connection = factory.newConnection();
Channel channel = connection.createChannel()) {
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
String message = "Log message...";
channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes(StandardCharsets.UTF_8));
System.out.println("Sent '" + message + "'");
}
}
}在上述代碼的channel.exchangeDeclare(EXCHANGE_NAME, "fanout")，我們聲明一個名為"log"的exchange，同時我們定義其類型為"fanout"，意味著它會將接收到的所有消息廣播給所有它所知道的隊列。
Consumer
每一個訂閱者都需要擁有一個queue，因此，我們需要在客戶端中創建queue。
import com.rabbitmq.client.*;
import java.io.IOException;
public class ReceiveLogs {
private static final String EXCHANGE_NAME = "logs";
public static void main(String[] argv) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
String queueName = channel.queueDeclare().getQueue();
channel.queueBind(queueName, EXCHANGE_NAME, "");
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
System.out.println("Received '" + message + "'");
};
channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { });
}
}在這個例子中，我們聲明一個新的queue，并將其與"logs"的exchange綁定。然后我們定義了消息的接收以及處理方式。
處理消息發送失敗的情況
在使用消息中間件的過程中，消息發送失敗是無法避免的情況。因此，我們需要對此進行正確的處理以避免因此而導致的系統問題。
對于消息發送失敗的處理，有以下幾種常用的方案：
- 重試: 對于有些暫時的問題，比如網絡波動，可以通過簡單的重試來解決。
- 消息持久化：將消息存儲在某處（例如數據庫），只有當消息成功發送后，再刪除它。
- 死信隊列：把無法處理的消息放入"死信隊列"，然后由專門的消費者來進行處理。
RabbitMQ中的消息確認（publisher confirms）和消費者應答（Consumer Acknowledgements）就是為了解決此類問題。
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
try (Connection connection = factory.newConnection()) {
Channel channel = connection.createChannel();
String queueName = "test";
String message = "Hello world";
try {
channel.queueDeclare(queueName, false, false, false, null);
channel.confirmSelect();
channel.basicPublish("", queueName, null, message.getBytes());
if (!channel.waitForConfirms()) {
System.out.println("消息發送失敗");
}
} catch (Exception e) {
System.out.println("錯誤: " + e.getMessage());
}
}上述代碼中執行channel.confirmSelect();后，當前channel被設置為publisher confirm模式。在此模式下，當消息被RabbitMQ成功接收后，會發送一個確認給生產者。如果RabbitMQ沒有發送確認，那么生產者可以認定該消息發送失敗。
結論：掌握發布-訂閱模式和消息發送失敗處理策略，對于掌握消息隊列的使用至關重要，可為系統的穩定性和擴展性提供保障。
