在這個(gè)充滿挑戰(zhàn)和收獲的60天學(xué)習(xí)之旅中，你將迅速提升成為一名全棧工程師。專注于Spring Boot框架，我們將深入研究高級特性，從項(xiàng)目初始化到微服務(wù)架構(gòu)，再到性能優(yōu)化和持續(xù)集成部署。無論你是初學(xué)者還是有一定經(jīng)驗(yàn)的開發(fā)者，這個(gè)專題都將帶你穿越從零到全面掌握Spring Boot的學(xué)習(xí)曲線。oOo28資訊網(wǎng)——每日最新資訊28at.com

oOo28資訊網(wǎng)——每日最新資訊28at.com

Day 32 ~ Springboot3.1.x|3分鐘學(xué)會(huì)在 RabbitMQ 中實(shí)現(xiàn)發(fā)布訂閱模式oOo28資訊網(wǎng)——每日最新資訊28at.com

實(shí)現(xiàn)發(fā)布與訂閱消息模式

發(fā)布-訂閱模式是一種消息傳遞方式，其中發(fā)送者（發(fā)布者）不會(huì)將消息直接發(fā)送到特定的接收者（訂閱者）。發(fā)布者類別定義了哪些訂閱者因?yàn)橛嗛喺咂ヅ淞税l(fā)布者的類別而接收消息。oOo28資訊網(wǎng)——每日最新資訊28at.com

以下是使用RabbitMQ實(shí)現(xiàn)發(fā)布-訂閱模式的一種例子，我們將使用RabbitMQ的Fanout Exchange。oOo28資訊網(wǎng)——每日最新資訊28at.com

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ì)列。oOo28資訊網(wǎng)——每日最新資訊28at.com

Consumer

每一個(gè)訂閱者都需要擁有一個(gè)queue，因此，我們需要在客戶端中創(chuàng)建queue。oOo28資訊網(wǎng)——每日最新資訊28at.com

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綁定。然后我們定義了消息的接收以及處理方式。oOo28資訊網(wǎng)——每日最新資訊28at.com

處理消息發(fā)送失敗的情況

在使用消息中間件的過程中，消息發(fā)送失敗是無法避免的情況。因此，我們需要對此進(jìn)行正確的處理以避免因此而導(dǎo)致的系統(tǒng)問題。oOo28資訊網(wǎng)——每日最新資訊28at.com

對于消息發(fā)送失敗的處理，有以下幾種常用的方案：oOo28資訊網(wǎng)——每日最新資訊28at.com

• 重試: 對于有些暫時(shí)的問題，比如網(wǎng)絡(luò)波動(dòng)，可以通過簡單的重試來解決。
• 消息持久化：將消息存儲在某處（例如數(shù)據(jù)庫），只有當(dāng)消息成功發(fā)送后，再刪除它。
• 死信隊(duì)列：把無法處理的消息放入"死信隊(duì)列"，然后由專門的消費(fèi)者來進(jìn)行處理。

RabbitMQ中的消息確認(rèn)（publisher confirms）和消費(fèi)者應(yīng)答（Consumer Acknowledgements）就是為了解決此類問題。oOo28資訊網(wǎng)——每日最新資訊28at.com

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ā)送失敗。oOo28資訊網(wǎng)——每日最新資訊28at.com

結(jié)論：掌握發(fā)布-訂閱模式和消息發(fā)送失敗處理策略，對于掌握消息隊(duì)列的使用至關(guān)重要，可為系統(tǒng)的穩(wěn)定性和擴(kuò)展性提供保障。oOo28資訊網(wǎng)——每日最新資訊28at.com

oOo28資訊網(wǎng)——每日最新資訊28at.com

本文鏈接：http://m.www897cc.com/showinfo-26-79990-0.html三分鐘學(xué)會(huì)在 RabbitMQ 中實(shí)現(xiàn)發(fā)布訂閱模式

聲明：本網(wǎng)頁內(nèi)容旨在傳播知識，若有侵權(quán)等問題請及時(shí)與本網(wǎng)聯(lián)系，我們將在第一時(shí)間刪除處理。郵件：2376512515@qq.com

上一篇： SpringCloud項(xiàng)目開發(fā)中實(shí)用技巧總結(jié)

下一篇： 2024 年，這些 VS Code 插件可以卸載了！