Spring Boot項目使用RabbitMQ隊列

一、Rabbitmq的安裝

RabbitMQ是一個開源的遵循 AMQP協(xié)議實現(xiàn)的基于 Erlang語言編寫，**即需要先安裝部署Erlang環(huán)境再安裝RabbitMQ環(huán)境。

erlang的安裝在windows中直接點擊安裝即可。

安裝完erlang后設(shè)置erlang的環(huán)境變量ERLANG_HOME。

然后安裝rabbitmq。

安裝成功后。

可以在瀏覽器中輸入http://localhost:15672,訪問地址顯示的結(jié)果如下。

?

這里的username輸入“guest”， password輸入“guest”，然后點擊“l(fā)ogin”，之后進(jìn)入rabbitmq的界面。如下圖所示。

?

注意：這里rabbitmq的默認(rèn)網(wǎng)頁訪問端口15672，rabbitmq的訪問端口是5672。新建的sprint boot項目需要進(jìn)行交互的端口號就是5672。

下面新建一個Spring boot項目。

二、Spring boot新建rabbitmq項目

1、點擊“ File”-->“New”---->“Project”。如下圖。

?

2、點擊“Project”后，就可以彈出對話框。

?

3、注意對話框中java的版本。然后點擊“Next”。在彈出的對話框中選擇Spring Boot的版本和Spring Web的模塊。如下圖所示。

?4、選擇成功后，直接點擊“Finish”完成項目的構(gòu)建。

?5、這里可以選擇“New Window”，在新窗口中打開項目，然后在pom.xml中添加依賴。rabbitmq的依賴需要使用spring-boot-starter-amqp，因為這里還需要lombok。添加的依賴代碼如下。

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</artifactId>
    <version>1.18.0</version>
</dependency>

6、pom.xml中添加依賴后的代碼如下圖所示。

?

7、編寫application.yml文件，可以把a(bǔ)pplication.properties修改成application.yml。

?

8、在application.yml中設(shè)置spring boot連接rabbit的配置。代碼如下。

?

這里的port是默認(rèn)的端口，用戶名和密碼是默認(rèn)的guest，virtual-host可以通過rabbitmq的web界面中查看出來。如下圖。

?

下面需要進(jìn)行rabbitmq的開發(fā)設(shè)置。

三、rabbitmq的開發(fā)。

rabbitmq是部署最廣泛的開源消息代理。

其是一個消息隊列，消費隊列具備：

代碼解耦,提高系統(tǒng)穩(wěn)定性

應(yīng)對流量高峰,降低流量沖擊，面對秒殺這種情況時，請求進(jìn)來先去排隊，可以保證系統(tǒng)的穩(wěn)定

異步執(zhí)行,提高系統(tǒng)響應(yīng)速度

因為是消息隊列，必然有生產(chǎn)者和消費者，其工作模式最簡單的類型是簡單類型，也就是消息只能有一個被消費，原理圖如下。

?

圖中有兩個消費者，因為消息只能有一個被消費，會造成消息競爭，因此此方式不適合于多個消費者共享生產(chǎn)者的消息，需要使用發(fā)布訂閱模式。

發(fā)布訂閱模式這里有exchange交換機(jī)，生產(chǎn)者將消息給交換機(jī)，交換機(jī)根據(jù)自身的類型（fanout）將會把所有消息復(fù)制同步到所有與其綁定的隊列，每個隊列可以有一個消費者接收消息進(jìn)行消費邏輯。原理如下圖。

?

Spring boot中可以使用amqp的ConvertAndSendReceive方法來現(xiàn)controller的控制邏輯，其意義：按照一定的順序，只有確定消費者接收到消息，才會發(fā)送下一條信息，每條消息之間會有間隔時間。

代碼如下。

package com.example.demo.controller;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
@RestController
public class RabbitMqController {
    @Autowired
    private RabbitTemplate rabbit;
    @RequestMapping("/rabbitmq/sendPublisher")
    public String sendPublisher(){
        for(int i=0;i<20;i++){
            rabbit.convertSendAndReceive("exchange_fanout","","測試發(fā)布訂閱模式:"+i);
        }
        return "發(fā)送成功....";
    }
}

