您好,登錄后才能下訂單哦!
本篇內容主要講解“Rabbit MQ的廣播模式是什么意思”,感興趣的朋友不妨來看看。本文介紹的方法操作簡單快捷,實用性強。下面就讓小編來帶大家學習“Rabbit MQ的廣播模式是什么意思”吧!
最近做了一個需求,需要一個接口,可以給3萬多人發消息,因為人數較多,用一條線程同步發送肯定是不行的,如果是一條線程,同步發送的話,哪怕有一條數據出現了問題,都會導致本次發送的失敗,所以,想到了用Rabbit MQ的廣播模式來做。
廣播模式其實就是將你要發送的消息放到“交換機”中,然后對應的隊列去交換機中獲取消息給消費者(接收消息的一端)消費。如下圖所示:
(emmm 圖片是我在網上找的,自己畫的太丑了)
在上述過程中,主線程執行到第一步之后就可以返回了,這是一個異步的過程,不需要等待隊列去交換機中取完數據,“主線程返回“和“隊列取數據給消費者”這兩個過程是同時進行的。這樣的話,發送消息的時候就不必等到3萬多條消息都發送完成才能繼續進行下一步了。
功能大概介紹了一下,接下來就是代碼了,我只做一個簡單的記錄
首先是一個配置類
package com.apps.rabbit.FanoutMessage; import org.springframework.amqp.core.Binding; import org.springframework.amqp.core.BindingBuilder; import org.springframework.amqp.core.FanoutExchange; import org.springframework.amqp.core.Queue; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @Configuration public class TestExchangeConfiguration { /** * 創建廣播形式的交換機 * * @return 交換機實例 */ @Bean public FanoutExchange fanoutExchanges() { System.out.println("【【【交換機實例創建成功】】】"); return new FanoutExchange(Constants.FANOUT_EXCHANGE_NAME); } /** * 測試隊列一 * * @return 隊列實例 */ @Bean public Queue queue1() { System.out.println("【【【測試隊列一實例創建成功】】】"); return new Queue(Constants.TEST_QUEUE1_NAME); } /** * 綁定隊列一到交換機 * * @return 綁定對象 */ @Bean public Binding bingQueue1ToExchange() { System.out.println("【【【綁定隊列一到交換機成功】】】"); return BindingBuilder.bind(queue1()).to(fanoutExchanges()); } }
上邊這個用來創建交換機和隊列,并將隊列和交換機進行綁定
下邊這個是消息生產者
package com.apps.rabbit.FanoutMessage; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; @Component public class Sender { @Autowired private RabbitTemplate rabbitTemplate; /** * 發送消息 * * @param message 消息內容 * 說明: routingKey可以指定也可以不指定,這里我們給的是隊列名 */ public void sendMessage(Object message) { System.out.println("【消息發送者】發送消息到fanout交換機,消息內容為: {}"+message); rabbitTemplate.convertAndSend(Constants.FANOUT_EXCHANGE_NAME, Constants.TEST_QUEUE1_NAME, message); } }
再添加一個輔助類,存常量(別問為啥,大家都是這么做的哈哈哈哈)
package com.apps.rabbit.FanoutMessage; public class Constants { /** * 交換機名稱 */ public static final String FANOUT_EXCHANGE_NAME = "message.fanout.exchange"; /** * 測試隊列名稱1 */ public static final String TEST_QUEUE1_NAME = "messageReceiverQueue"; }
最后是消息的消費者
package com.apps.rabbit.FanoutMessage; import com.alibaba.fastjson.JSONArray; import com.alibaba.fastjson.JSONException; import com.alibaba.fastjson.JSONObject; import com.apps.bcodemsg.MsgResponse; import com.apps.model.basic.message.InfoMessageUser; import com.apps.model.basic.message.InfoMessageUserVO; import com.apps.service.basic.message.InfoMessageUserService; import com.rabbitmq.client.Channel; import org.apache.commons.lang.StringUtils; import org.springframework.amqp.core.Message; import org.springframework.amqp.rabbit.annotation.RabbitHandler; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import java.io.IOException; import java.util.ArrayList; import java.util.List; @Component public class Receiver1 { @Autowired private InfoMessageUserService infoMessageUserService; private static String CONTENT_KEY="message"; private static String MESSAGE_ID_KEY="messageId"; private static String USER_ID_KEY="userId"; @RabbitListener(queues = "messageReceiverQueue") @RabbitHandler public void receiveMessage(String messageInfo, Channel channel, Message message) throws IOException { doSomething(); } }
到此就完事了,只需要在你的業務中調用生產者就行了,消息會被隊列自動接收,而接收到的消息需要怎么處理,就是業務邏輯的事了。
上邊只是一個最簡單的例子,也只用了一個隊列,如果想要添加多個隊列,需要在配置類中進行聲明,并進行綁定,此外,在調用生產者的時候,需要將rabbitTemplate.convertAndSend()方法的第二個參數去掉。
ok!
到此,相信大家對“Rabbit MQ的廣播模式是什么意思”有了更深的了解,不妨來實際操作一番吧!這里是億速云網站,更多相關內容可以進入相關頻道進行查詢,關注我們,繼續學習!
免責聲明:本站發布的內容(圖片、視頻和文字)以原創、轉載和分享為主,文章觀點不代表本網站立場,如果涉及侵權請聯系站長郵箱:is@yisu.com進行舉報,并提供相關證據,一經查實,將立刻刪除涉嫌侵權內容。