SpringBoot學(xué)習(xí)（六）——springboot快速整合RabbitMQ

Rabbit MQ消息隊(duì)列

@[toc]

為寧陵等地區(qū)用戶提供了全套網(wǎng)頁(yè)設(shè)計(jì)制作服務(wù)，及寧陵網(wǎng)站建設(shè)行業(yè)解決方案。主營(yíng)業(yè)務(wù)為網(wǎng)站設(shè)計(jì)制作、成都網(wǎng)站設(shè)計(jì)、寧陵網(wǎng)站設(shè)計(jì)，以傳統(tǒng)方式定制建設(shè)網(wǎng)站，并提供域名空間備案等一條龍服務(wù)，秉承以專業(yè)、用心的態(tài)度為用戶提供真誠(chéng)的服務(wù)。我們深信只要達(dá)到每一位用戶的要求，就會(huì)得到認(rèn)可，從而選擇與我們長(zhǎng)期合作。這樣，我們也可以走得更遠(yuǎn)！

簡(jiǎn)介

優(yōu)點(diǎn)

erlang開(kāi)發(fā)，并發(fā)能力強(qiáng)。

社區(qū)活躍，使用的人多，穩(wěn)定性較強(qiáng)。

延時(shí)低

缺點(diǎn)

erlang語(yǔ)言開(kāi)發(fā)的，國(guó)內(nèi)精通的不多，日后定制開(kāi)發(fā)困難。

Rabbit MQ工作模式

1，"Hello World!"模式

簡(jiǎn)單模式是Rabbit MQ最簡(jiǎn)單入門(mén)的案例，其中一個(gè)生產(chǎn)者，一個(gè)消費(fèi)者。無(wú)需聲明交換機(jī)（其實(shí)是有個(gè)默認(rèn)的交換機(jī)），聲明完一個(gè)隊(duì)列以后，生產(chǎn)者往這里發(fā)送，消費(fèi)者則監(jiān)聽(tīng)該隊(duì)列，對(duì)消息做出響應(yīng)。

應(yīng)用場(chǎng)景：如兩個(gè)用戶之間的聊天。

2，Work queues 模式

工作隊(duì)列模式在工人之間分配任務(wù)（競(jìng)爭(zhēng)的消費(fèi)者模式），就是有多個(gè)消費(fèi)者的簡(jiǎn)單模式，多個(gè)消費(fèi)者一起監(jiān)聽(tīng)該隊(duì)列，Rabbit MQ 會(huì)確保隊(duì)列中的某一條消息只會(huì)被某一個(gè)消費(fèi)者消費(fèi)。

應(yīng)用場(chǎng)景：如多個(gè)用戶搶某個(gè)訂單（高并發(fā)下?tīng)?zhēng)搶同一個(gè)資源的記得考慮加鎖）

3，Publish/Subscribe 模式

發(fā)布/訂閱模式一次向許多消費(fèi)者發(fā)送消息，還是一個(gè)生產(chǎn)者多個(gè)消費(fèi)者，但是該模式需要顯示聲明交換機(jī)了，將創(chuàng)建的多個(gè)隊(duì)列與該交換機(jī)進(jìn)行綁定，生產(chǎn)者將消息發(fā)給交換機(jī)，由交換機(jī)發(fā)給已經(jīng)綁定好的隊(duì)列，對(duì)應(yīng)的某隊(duì)列的消費(fèi)者消費(fèi)該隊(duì)列中的信息。

應(yīng)用場(chǎng)景：群發(fā)某個(gè)郵件，推送某條廣告

4，Routing 模式

路由模式是有選擇地接收消息，還是顯示聲明交換機(jī)，將隊(duì)列與交換機(jī)綁定，由生產(chǎn)者將消息發(fā)給交換機(jī)，由交換機(jī)發(fā)給對(duì)應(yīng)的隊(duì)列，由對(duì)應(yīng)的消費(fèi)者消費(fèi)信息。但是隊(duì)列綁定交換機(jī)時(shí)需要指定一個(gè)routingkey ，這樣的話，生產(chǎn)者再向生產(chǎn)者發(fā)消息的時(shí)候，會(huì)判斷具體的 routingkey 值，并將符合對(duì)應(yīng) routingkey 值的消息發(fā)送給對(duì)應(yīng)的隊(duì)列。

應(yīng)用場(chǎng)景：封裝若干類錯(cuò)誤類型通知

5，Topics 模式

主題模式根據(jù)模式（主題）接收消息，同路由模式一樣，只不過(guò)不再是根據(jù)具體的 routingkey 值做判斷，是根據(jù) routingkey 值模糊匹配的。其中，星號(hào)代表多個(gè)單詞 , 井號(hào)代表一個(gè)單詞。