這里定義一個restful接口，接口請求sendPublisher傳輸生產(chǎn)者的接口方法，在邏輯中使用convertSendAndReceive方法使用交換器進(jìn)行發(fā)送和接收，方法中第一個參數(shù)是主題，后面是具體生產(chǎn)者發(fā)送的相關(guān)信息。

當(dāng)生產(chǎn)者發(fā)送信息后，需要消費者進(jìn)行消費，消費者的消費是通過監(jiān)聽器完成的，通過listener來控制的。接下來需要寫Listener監(jiān)聽器的代碼。

監(jiān)聽器的代碼必須定義成Component組件，也就是使用Component注解。

首先是定義一個新的交換機(jī)，主題是“ exchange_fanout”，將此返回的FanoutExchange定義成Bean。代碼如下。

@Bean
public FanoutExchange exchangeFanout(){
    return new FanoutExchange("exchange_fanout");
}

這里定義兩個消費者，每個消費者都需要綁定這個交換機(jī)，定義的兩個消費者也需要定義成Bean。并且和exchange_fanout進(jìn)行綁定，代碼如下。

@Bean
public Queue queueFanout1(){
    return new Queue("queue_fanout1");
}
@Bean
public Queue queueFanout2(){
    return new Queue("queue_fanout2");
}
@Bean
public Binding bindingExchange1(Queue queueFanout1,FanoutExchange exchangeFanout){
    return BindingBuilder.bind(queueFanout1).to(exchangeFanout);
}
@Bean
public Binding bindingExchange2(Queue queueFanout2,FanoutExchange exchangeFanout){
    return BindingBuilder.bind(queueFanout2).to(exchangeFanout);
}

這里的bind方法實現(xiàn)的是消費者，交換機(jī)及生產(chǎn)者之間的綁定。

接下來通過RabbitListener的Rabbitmq的監(jiān)聽器來獲取消息。

代碼如下。

@RabbitListener(queues="queue_fanout1")
public void receiverMsg1(String msg){
    System.out.println("隊列1接收到的信息:"+msg);
}
@RabbitListener(queues="queue_fanout2")
public void receiverMsg2(String msg){
    System.out.println("隊列2接收到的信息:"+msg);
}

注意每一個RabbitListener的參數(shù)queues隊列中的兩個隊列名稱，這兩個隊列都是隊列bean中產(chǎn)生的消息。

整體監(jiān)聽器的代碼如下。

package com.example.demo.listener;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.context.annotation.Bean;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.TopicExchange;
@Slf4j
@Component
public class RabbitMqConsumer {
    @Bean
    public Queue queueFanout1(){
        return new Queue("queue_fanout1");
    }
    @Bean
    public Queue queueFanout2(){
        return new Queue("queue_fanout2");
    }
    @Bean
    public FanoutExchange exchangeFanout(){
        return new FanoutExchange("exchange_fanout");
    }
    @Bean
    public Binding bindingExchange1(Queue queueFanout1,FanoutExchange exchangeFanout){
        return BindingBuilder.bind(queueFanout1).to(exchangeFanout);
    }
    @Bean
    public Binding bindingExchange2(Queue queueFanout2,FanoutExchange exchangeFanout){
        return BindingBuilder.bind(queueFanout2).to(exchangeFanout);
    }
    @RabbitListener(queues="queue_fanout1")
    public void receiverMsg1(String msg){
        System.out.println("隊列1接收到的信息:"+msg);
    }
    @RabbitListener(queues="queue_fanout2")
    public void receiverMsg2(String msg){
        System.out.println("隊列2接收到的信息:"+msg);
    }
}

啟動服務(wù)后，當(dāng)瀏覽器訪問“/rabbitmq/sendPublisher”后，在瀏覽器中顯示文字“發(fā)送成功”，在控制臺輸出隊列1和隊列2的相關(guān)信息。瀏覽器顯示如下圖。

?

控制臺輸出結(jié)果如下圖。

?

這里有兩個隊列信息，輸出的都是1-20的相關(guān)數(shù)字。

也可以把1-20個數(shù)字分為奇數(shù)和偶數(shù)，訂閱不同的主題來消費。

