消息隊(duì)列RabbitMQ.01.基本使用

這篇具有很好參考價(jià)值的文章主要介紹了消息隊(duì)列RabbitMQ.01.基本使用。希望對(duì)大家有所幫助。如果存在錯(cuò)誤或未考慮完全的地方，請(qǐng)大家不吝賜教，您也可以點(diǎn)擊"舉報(bào)違法"按鈕提交疑問。

?RabbitMQ的作用

Message queue 釋義

?問題思考

?存在的問題

優(yōu)化方案?

案例分析?

帶來的好處?

消息隊(duì)列特點(diǎn)

?Email郵件案例分析

Docker安裝部署RabbitMQ

1.下拉鏡像

2.運(yùn)行RabbitMQ

?3.打開防火墻端口號(hào)并重新運(yùn)行防火墻

4.容器啟動(dòng)后,可以通過 docker logs 容器查看日志

6.通過剛才填寫的用戶名和密碼登錄 admin?

?springboot連接配置

1.配置spring賬號(hào)

2.創(chuàng)建兩個(gè)springboot項(xiàng)目publisher consumer 選擇依賴?

3.配置yml文件?

4.生產(chǎn)者配置類

5.生成者測(cè)試類

6.消費(fèi)者接受信息

消息隊(duì)列RabbitMQ.01.基本使用,網(wǎng)絡(luò),rabbitmq

?RabbitMQ的作用

Message queue 釋義

服務(wù)之間最常見的通信方式是直接調(diào)用彼此來通信 , 消息從一端發(fā)出后立即就可以達(dá)到另一端 , 稱為即時(shí)消息通訊 ( 同步通信 )

消息從某一端發(fā)出后 , 首先進(jìn)入一個(gè)容器進(jìn)行臨時(shí)存儲(chǔ) , 當(dāng)達(dá)到某種條件后 , 再由這個(gè)容器發(fā)送給另一端 , 稱為延遲消息通訊 ( 異步通信 )

?問題思考

假設(shè)我們?cè)谔詫毾铝艘还P訂單后 , 淘寶后臺(tái)需要做這些事情：

?消息通知系統(tǒng)：通知商家,你有一筆新的訂單,請(qǐng)及時(shí)發(fā)貨

?推薦系統(tǒng)：更新用戶畫像,重新給用戶推薦他可能感興趣的商品

?會(huì)員系統(tǒng)：更新用戶的積分和等級(jí)信息

createOrder(...) {
// 完成訂單服務(wù)
doCreateOrder(...);
// 調(diào)用其他服務(wù)接口
sendMsg(...);
updateUserInterestedGoods(...);
updateMemberCreditInfo(...);
}

?存在的問題

過度耦合：如果后面創(chuàng)建訂單時(shí),需要觸發(fā)新的動(dòng)作,那就得去改代碼,在原有的創(chuàng)建訂單函數(shù)末尾,再追加一行代碼

缺少緩沖：如果創(chuàng)建訂單時(shí),會(huì)員系統(tǒng)恰好處于非常忙碌或者宕機(jī)的狀態(tài),那這時(shí)更新會(huì)員信息就會(huì)失敗,我們需要一個(gè)地方,來暫時(shí)存放無法被消費(fèi)的消息

優(yōu)化方案?

我們需要一個(gè)消息中間件 , 來實(shí)現(xiàn)解耦和緩沖的功能 .

消息隊(duì)列RabbitMQ.01.基本使用,網(wǎng)絡(luò),rabbitmq

Server(Broker): 接收客戶端連接 , 實(shí)現(xiàn) AMQP 協(xié)議的消息隊(duì)列和路由功能的進(jìn)程 .

Virtual Host ：虛擬主機(jī)的概念 , 類似權(quán)限控制組 , 一個(gè) Virtual Host 里可以有多個(gè) Exchange 和 Queue.

Exchange: 交換機(jī) , 接收生產(chǎn)者發(fā)送的消息 , 并根據(jù) Routing Key 將消息路由到服務(wù)器中的隊(duì)列 Queue.

ExchangeType: 交換機(jī)類型決定了路由消息行為 ,RabbitMQ 中有三種類型 Exchange, 分別是 fanout 、 direct 、 topic.

Message Queue ：消息隊(duì)列 , 用于存儲(chǔ)還未被消費(fèi)者消費(fèi)的消息 .