更加詳細(xì)的資料可訪問(wèn)官網(wǎng) ： RabbitMQ官網(wǎng)

交換機(jī)模式

fanout模式：不處理路由鍵，即沒(méi)有 routingkey 值，很像子網(wǎng)廣播，每臺(tái)子網(wǎng)內(nèi)的主機(jī)都獲得了一份復(fù)制的消息。

direct模式：處理路由鍵，需要路由鍵匹配才能轉(zhuǎn)發(fā)，即完全匹配上 routingkey 值。

topic模式：處理路由鍵，需要路由鍵模糊匹配才能轉(zhuǎn)發(fā)，即模糊匹配上 routingkey 值。

引入RabbitMQ隊(duì)列

RabbitMQ 依賴與 erlang 語(yǔ)言，點(diǎn)擊 官網(wǎng)下載鏈接 下載 erlang。

點(diǎn)擊 官網(wǎng)下載鏈接 下載RabbitMQ。

安裝兩者的時(shí)候，一路默認(rèn)設(shè)置即可。

啟動(dòng) RabbitMQ 服務(wù)，cmd模式下輸入，進(jìn)入默認(rèn)安裝路徑

C:\Program Files (x86)\RabbitMQ Server\rabbitmq_server-3.6.10\sbin

輸入命令

rabbitmq-plugins enable rabbitmq_management

服務(wù)啟動(dòng)成功，如圖所示

點(diǎn)擊 http://localhost:15672/ ，如下圖所示

pom.xml中新增

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

application.properties中加入

# RabbitMQ
spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
# 并發(fā)消費(fèi)者的初始化值
spring.rabbitmq.listener.concurrency=10
# 并發(fā)消費(fèi)者的最大值
spring.rabbitmq.listener.max-concurrency=20
# 每個(gè)消費(fèi)者每次監(jiān)聽(tīng)時(shí)可拉取處理的消息數(shù)量
spring.rabbitmq.listener.prefetch=5

代碼實(shí)戰(zhàn)

簡(jiǎn)單模式和工作隊(duì)列模式?jīng)]有指定交換機(jī)，rabbitMQ的精妙之處不能體現(xiàn)出來(lái)，而路由模式和主體模式都有 routingkey 值，故挑其一，本文以發(fā)布/訂閱模式和路由模式為例，其余的都可類比。不全寫(xiě)，是因?yàn)檫@是快速上手使用系列，最短的時(shí)間精力明白來(lái)龍去脈，才是該系列的宗旨，人類畏懼晦澀冗長(zhǎng)的，喜歡簡(jiǎn)短直接的。

新增代碼后的目錄如下所示

RabbitConfig.java

package com.example.config;

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitConfig {

    //==============發(fā)布/訂閱模式==============

    //隊(duì)列A
    @Bean()
    public Queue queueA() {
        return new Queue("fanoutQueueA");
    }

    //隊(duì)列B
    @Bean()
    public Queue queueB() {
        return new Queue("fanoutQueueB");
    }

    //發(fā)布訂閱模式下的 fanout 交換機(jī)
    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange("fanoutExchange");
    }

    // 把 隊(duì)列A 和 fanout交換機(jī) 綁定在一起
    @Bean
    public Binding bindingWithQueueA() {
        return BindingBuilder.bind(queueA()).to(fanoutExchange());
    }

    // 把 隊(duì)列B 和 fanout交換機(jī) 綁定在一起
    @Bean
    public Binding bindingWithQueueB() {
        return BindingBuilder.bind(queueB()).to(fanoutExchange());
    }

    //==============路由模式==============
    //隊(duì)列C
    @Bean()
    public Queue queueC() {
        return new Queue("directQueueC");
    }

    //隊(duì)列D
    @Bean()
    public Queue queueD() {
        return new Queue("directQueueD");
    }

    //發(fā)布訂閱模式下的 direct 交換機(jī)
    @Bean
    public DirectExchange directExchange() {
        return new DirectExchange("directExchange");
    }

    // 把 隊(duì)列C 和 direct交換機(jī) 綁定在一起
    @Bean
    public Binding bindingWithQueueC() {
        return BindingBuilder.bind(queueC()).to(directExchange()).with("directRoutingkey");
    }

    // 把 隊(duì)列B 和 direct交換機(jī) 綁定在一起
    @Bean
    public Binding bindingWithQueueD() {
        return BindingBuilder.bind(queueD()).to(directExchange()).with("directRoutingkey");
    }

}

RabbitController.java

package com.example.controller;

import com.example.service.IRabbitProducerService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;

