您好,登錄后才能下訂單哦!
這篇文章給大家分享的是有關(guān)RabbitMQ中七種模式介紹與應(yīng)用場(chǎng)景的示例分析的內(nèi)容。小編覺得挺實(shí)用的,因此分享給大家做個(gè)參考,一起跟隨小編過來看看吧。
做最簡(jiǎn)單的事情,一個(gè)生產(chǎn)者對(duì)應(yīng)一個(gè)消費(fèi)者,RabbitMQ相當(dāng)于一個(gè)消息代理,負(fù)責(zé)將A的消息轉(zhuǎn)發(fā)給B
應(yīng)用場(chǎng)景:將發(fā)送的電子郵件放到消息隊(duì)列,然后郵件服務(wù)在隊(duì)列中獲取郵件并發(fā)送給收件人
在多個(gè)消費(fèi)者之間分配任務(wù)(競(jìng)爭(zhēng)的消費(fèi)者模式),一個(gè)生產(chǎn)者對(duì)應(yīng)多個(gè)消費(fèi)者,一般適用于執(zhí)行資源密集型任務(wù),單個(gè)消費(fèi)者處理不過來,需要多個(gè)消費(fèi)者進(jìn)行處理
應(yīng)用場(chǎng)景:一個(gè)訂單的處理需要10s,有多個(gè)訂單可以同時(shí)放到消息隊(duì)列,然后讓多個(gè)消費(fèi)者同時(shí)處理,這樣就是并行了,而不是單個(gè)消費(fèi)者的串行情況
一次向許多消費(fèi)者發(fā)送消息,一個(gè)生產(chǎn)者發(fā)送的消息會(huì)被多個(gè)消費(fèi)者獲取,也就是將消息將廣播到所有的消費(fèi)者中。
應(yīng)用場(chǎng)景:更新商品庫存后需要通知多個(gè)緩存和多個(gè)數(shù)據(jù)庫,這里的結(jié)構(gòu)應(yīng)該是:
一個(gè)fanout類型交換機(jī)扇出兩個(gè)個(gè)消息隊(duì)列,分別為緩存消息隊(duì)列、數(shù)據(jù)庫消息隊(duì)列
一個(gè)緩存消息隊(duì)列對(duì)應(yīng)著多個(gè)緩存消費(fèi)者
一個(gè)數(shù)據(jù)庫消息隊(duì)列對(duì)應(yīng)著多個(gè)數(shù)據(jù)庫消費(fèi)者
有選擇地(Routing key)接收消息,發(fā)送消息到交換機(jī)并且要指定路由key ,消費(fèi)者將隊(duì)列綁定到交換機(jī)時(shí)需要指定路由key,僅消費(fèi)指定路由key的消息
應(yīng)用場(chǎng)景:如在商品庫存中增加了1臺(tái)iphone12,iphone12促銷活動(dòng)消費(fèi)者指定routing key為iphone12,只有此促銷活動(dòng)會(huì)接收到消息,其它促銷活動(dòng)不關(guān)心也不會(huì)消費(fèi)此routing key的消息
根據(jù)主題(Topics)來接收消息,將路由key和某模式進(jìn)行匹配,此時(shí)隊(duì)列需要綁定在一個(gè)模式上,#
匹配一個(gè)詞或多個(gè)詞,*
只匹配一個(gè)詞。
應(yīng)用場(chǎng)景:同上,iphone促銷活動(dòng)可以接收主題為iphone的消息,如iphone12、iphone13等
如果我們需要在遠(yuǎn)程計(jì)算機(jī)上運(yùn)行功能并等待結(jié)果就可以使用RPC,具體流程可以看圖。
應(yīng)用場(chǎng)景:需要等待接口返回?cái)?shù)據(jù),如訂單支付
與發(fā)布者進(jìn)行可靠的發(fā)布確認(rèn),發(fā)布者確認(rèn)是RabbitMQ擴(kuò)展,可以實(shí)現(xiàn)可靠的發(fā)布。在通道上啟用發(fā)布者確認(rèn)后,RabbitMQ將異步確認(rèn)發(fā)送者發(fā)布的消息,這意味著它們已在服務(wù)器端處理
應(yīng)用場(chǎng)景:對(duì)于消息可靠性要求較高,比如錢包扣款
代碼中沒有對(duì)后面兩種模式演示,有興趣可以自己研究
import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import java.io.IOException; import java.util.concurrent.TimeoutException; public class Sender { private final static String QUEUE_NAME = "simple_queue"; public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); factory.setPort(5672); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); // 聲明隊(duì)列 // queue:隊(duì)列名 // durable:是否持久化 // exclusive:是否排外 即只允許該channel訪問該隊(duì)列 一般等于true的話用于一個(gè)隊(duì)列只能有一個(gè)消費(fèi)者來消費(fèi)的場(chǎng)景 // autoDelete:是否自動(dòng)刪除 消費(fèi)完刪除 // arguments:其他屬性 channel.queueDeclare(QUEUE_NAME, false, false, false, null); //消息內(nèi)容 String message = "simplest mode message"; channel.basicPublish("", QUEUE_NAME, null, message.getBytes()); System.out.println("[x]Sent '" + message + "'"); //最后關(guān)閉通關(guān)和連接 channel.close(); connection.close(); } }
import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.DeliverCallback; import java.io.IOException; import java.util.concurrent.TimeoutException; public class Receiver { private final static String QUEUE_NAME = "simplest_queue"; public static void main(String[] args) throws IOException, InterruptedException, TimeoutException { // 獲取連接 ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); factory.setPort(5672); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.queueDeclare(QUEUE_NAME, false, false, false, null); DeliverCallback deliverCallback = (consumerTag, delivery) -> { String message = new String(delivery.getBody(), "UTF-8"); System.out.println(" [x] Received '" + delivery.getEnvelope().getRoutingKey() + "':'" + message + "'"); }; channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { }); } }
import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.DeliverCallback; import java.io.IOException; import java.util.concurrent.TimeoutException; public class Receiver1 { private final static String QUEUE_NAME = "queue_work"; public static void main(String[] args) throws IOException, InterruptedException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); factory.setPort(5672); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 同一時(shí)刻服務(wù)器只會(huì)發(fā)送一條消息給消費(fèi)者 channel.basicQos(1); DeliverCallback deliverCallback = (consumerTag, delivery) -> { String message = new String(delivery.getBody(), "UTF-8"); System.out.println(" [x] Received '" + delivery.getEnvelope().getRoutingKey() + "':'" + message + "'"); }; channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { }); } }
import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.DeliverCallback; import java.io.IOException; import java.util.concurrent.TimeoutException; public class Receiver2 { private final static String QUEUE_NAME = "queue_work"; public static void main(String[] args) throws IOException, InterruptedException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); factory.setPort(5672); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 同一時(shí)刻服務(wù)器只會(huì)發(fā)送一條消息給消費(fèi)者 channel.basicQos(1); DeliverCallback deliverCallback = (consumerTag, delivery) -> { String message = new String(delivery.getBody(), "UTF-8"); System.out.println(" [x] Received '" + delivery.getEnvelope().getRoutingKey() + "':'" + message + "'"); }; channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { }); } }
import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import java.io.IOException; import java.util.concurrent.TimeoutException; public class Sender { private final static String QUEUE_NAME = "queue_work"; public static void main(String[] args) throws IOException, InterruptedException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); factory.setPort(5672); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); // 聲明隊(duì)列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); for (int i = 0; i < 100; i++) { String message = "work mode message" + i; channel.basicPublish("", QUEUE_NAME, null, message.getBytes()); System.out.println("[x] Sent '" + message + "'"); Thread.sleep(i * 10); } channel.close(); connection.close(); } }
import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.DeliverCallback; public class Receive1 { 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, ""); System.out.println(" [*] Waiting for messages. To exit press CTRL+C"); // 訂閱消息的回調(diào)函數(shù) DeliverCallback deliverCallback = (consumerTag, delivery) -> { String message = new String(delivery.getBody(), "UTF-8"); System.out.println(" [x] Received '" + message + "'"); }; // 消費(fèi)者,有消息時(shí)出發(fā)訂閱回調(diào)函數(shù) channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { }); } }
import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.DeliverCallback; public class Receive2 { 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, ""); System.out.println(" [*] Waiting for messages. To exit press CTRL+C"); // 訂閱消息的回調(diào)函數(shù) DeliverCallback deliverCallback = (consumerTag, delivery) -> { String message = new String(delivery.getBody(), "UTF-8"); System.out.println(" [x] Received2 '" + message + "'"); }; // 消費(fèi)者,有消息時(shí)出發(fā)訂閱回調(diào)函數(shù) channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { }); } }
import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; public class Sender { 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 message = "publish subscribe message"; channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8")); System.out.println(" [x] Sent '" + message + "'"); channel.close(); connection.close(); } }
import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.DeliverCallback; import java.io.IOException; import java.util.concurrent.TimeoutException; public class Receiver1 { private final static String QUEUE_NAME = "queue_routing"; private final static String EXCHANGE_NAME = "exchange_direct"; public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); factory.setPort(5672); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 指定路由的key,接收key和key2 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "key"); channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "key2"); channel.basicQos(1); DeliverCallback deliverCallback = (consumerTag, delivery) -> { String message = new String(delivery.getBody(), "UTF-8"); System.out.println(" [x] Received '" + delivery.getEnvelope().getRoutingKey() + "':'" + message + "'"); }; channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { }); } }
import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.DeliverCallback; import java.io.IOException; import java.util.concurrent.TimeoutException; public class Receiver2 { private final static String QUEUE_NAME = "queue_routing2"; private final static String EXCHANGE_NAME = "exchange_direct"; public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); factory.setPort(5672); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 僅接收key2 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "key2"); channel.basicQos(1); DeliverCallback deliverCallback = (consumerTag, delivery) -> { String message = new String(delivery.getBody(), "UTF-8"); System.out.println(" [x] Received '" + delivery.getEnvelope().getRoutingKey() + "':'" + message + "'"); }; channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { }); } }
import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import java.io.IOException; import java.util.concurrent.TimeoutException; public class Sender { private final static String EXCHANGE_NAME = "exchange_direct"; private final static String EXCHANGE_TYPE = "direct"; public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); factory.setPort(5672); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); // 交換機(jī)聲明 channel.exchangeDeclare(EXCHANGE_NAME, EXCHANGE_TYPE); // 只有routingKey相同的才會(huì)消費(fèi) String message = "routing mode message"; channel.basicPublish(EXCHANGE_NAME, "key2", null, message.getBytes()); System.out.println("[x] Sent '" + message + "'"); // channel.basicPublish(EXCHANGE_NAME, "key", null, message.getBytes()); // System.out.println("[x] Sent '" + message + "'"); channel.close(); connection.close(); } }
import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.DeliverCallback; import java.io.IOException; import java.util.concurrent.TimeoutException; public class Receiver1 { private final static String QUEUE_NAME = "queue_topic"; private final static String EXCHANGE_NAME = "exchange_topic"; public static void main(String[] args) throws IOException, InterruptedException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); factory.setPort(5672); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 可以接收key.1 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "key.*"); channel.basicQos(1); DeliverCallback deliverCallback = (consumerTag, delivery) -> { String message = new String(delivery.getBody(), "UTF-8"); System.out.println(" [x] Received '" + delivery.getEnvelope().getRoutingKey() + "':'" + message + "'"); }; channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { }); } }
import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.DeliverCallback; import java.io.IOException; import java.util.concurrent.TimeoutException; public class Receiver2 { private final static String QUEUE_NAME = "queue_topic2"; private final static String EXCHANGE_NAME = "exchange_topic"; private final static String EXCHANGE_TYPE = "topic"; public static void main(String[] args) throws IOException, InterruptedException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); factory.setPort(5672); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.queueDeclare(QUEUE_NAME, false, false, false, null); // *號(hào)代表單個(gè)單詞,可以接收key.1 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "*.*"); // #號(hào)代表多個(gè)單詞,可以接收key.1.2 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "*.#"); channel.basicQos(1); DeliverCallback deliverCallback = (consumerTag, delivery) -> { String message = new String(delivery.getBody(), "UTF-8"); System.out.println(" [x] Received '" + delivery.getEnvelope().getRoutingKey() + "':'" + message + "'"); }; channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { }); } }
import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import java.io.IOException; import java.util.concurrent.TimeoutException; public class Sender { private final static String EXCHANGE_NAME = "exchange_topic"; private final static String EXCHANGE_TYPE = "topic"; public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); factory.setPort(5672); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.exchangeDeclare(EXCHANGE_NAME, EXCHANGE_TYPE); String message = "topics model message with key.1"; channel.basicPublish(EXCHANGE_NAME, "key.1", null, message.getBytes()); System.out.println("[x] Sent '" + message + "'"); String message2 = "topics model message with key.1.2"; channel.basicPublish(EXCHANGE_NAME, "key.1.2", null, message2.getBytes()); System.out.println("[x] Sent '" + message2 + "'"); channel.close(); connection.close(); } }
直連交換機(jī)(Direct exchange):具有路由功能的交換機(jī),綁定到此交換機(jī)的時(shí)候需要指定一個(gè)routing_key
,交換機(jī)發(fā)送消息的時(shí)候需要routing_key
,會(huì)將消息發(fā)送道對(duì)應(yīng)的隊(duì)列
扇形交換機(jī)(Fanout exchange):廣播消息到所有隊(duì)列,沒有任何處理,速度最快
主題交換機(jī)(Topic exchange):在直連交換機(jī)基礎(chǔ)上增加模式匹配,也就是對(duì)routing_key
進(jìn)行模式匹配,*
代表一個(gè)單詞,#
代表多個(gè)單詞
首部交換機(jī)(Headers exchange):忽略routing_key
,使用Headers
信息(一個(gè)Hash的數(shù)據(jù)結(jié)構(gòu))進(jìn)行匹配,優(yōu)勢(shì)在于可以有更多更靈活的匹配規(guī)則
感謝各位的閱讀!關(guān)于“RabbitMQ中七種模式介紹與應(yīng)用場(chǎng)景的示例分析”這篇文章就分享到這里了,希望以上內(nèi)容可以對(duì)大家有一定的幫助,讓大家可以學(xué)到更多知識(shí),如果覺得文章不錯(cuò),可以把它分享出去讓更多的人看到吧!
免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng),如果涉及侵權(quán)請(qǐng)聯(lián)系站長(zhǎng)郵箱:is@yisu.com進(jìn)行舉報(bào),并提供相關(guān)證據(jù),一經(jīng)查實(shí),將立刻刪除涉嫌侵權(quán)內(nèi)容。