快捷導(dǎo)航

一篇文章教你將JAVA的RabbitMQz與SpringBoot整合

更新時(shí)間：2021年09月16日 09:39:24 作者：qiuxinfa123

這篇文章主要介紹了如何將JAVA的RabbitMQz與SpringBoot整合，小編覺(jué)得挺不錯(cuò)的，現(xiàn)在分享給大家，也給大家做個(gè)參考。一起跟隨小編過(guò)來(lái)看看吧

1.虛擬主機(jī)vhost：vhost是物理隔離的，你可以將vhost看作是一個(gè)個(gè)小型的RabbitMQ

2.交換機(jī)exchange：生產(chǎn)者發(fā)送的消息不是直接到達(dá)隊(duì)列的，而是交換機(jī)，然后交換機(jī)再根據(jù)路由key，路由到指定的隊(duì)列，可以理解為一個(gè)分發(fā)消息的角色

3.隊(duì)列：存放消息的地方，消費(fèi)者拿到的消息就是通過(guò)隊(duì)列，可以理解為存放消息的容器

4.bindingKey和routingKey：這兩個(gè)可能比較容易搞混，bindingKey是交換機(jī)和隊(duì)列建立關(guān)系時(shí)候的key，而routingKey則是生產(chǎn)者發(fā)送消息時(shí)候的key。在DirectExchange類(lèi)型的交換機(jī)中，bindingKey和routingKey是一致的，而在TopicExchange類(lèi)型的交換機(jī)中，兩者一般是不一致的。

開(kāi)發(fā)環(huán)境：RabbitMQ:3.7.7、SpringBoot2.1.7

引入相關(guān)依賴：

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
            <version>2.1.7.RELEASE</version>
        </dependency>
        <!-- rabbitmq依賴 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
            <version>2.1.7.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <version>2.1.7.RELEASE</version>
        </dependency>
    </dependencies>

在生產(chǎn)者和消費(fèi)者工程的application.yml文件添加RabbitMQ配置：

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest

如果是和上面一樣的值，其實(shí)也可以不用配置，因?yàn)镽abbitMQ的默認(rèn)配置就是它們：

接下來(lái)，看看幾種交換機(jī)的使用場(chǎng)景，包括：

（1）fanout：發(fā)布訂閱型，可以理解為廣播

（2）direct：直連型，可以理解為點(diǎn)對(duì)點(diǎn)

（3）topic：通配符模式

一、fanout：發(fā)布訂閱型

在生產(chǎn)者工程，添加如下配置：

@Configuration
public class FanoutRabbitmqConfig {
    public static final String EMAIL_QUEUE = "email_fanout";
    public static final String SMS_QUEUE = "sms_fanout";
    public static final String EXCHANGE_NAME = "exchange_fanout";
    // 發(fā)布訂閱模式，不用routingKey
    @Bean
    public FanoutExchange getFanoutExchange(){
        return new FanoutExchange(EXCHANGE_NAME);
    }
    @Bean
    public Queue getFanoutEmailQueue(){
        return new Queue(EMAIL_QUEUE);
    }
    @Bean
    public Queue getFanoutSmsQueue(){
        return new Queue(SMS_QUEUE);
    }
    // 綁定交換機(jī)和隊(duì)列
    @Bean
    public Binding fanoutEmailExchange(){
        return BindingBuilder.bind(getFanoutEmailQueue()).to(getFanoutExchange());
    }
    @Bean
    public Binding fanoutSmsExchange(){
        return BindingBuilder.bind(getFanoutSmsQueue()).to(getFanoutExchange());
    }
    // 用于測(cè)試找到了交換機(jī)，但是沒(méi)有找到隊(duì)列的情況
    @Bean
    public FanoutExchange notBindingQueue(){
        return new FanoutExchange("testNotBindingQueue");
    }
}

可以看到聲明了名為exchange_fanout的交換機(jī)和兩個(gè)隊(duì)列，并進(jìn)行了綁定，如果向exchange_fanout交換機(jī)發(fā)送消息，則兩個(gè)隊(duì)列都會(huì)收到消息。

二、direct：直連型

在生產(chǎn)者工程添加如下配置：

@Configuration
public class RoutingRabbitmqConfig {
    public static final String EMAIL_QUEUE = "email_routing";
    public static final String SMS_QUEUE = "sms_routing";
    public static final String EXCHANGE_NAME = "exchange_routing";
    public static final String EMAIL_ROUTINGKEY = "inform.email.routingKey";
    public static final String SMS_ROUTINGKEY = "inform.sms.routingKey";
    @Bean
    public DirectExchange getDirectExchange(){
        return new DirectExchange(EXCHANGE_NAME);
    }
    @Bean
    public Queue getDirectEmailQueue(){
        return new Queue(EMAIL_QUEUE);
    }
    @Bean
    public Queue getDirectSmsQueue(){
        return new Queue(SMS_QUEUE);
    }
    // 綁定交換機(jī)和隊(duì)列
    @Bean
    public Binding directEmailExchange(){
        return BindingBuilder.bind(getDirectEmailQueue()).to(getDirectExchange()).with(EMAIL_ROUTINGKEY);
    }
    @Bean
    public Binding directSmsExchange(){
        return BindingBuilder.bind(getDirectSmsQueue()).to(getDirectExchange()).with(SMS_ROUTINGKEY);
    }
}