@Controller
@RequestMapping("rabbit")
public class RabbitController {

    @Autowired
    private IRabbitProducerService rabbitProducerService;

    @RequestMapping("/fanout")
    public void fanout() {
        for (int i = 0; i < 5; i++) {
            this.rabbitProducerService.producerFanout("發(fā)布訂閱模式下的第" + i + "條信息");
        }
    }

    @RequestMapping("/direct")
    public void direct() {
        for (int i = 0; i < 5; i++) {
            this.rabbitProducerService.producerDirect("路由模式下的第" + i + "條信息");
        }
    }

}

IRabbitProducerService.java

package com.example.service;

public interface IRabbitProducerService {

    void producerFanout(String message);

    void producerDirect(String message);

}

RabbitProducerServiceIml.java

package com.example.service;

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service("rabbitProducerService")
public class RabbitProducerServiceIml implements IRabbitProducerService {

    @Autowired
    RabbitTemplate rabbitTemplate;

    public void producerFanout(String message) {
        rabbitTemplate.convertAndSend("fanoutExchange", null, message);
    }

    public void producerDirect(String message) {
        rabbitTemplate.convertAndSend("directExchange", "directRoutingkey", message);
    }

}

IRabbitConsumerService.java

package com.example.service;

public interface IRabbitConsumerService {

    void consumerFanoutA(String message);

    void consumerFanoutB(String message);

    void consumerDirectC(String message);

    void consumerDirectD(String message);

}

RabbitConsumerServiceIml.java

package com.example.service;

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class RabbitConsumerServiceIml implements IRabbitConsumerService {

    @RabbitListener(queues = "fanoutQueueA")
    public void consumerFanoutA(String message) {
        System.out.println("消費(fèi)者收到隊(duì)列A中的消息:" + message);
    }

    @RabbitListener(queues = "fanoutQueueB")
    public void consumerFanoutB(String message) {
        System.out.println("消費(fèi)者收到隊(duì)列B中的消息:" + message);
    }

    @RabbitListener(queues = "routingQueueC")
    public void consumerDirectC(String message) {
        System.out.println("消費(fèi)者收到隊(duì)列C中的消息:" + message);
    }

    @RabbitListener(queues = "routingQueueD")
    public void consumerDirectD(String message) {
        System.out.println("消費(fèi)者收到隊(duì)列D中的消息:" + message);
    }

}

啟動(dòng)項(xiàng)目后，瀏覽器輸入 http://localhost:8080/rabbit/fanout ，（雖然前臺(tái)會(huì)報(bào)錯(cuò)，因?yàn)檫@個(gè)請(qǐng)求只是觸發(fā)消息的生產(chǎn)，并沒(méi)有對(duì)應(yīng)的界面展示。生產(chǎn)出來(lái)的消息經(jīng)過(guò)交換機(jī)傳給隊(duì)列，消費(fèi)者則監(jiān)聽(tīng)隊(duì)列，做出對(duì)應(yīng)的響應(yīng)。）發(fā)布/訂閱模式下使用 fanout 交換機(jī)生產(chǎn)消費(fèi)消息的效果圖如下所示；

瀏覽器輸入 http://localhost:8080/rabbit/direct ，發(fā)路由模式下使用 direct 交換機(jī)生產(chǎn)消費(fèi)消息的效果圖如下所示；

注：如果是根據(jù)本文系列文章來(lái)的，因?yàn)橐婚_(kāi)始就配置好了spring security，所以記得將該地址配給所登錄的用戶?；蛘唛_(kāi)一個(gè)超級(jí)管理員賬號(hào)，可以訪問(wèn)項(xiàng)目的任意目錄，使用該管理員賬號(hào)訪問(wèn)這些地址。

到這里，篇幅已經(jīng)很長(zhǎng)了，本來(lái)想準(zhǔn)備寫(xiě)完死信隊(duì)列，和消息確認(rèn)機(jī)制。想想這是快速整合上手系列，就把深入的內(nèi)容放到踩坑深入系列，盡情期待后續(xù)開(kāi)辟新的系列文章吧。

當(dāng)前題目：SpringBoot學(xué)習(xí)（六）——springboot快速整合RabbitMQ
網(wǎng)頁(yè)鏈接：http://bm7419.com/article4/gocpie.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供手機(jī)網(wǎng)站建設(shè)、網(wǎng)站策劃、云服務(wù)器、商城網(wǎng)站、網(wǎng)站設(shè)計(jì)、網(wǎng)站設(shè)計(jì)公司

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請(qǐng)盡快告知，我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng)，如需處理請(qǐng)聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時(shí)需注明來(lái)源： 創(chuàng)新互聯(lián)