Message ：由 Header 和 body 組成 ,Header 是由生產(chǎn)者添加的各種屬性的集合 , 包括 Message 是否被持久化、優(yōu)先級(jí)是多少、由哪個(gè) Message Queue 接收等 .body 是真正需要發(fā)送的數(shù)據(jù)內(nèi)

容 .

BindingKey ：綁定關(guān)鍵字 , 將一個(gè)特定的 Exchange 和一個(gè)特定的 Queue 綁定起來 .

案例分析?

小紅希望小明多讀書 , 常尋找好書給小明看 , 之前的方式是這樣：小紅問小明什么時(shí)候有空 , 把書給小明送去 , 并親眼監(jiān)督小明讀完書才走 . 久而久之 , 兩人都覺得麻煩 .

后來的方式改成了：小紅對(duì)小明說「我放到書架上的書你都要看」 , 然后小紅每次發(fā)現(xiàn)不錯(cuò)的書都放到書架上 , 小明則看到書架上有書就拿下來看 .

書架就是一個(gè)消息隊(duì)列 , 小紅是生產(chǎn)者 , 小明是消費(fèi)者

帶來的好處?

1. 小紅想給小明書的時(shí)候 , 不必問小明什么時(shí)候有空 , 親手把書交給他了 , 小紅只把書放到書架上就行了 . 這樣小紅小明的時(shí)間都更自由 .

2. 小紅相信小明的讀書自覺和讀書能力 , 不必親眼觀察小明的讀書過程 , 小紅只要做一個(gè)放書的動(dòng)作 , 很節(jié)省時(shí)間 .

3. 當(dāng)明天有另一個(gè)愛讀書的小伙伴小強(qiáng)加入 , 小紅仍舊只需要把書放到書架上 , 小明和小強(qiáng)從書架上取書即可

4. 書架上的書放在那里 , 小明閱讀速度快就早點(diǎn)看完 , 閱讀速度慢就晚點(diǎn)看完 , 沒關(guān)系 , 比起小紅把書遞給小明并監(jiān)督小明讀完的方式 , 小明的壓力會(huì)小一些 .

消息隊(duì)列特點(diǎn)

1. 解耦 : 每個(gè)成員不必受其他成員影響 , 可以更獨(dú)立自主 , 只通過一個(gè)簡(jiǎn)單的容器來聯(lián)系 .

2. 提速 : 小紅選只要做一個(gè)放書的動(dòng)作 , 為自己節(jié)省了大量時(shí)間 .

3. 廣播 : 小紅只需要?jiǎng)趧?dòng)一次 , 就可以讓多個(gè)小伙伴有書可讀 , 這大大地節(jié)省了她的時(shí)間 , 也讓新的小伙伴的加入成本很低 .

4. 錯(cuò)峰與流控 : 小紅給書的頻率不穩(wěn)定 , 如果今明兩天連給了五本 , 之后隔三個(gè)月才又給一本 , 那小明只要在三個(gè)月內(nèi)從書架上陸續(xù)取走五本書讀完就行了 , 壓力就不那么大了 .

?Email郵件案例分析

有大量用戶注冊(cè)你的軟件 , 再高并發(fā)情況下注冊(cè)請(qǐng)求開始出現(xiàn)一些問題 .

例如郵件接口承受不住 , 或是分析信息時(shí)的大量計(jì)算使 cpu 滿載 , 這將會(huì)出現(xiàn)雖然用戶數(shù)據(jù)記錄很快的添加到數(shù)據(jù)庫(kù)中了 , 但是卻卡在發(fā)郵件或分析信息時(shí)的情況 .

導(dǎo)致請(qǐng)求的響應(yīng)時(shí)間大幅增長(zhǎng) , 甚至出現(xiàn)超時(shí) , 這就有點(diǎn)不劃算了 . 面對(duì)這種情況一般也是將這些操作放入消息隊(duì)列 ( 生產(chǎn)者消費(fèi)者模型 ), 消息隊(duì)列慢慢的進(jìn)行處理 , 同時(shí)可以很快的完成注冊(cè)請(qǐng)

求 , 不會(huì)影響用戶使用其他功能

Docker安裝部署RabbitMQ

1.下拉鏡像

docker pull rabbitmq:management

注意獲取鏡像的時(shí)候要獲取 management 版本的 , 不要獲取 last 版本的 ,management 版本的才帶有管理界面

2.運(yùn)行RabbitMQ