和fanout類(lèi)型不同的是，這里使用了routingKey進(jìn)行交換機(jī)和隊(duì)列的綁定，如果發(fā)送消息時(shí)的routingKey是inform.sms.routingKey，那么只有sms_routing這個(gè)隊(duì)列可以收到消息，email_routing隊(duì)列則不會(huì)收到消息，這正是直連的含義所在。

三、topic：通配符模式

topic類(lèi)型的交換機(jī)，常使用的通配符是：*和#

（1）星號(hào)*：匹配一個(gè)詞

（2）井號(hào)：匹配0個(gè)或多個(gè)詞

詞是通過(guò)點(diǎn)分割的，被點(diǎn)分割的就是一個(gè)詞

先看配置：

@Configuration
public class TopicsRabbitmqConfig {
    public static final String EMAIL_QUEUE = "email_topics";
    public static final String SMS_QUEUE = "sms_topics";
    public static final String EXCHANGE_NAME = "exchange_topics";
    public static final String EMAIL_ROUTINGKEY = "inform.#.email.#";
    // * 必須匹配一個(gè)詞，不能是0個(gè)
//    public static final String EMAIL_ROUTINGKEY = "inform.*.email";
    public static final String SMS_ROUTINGKEY = "inform.#.sms.#";
    @Bean
    public TopicExchange getTopicsExchange(){
        return new TopicExchange(EXCHANGE_NAME);
    }
    @Bean
    public Queue getTopicsEmailQueue(){
        return new Queue(EMAIL_QUEUE);
    }
    @Bean
    public Queue getTopicsSmsQueue(){
        return new Queue(SMS_QUEUE);
    }
    // 綁定交換機(jī)和隊(duì)列
    @Bean
    public Binding topicsEmailExchange(){
        return BindingBuilder.bind(getTopicsEmailQueue()).to(getTopicsExchange()).with(EMAIL_ROUTINGKEY);
    }
    @Bean
    public Binding topicsSmsExchange(){
        return BindingBuilder.bind(getTopicsSmsQueue()).to(getTopicsExchange()).with(SMS_ROUTINGKEY);
    }
}

和direct不同的是：bindingKey中可以含有星號(hào)和井號(hào)，可以看到：隊(duì)列email_topics的bindingKey（注意，這是bindingKey，不是routingKey）是inform.#.email.#，而隊(duì)列sms_topics則綁定了inform.#.sms.#。如果發(fā)送消息的routingKey是inform開(kāi)頭，后面含有email，則email_topics隊(duì)列會(huì)收到消息，而sms_topics隊(duì)列不會(huì)收到消息。

topic類(lèi)型交換機(jī)的強(qiáng)大之處在于，它可以替代前面兩種交換機(jī)：

（1）如果bindingKey只有一個(gè)井號(hào)，在是fanout

（2）如果bindingKey不含有星號(hào)和井號(hào)，則是direct

四、消費(fèi)者端接收消息

消費(fèi)者接收消息配置如下：

@Component
public class TopicsConsumer {
    // topics模式
    @RabbitListener(queues = TopicsRabbitmqConfig.EMAIL_QUEUE)
    public void receiveEmail(String msg){
        System.out.println("topics收到email消息： "+msg);
    }
    @RabbitListener(queues = TopicsRabbitmqConfig.SMS_QUEUE)
    public void receiveSms(String msg){
        System.out.println("topics收到sms消息： "+msg);
    }
    // routing模式
    @RabbitListener(queues = RoutingRabbitmqConfig.EMAIL_QUEUE)
    public void receiveRoutingEmail(String msg){
        System.out.println("routing收到email消息： "+msg);
    }
    @RabbitListener(queues = RoutingRabbitmqConfig.SMS_QUEUE)
    public void receiveRoutingSms(String msg){
        System.out.println("routing收到sms消息： "+msg);
    }
    // 發(fā)布訂閱模式
    @RabbitListener(queues = FanoutRabbitmqConfig.EMAIL_QUEUE)
    public void receiveFanoutEmail(String msg){
        System.out.println("fanout收到email消息： "+msg);
    }
    @RabbitListener(queues = FanoutRabbitmqConfig.SMS_QUEUE)
    public void receiveFanoutSms(String msg){
        System.out.println("fanout收到sms消息： "+msg);
    }
}

這里主要是使用了RabbitListener注解，指定了接收消息的隊(duì)列。