【SpringBoot系列5】SpringBoot整合RabbitMQ

前言：

因為項目需要用到RabbitMQ，前幾天就看了看RabbitMQ的知識，記錄下SpringBoot整合RabbitMQ的過程。

給出兩個網址：

RabbitMQ官方教程：http://www.rabbitmq.com/getstarted.html

SpringBoot整個RabbitMQ教程：https://docs.spring.io/spring-boot/docs/2.0.1.RELEASE/reference/htmlsingle/#boot-features-amqp

正文：

1：最簡單的HelloWorld

首先加上SpringBoot的依賴

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

其實我們要申明一個隊列，這里隊列存放生產者生產的消息，然后消費者來這里面取出來

@Configuration
public class RabbitBeanConf {
    
    public static final String QUEUE = "queue";
    
    @Bean
    public Queue queue() {
        return new Queue(QUEUE, true);
    }
}

生產者：

@Service
public class RabbitSender {

    @Autowired
    AmqpTemplate amqpTemplate;
    
    public void send(String msg) {
        amqpTemplate.convertAndSend(RabbitBeanConf.QUEUE, msg);
        System.out.println("生產者生產了一個消息： " + msg + "  " + new Date().getTime());
    }
}

消費者：

@Service
public class RabbitReceiver {

    @RabbitListener(queues = RabbitBeanConf.QUEUE)
    public void receive(String msg) {
        
        System.out.println("消費者收到了一個消息: " + msg + "  " + new Date().getTime());
    }
}

 

2：廣播模式(Fanout模式)

這個模式不同於上面的模式，這個模式只要一個生產者生產出一個消息，那么所有消費者全部可以接收到。

最簡單一個例子就是你微信群發消息的時候，你發了一條消息，但你的好友全部都能收到這條消息

RabbitMQ配置：

@Configuration
public class FanoutRabbitMQConfig {

    //默認持久化durable為true
    //根據方法名來進行綁定的 firstFanoutQueue()
    @Bean
    public Queue firstFanoutQueue() {
        return new Queue("firstFanoutQueue");
    }
    
    @Bean
    public Queue secondFanoutQueue() {
        return new Queue("secondFanoutQueue");
    }
    
    //默認持久化durbale為true
    //exchange交換機
    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange("fanoutExchange");
    }
    
    /**
     * 以下兩種方法都可以成功
     * 
     * @return
     * 2018年6月11日
     */
    @Bean
    public Binding bindingFirst() {
        return BindingBuilder.bind(firstFanoutQueue()).to(fanoutExchange());
    }
    
    @Bean
    public Binding bindingSecond(FanoutExchange fanoutExchange, Queue secondFanoutQueue) {
        return BindingBuilder.bind(secondFanoutQueue).to(fanoutExchange);
    }
}

這里我解釋一下，官方引入了一個叫做 “exchange”交換機的東西。

我是這樣理解的：這個交換機主要是為了解耦生產了和消費者直接的強聯系。生產者不直接往隊列里面送東西了，而且往exchange里面送東西，然后exchange綁定想要送消息進去的隊列，然后消費者監聽即可。主要解耦。

生產者：

@Service
public class FanoutSender {
    
    @Autowired
    AmqpTemplate amqpTemplate;
    
    public void send(String msg) {
        
        amqpTemplate.convertAndSend("fanoutExchange", "", msg);
        System.out.println("生產者生產了一個消息： " + msg + "  " + new Date().getTime());
    }
}

消費者：

@Service
public class FanoutRecevi {

    @RabbitListener(queues = {"firstFanoutQueue", "secondFanoutQueue"})
    public void receive(String msg) {
        
        System.out.println("消費者收到了一個消息: " + msg + "  " + new Date().getTime());
    }
}

 

3：Topic模式

感覺和直接模式(direct模式)有點相像。只不過這里用了類似正則的東西，一個exchange匹配符合規則的隊列。

topic 和 direct 類似, 只是匹配上支持了"模式", 在"點分"的 routing_key 形式中, 可以使用兩個通配符:

• *表示一個詞.

• #表示零個或多個詞.

 

4：Header模式

headers 也是根據規則匹配, 相較於 direct 和 topic 固定地使用 routing_key , headers 則是一個自定義匹配規則的類型.
在隊列與交換器綁定時, 會設定一組鍵值對規則, 消息中也包括一組鍵值對( headers 屬性), 當這些鍵值對有一對, 或全部匹配時, 消息被投送到對應隊列.