這里，在控制器中實現(xiàn)奇數(shù)和偶數(shù)發(fā)送不同的主題信息。代碼如下。

@RequestMapping("/rabbitmq/sendTopic")
public String sendtopic(){
    for(int i=0;i<20;i++){
        if(i%2==0){
            rabbit.convertAndSend("exchange_topic","topic.fat.topic","測試發(fā)布訂閱模型:"+i);
        }else{
            rabbit.convertAndSend("exchange_topic","topic.fat","測試發(fā)布訂閱模型："+i);
        }

    }
    return "發(fā)布成功.....";
}

生產(chǎn)者發(fā)送信息后，雖然交換的主題都是一個主題，但是路由的內(nèi)容不同，接下來需要通過Listener監(jiān)聽器進(jìn)行信息收集。

在Listener監(jiān)聽器中需要定義Topic的Exchange交換機(jī)，代碼如下。

@Bean
public TopicExchange exchangeTopic(){
    return new TopicExchange("exchange_topic");
}

這個exchangeTopic也需要定義成Bean。還需要根據(jù)兩個路由規(guī)則產(chǎn)生兩個綁定規(guī)則，不同的路由規(guī)則產(chǎn)生不同的消費信息結(jié)果，需要定義兩個不同的消費內(nèi)容。

@Bean
public Queue queueTopic1(){
    return new Queue("queue_topic1");
}
@Bean
public Queue queueTopic2(){
    return new Queue("queue_topic2");
}

接下來通過bind方法綁定主題和生產(chǎn)者路由規(guī)則之間的聯(lián)系，代碼如下。

@Bean
public Binding bindingTopic1(Queue queueTopic1,TopicExchange exchangeTopic){
    return BindingBuilder.bind(queueTopic1).to(exchangeTopic).with("topic.#");
}
@Bean
public Binding bindingTopic2(Queue queueTopic2,TopicExchange exchangeTopic){
    return BindingBuilder.bind(queueTopic2).to(exchangeTopic).with("topic.*");
}

這里再通過RabbitListener監(jiān)聽器注解根據(jù)不同的主題，輸出不同的信息，代碼如下。

@RabbitListener(queues="queue_topic1")
public void receiverTopicMsg1(String msg){
    System.out.println("消費者1接收到:"+msg);
}
@RabbitListener(queues="queue_topic2")
public void receiverTopicMsg2(String msg){
    System.out.println("消費者2接收到:"+msg);
}

使用topic主題進(jìn)行奇數(shù)和偶數(shù)分離消費的整體代碼如下。

package com.example.demo.listener;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.context.annotation.Bean;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.TopicExchange;
@Slf4j
@Component
public class RabbitMqConsumer {
    @Bean
    public Queue queueTopic1(){
        return new Queue("queue_topic1");
    }
    @Bean
    public Queue queueTopic2(){
        return new Queue("queue_topic2");
    }
    @Bean
    public TopicExchange exchangeTopic(){
        return new TopicExchange("exchange_topic");
    }
    @Bean
    public Binding bindingTopic1(Queue queueTopic1,TopicExchange exchangeTopic){
        return BindingBuilder.bind(queueTopic1).to(exchangeTopic).with("topic.#");
    }
    @Bean
    public Binding bindingTopic2(Queue queueTopic2,TopicExchange exchangeTopic){
        return BindingBuilder.bind(queueTopic2).to(exchangeTopic).with("topic.*");
    }
    @RabbitListener(queues="queue_topic1")
    public void receiverTopicMsg1(String msg){
        System.out.println("消費者1接收到:"+msg);
    }
    @RabbitListener(queues="queue_topic2")
    public void receiverTopicMsg2(String msg){
        System.out.println("消費者2接收到:"+msg);
    }
}

運行代碼后，在瀏覽器中訪問“/rabbitmq/sendTopic”，如下圖所示。

?

控制臺輸出的結(jié)果如下圖所示。

?

至此，spring boot使用rabbitmq的基本流程結(jié)束，歡迎期待rabbitmq的實戰(zhàn)博客。文章來源地址http://www.zghlxwxcb.cn/news/detail-433647.html

到了這里，關(guān)于Spring Boot項目使用RabbitMQ隊列的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！