三分鐘學(xué)會(huì)在 RabbitMQ 中實(shí)現(xiàn)發(fā)布訂閱模式
在這個(gè)充滿挑戰(zhàn)和收獲的60天學(xué)習(xí)之旅中，你將迅速提升成為一名全棧工程師。專注于Spring Boot框架，我們將深入研究高級(jí)特性，從項(xiàng)目初始化到微服務(wù)架構(gòu)，再到性能優(yōu)化和持續(xù)集成部署。無(wú)論你是初學(xué)者還是有一定經(jīng)驗(yàn)的開發(fā)者，這個(gè)專題都將帶你穿越從零到全面掌握Spring Boot的學(xué)習(xí)曲線。
Day 32 ~ Springboot3.1.x|3分鐘學(xué)會(huì)在 RabbitMQ 中實(shí)現(xiàn)發(fā)布訂閱模式
實(shí)現(xiàn)發(fā)布與訂閱消息模式
發(fā)布-訂閱模式是一種消息傳遞方式，其中發(fā)送者（發(fā)布者）不會(huì)將消息直接發(fā)送到特定的接收者（訂閱者）。發(fā)布者類別定義了哪些訂閱者因?yàn)橛嗛喺咂ヅ淞税l(fā)布者的類別而接收消息。
以下是使用RabbitMQ實(shí)現(xiàn)發(fā)布-訂閱模式的一種例子，我們將使用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")，我們聲明一個(gè)名為"log"的exchange，同時(shí)我們定義其類型為"fanout"，意味著它會(huì)將接收到的所有消息廣播給所有它所知道的隊(duì)列。
Consumer
每一個(gè)訂閱者都需要擁有一個(gè)queue，因此，我們需要在客戶端中創(chuàng)建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 -> { });
}
}在這個(gè)例子中，我們聲明一個(gè)新的queue，并將其與"logs"的exchange綁定。然后我們定義了消息的接收以及處理方式。
處理消息發(fā)送失敗的情況
在使用消息中間件的過(guò)程中，消息發(fā)送失敗是無(wú)法避免的情況。因此，我們需要對(duì)此進(jìn)行正確的處理以避免因此而導(dǎo)致的系統(tǒng)問題。
對(duì)于消息發(fā)送失敗的處理，有以下幾種常用的方案：
- 重試: 對(duì)于有些暫時(shí)的問題，比如網(wǎng)絡(luò)波動(dòng)，可以通過(guò)簡(jiǎn)單的重試來(lái)解決。
- 消息持久化：將消息存儲(chǔ)在某處（例如數(shù)據(jù)庫(kù)），只有當(dāng)消息成功發(fā)送后，再刪除它。
- 死信隊(duì)列：把無(wú)法處理的消息放入"死信隊(duì)列"，然后由專門的消費(fèi)者來(lái)進(jìn)行處理。
RabbitMQ中的消息確認(rèn)（publisher confirms）和消費(fèi)者應(yīng)答（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("消息發(fā)送失敗");
}
} catch (Exception e) {
System.out.println("錯(cuò)誤: " + e.getMessage());
}
}上述代碼中執(zhí)行channel.confirmSelect();后，當(dāng)前channel被設(shè)置為publisher confirm模式。在此模式下，當(dāng)消息被RabbitMQ成功接收后，會(huì)發(fā)送一個(gè)確認(rèn)給生產(chǎn)者。如果RabbitMQ沒有發(fā)送確認(rèn)，那么生產(chǎn)者可以認(rèn)定該消息發(fā)送失敗。
結(jié)論：掌握發(fā)布-訂閱模式和消息發(fā)送失敗處理策略，對(duì)于掌握消息隊(duì)列的使用至關(guān)重要，可為系統(tǒng)的穩(wěn)定性和擴(kuò)展性提供保障。