docker run -itd \
--name my-rabbitmq \
-p 5672:5672 -p 15672:15672 \
--hostname my-rabbitmq-host \
-e RABBITMQ_DEFAULT_VHOST=my_vhost \
-e RABBITMQ_DEFAULT_USER=admin \
-e RABBITMQ_DEFAULT_PASS=admin \
--restart=always \
rabbitmq:management

--hostname ：主機(jī)名 (RabbitMQ 的一個(gè)重要注意事項(xiàng)是它根據(jù)所謂的 “ 節(jié)點(diǎn)名稱 ” 存儲(chǔ)數(shù)據(jù) , 默認(rèn)為主機(jī)名 )

-e ：指定環(huán)境變量 :

RABBITMQ_DEFAULT_VHOST ：默認(rèn)虛擬機(jī)名

RABBITMQ_DEFAULT_USER ：默認(rèn)的用戶名

RABBITMQ_DEFAULT_PASS ：默認(rèn)用戶名的密碼

?3.打開防火墻端口號(hào)并重新運(yùn)行防火墻

firewall-cmd --zone=public --add-port=15672/tcp --permanent

firewall-cmd --reload

4.容器啟動(dòng)后,可以通過 docker logs 容器查看日志

docker logs my-rabbitmq

5.通過主機(jī)網(wǎng)址進(jìn)入管理后臺(tái)

虛擬機(jī)ip地址/15672

消息隊(duì)列RabbitMQ.01.基本使用,網(wǎng)絡(luò),rabbitmq

6.通過剛才填寫的用戶名和密碼登錄 admin?

消息隊(duì)列RabbitMQ.01.基本使用,網(wǎng)絡(luò),rabbitmq

?springboot連接配置

1.配置spring賬號(hào)

切記需要授權(quán)

2.創(chuàng)建兩個(gè)springboot項(xiàng)目publisher consumer 選擇依賴?

消息隊(duì)列RabbitMQ.01.基本使用,網(wǎng)絡(luò),rabbitmq

3.配置yml文件?

#publisher
server:
    port: 8888
spring:
    rabbitmq:
        host: 192.168.241.130
        username: spring
        password: 123456
        port: 5672
        virtual-host: my_vhost

#consumer
server:
    port: 9999
spring:
    rabbitmq:
        host: 192.168.241.130
        username: spring
        password: 123456
        port: 5672
        virtual-host: my_vhost

4.生產(chǎn)者配置類

@Configuration
@SuppressWarnings("all")
public class RabbitConfig {
    @Bean
    public Queue firstQueue() {
        return new Queue("firstQueue");
    }

    @Bean
    public Queue secondQueue() {
        return new Queue("secondQueue");
    }
}

5.生成者測(cè)試類

public class TestController {

    @Autowired
    private AmqpTemplate template;
    @Autowired
    private ObjectMapper objectMapper;

    @RequestMapping("/send1")
    public String send1(){
        //向消息隊(duì)列發(fā)送消息
        template.convertAndSend("firstQueue","hello world");
        return "??";
    }

    @RequestMapping("/send2")
    public String send2() throws Exception{
        User jack = new User("jack", "123");
        String json = objectMapper.writeValueAsString(jack);
        //向消息隊(duì)列發(fā)送消息
        template.convertAndSend("secondQueue",jack);
        return "??";
    }
}

因?yàn)橄㈥?duì)列支持的對(duì)象傳參必須consumer 和 publisher 兩個(gè)項(xiàng)目的pojo包路徑完全一致所以使用：

@Autowired
private ObjectMapper objectMapper;

User jack = new User("jack", "123");

記得拋出異常 throws Exception 不然會(huì)報(bào)錯(cuò)文章來源地址http://www.zghlxwxcb.cn/news/detail-810978.html

6.消費(fèi)者接受信息

@Component
@SuppressWarnings("all")
@Slf4j
@RabbitListener(queues = "firstQueue")
public class Receiver {

    @RabbitHandler
    public void process(String msg) {
        log.warn("接收到:" + msg);
    }
}

@Component
@SuppressWarnings("all")
@Slf4j
@RabbitListener(queues = "secondQueue")
public class PojoReceiver {

    @Autowired
    private ObjectMapper objectMapper;

    @RabbitHandler
    public void process(String json) throws Exception{
        User user=objectMapper.readValue(json,User.class);
        log.warn("接收到:" + json);
    }
}

到了這里，關(guān)于消息隊(duì)列RabbitMQ.01.基本使用的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！