rabbitMQ結(jié)合spring-boot使用(1)

從這一節(jié)開(kāi)始我們進(jìn)入rabbitMQ的實(shí)戰(zhàn)環(huán)節(jié)，項(xiàng)目環(huán)境是spring-boot 加maven。首先讓我們創(chuàng)建一個(gè)spring-boot項(xiàng)目，然后引入web依賴(lài)和 rabbitMQ的依賴(lài)

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

application.properties

spring.rabbitmq.host=localhost
spring.rabbitmq.username=guest
spring.activemq.password=guest

環(huán)境搭建好之后我們就可以開(kāi)始進(jìn)行實(shí)戰(zhàn)操作了。

簡(jiǎn)單消息隊(duì)列

springboot會(huì)默認(rèn)為你創(chuàng)建一個(gè)direct exchange類(lèi)型交換機(jī)，其名稱(chēng)為""空字符串，其路由鍵和綁定鍵都是隊(duì)列名稱(chēng)，未指定交換機(jī)的隊(duì)列都會(huì)綁定到這個(gè)交換機(jī)上去。我們就以這個(gè)最簡(jiǎn)單的消息隊(duì)列開(kāi)始來(lái)學(xué)習(xí)如何在項(xiàng)目中使用rabbitMQ。

我們先注冊(cè)兩個(gè)隊(duì)列，一個(gè)用于傳遞String類(lèi)型消息，一個(gè)傳遞Object類(lèi)型的數(shù)據(jù)。項(xiàng)目啟動(dòng)后springboot會(huì)為你在 rabbitMQ 中創(chuàng)建兩個(gè)隊(duì)列，啟動(dòng)項(xiàng)目后打開(kāi) rabbitMQ 的 web 管理界面（以下簡(jiǎn)稱(chēng)管理界面）會(huì)在 Queues 中看到這兩個(gè)隊(duì)列的相關(guān)信息。

@Component
public class QueueConfig {
    @Bean
    public Queue getSimpleQueue() {
        return new Queue("simple-queue");
    }

    @Bean
    public Queue getObjSimpleQueue() {
        return new Queue("obj-simple-queue");
    }
}

創(chuàng)建兩個(gè)定時(shí)任務(wù)，向 rabbitMQ 投遞消息，注意這里需要在啟動(dòng)類(lèi)上加 @EnableScheduling 注解以啟動(dòng)定時(shí)任務(wù)，而 Message 是我創(chuàng)建的實(shí)體類(lèi)：

@Component
public class ScheduleHandler {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Scheduled(fixedRate = 6000)
    private void simpleQueueSchedule() {
        System.out.println("<<<<<<<<<<");

        rabbitTemplate.convertAndSend("simple-queue","ni----hao");
    }

    @Scheduled(fixedRate = 6000)
    private void objSimpleQueueSchedule() {
        System.out.println("<<<<<<<<<<");
        Message message = new Message();
        message.setTitle("hello");
        message.setContent("how are you ");
        rabbitTemplate.convertAndSend("obj-simple-queue",message);
    }

}

消費(fèi)者消費(fèi)消息：

@Component
public class QueueMessageHandler {

    @RabbitListener(queues = { "simple-queue"})
    public void getSimpleQueueMessage(String msg){
        System.out.println(msg);
    }

    @RabbitListener(queues = { "obj-simple-queue"})
    public void getObjSimpleQueueMessage(Message msg){
        System.out.println(msg);
    }

}

rabbitTemplate.convertAndSend()方法是將數(shù)據(jù)序列化并寫(xiě)入隊(duì)列中，而其使用的序列化協(xié)議自然是java序列化協(xié)議（使用 ObjectInputStream 和 ObjectOutputStream 讀寫(xiě)），因此你如果調(diào)用這個(gè)方法則其實(shí)體類(lèi)需要實(shí)現(xiàn)Serializable接口，而如果跨虛擬機(jī)還需要注意 serialVersionUID。如果跨平臺(tái)了，那么最好使用其他序列化的方式,序列化反序列化配置在后文關(guān)于監(jiān)聽(tīng)器容器的章節(jié)介紹。

推模式和拉模式

對(duì)消費(fèi)端而言使用@RabbitListener監(jiān)聽(tīng)器獲取MQ消息的方式稱(chēng)為推模式，我們還可以使用拉模式，當(dāng)我們需要一條消息的時(shí)候才從隊(duì)列中拉一條消息出來(lái)，使用的方法為 rabbitTemplate.receiveAndConvert()，如：

  Message o = ((Message) rabbitTemplate.receiveAndConvert("obj-simple-queue"));

direct exchange 直連交換機(jī)

直連交換機(jī)，需要注冊(cè)一個(gè) DirectExchange , Queue , Binding 。Bingding 負(fù)責(zé)將 DirectExchange 和 Queue 綁定并指定 routingKey 生產(chǎn)者生產(chǎn)消息的時(shí)候也需要指定 routingKey。下面看示例：

//  生產(chǎn)端配置
    @Bean("directQueueFirst")
    public Queue directQueueFirst() {
        return new Queue("first-direct-queue");
    }

    @Bean("directQueueSecond")
    public Queue directQueueSecond() {
        return QueueBuilder.durable("second-direct-queue").build();
    }
    @Bean("directExchange")
    public DirectExchange directExchange() {
        return new DirectExchange("direct-exchange");
    }
    
    @Bean
    public Binding bingQueueFirstToDirect(@Qualifier("directQueueFirst") Queue queue,       @Qualifier("directExchange") DirectExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("first-key");
    }

    @Bean
    public Binding bingQueueSecondToDirect(@Qualifier("directQueueSecond") Queue queue, @Qualifier("directExchange") DirectExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("second-key");
    }
    
//  生產(chǎn)者發(fā)送消息
@Component
public class ScheduleHandler {

    @Scheduled(fixedRate = 6000)
    private void directMessageScheduleFirst() {
        Message message = new Message();
        message.setTitle("hello");
        message.setContent("how are you for direct first");
        rabbitTemplate.convertAndSend("direct-exchange","first-key",message);
    }

    @Scheduled(fixedRate = 6000)
    private void directMessageScheduleSecond() {
        Message message = new Message();
        message.setTitle("hello");
        message.setContent("how are you for direct second");
        rabbitTemplate.convertAndSend("topic-exchange","second-key",message);
    }
}
@Component
public class QueueMessageHandler {
//  消費(fèi)端
    @RabbitListener(queues = { "first-direct-queue"})
    public void firstDirectMessageQueue(Message msg){
        System.out.println(msg);
    }

    @RabbitListener(queues = { "second-direct-queue"})
    public void secondDirectMessageQueue(Message msg){
        System.out.println(msg);
    }
}

值得注意的是，springboot為了使我們的代碼可讀性更好，還非常貼心的提供 Exchange,Binding,Queue的Builder（建造者），因此你可以使用它們對(duì)應(yīng)建造者，也可以使用直接 new 的方式進(jìn)行創(chuàng)建。另外創(chuàng)建的這些 exchange queue 都能在管理界面上看到，如圖 2 ，圖 3 ：

? 圖 2：隊(duì)列信息


? 圖 3：交換機(jī)信息

fanout exchange 扇型交換機(jī)

使用上和 direct exchange 大同小異，只不過(guò)不需要指定路由鍵，而且所有和它綁定的隊(duì)列都會(huì)收到消息，直接上代碼：

// 生產(chǎn)者配置
    @Bean("fanoutQueueFirst")
    public Queue fanoutQueueFirst() {
        return new Queue("first-fanout-queue");
    }

    @Bean("fanoutQueueSecond")
    public Queue fanoutQueueSecond() {
        return new Queue("second-fanout-queue");
    }

    @Bean("fanoutExchange")
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange("fanout-exchange");
    }

    @Bean
    public Binding bingQueueFirstToExchange(@Qualifier("fanoutQueueFirst") Queue queue, @Qualifier("fanoutExchange") FanoutExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange);
    }

    @Bean
    public Binding bingQueueSecondToExchange(@Qualifier("fanoutQueueSecond") Queue queue, @Qualifier("fanoutExchange") FanoutExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange);
    }
@Component
public class ScheduleHandler {
//  生產(chǎn)者發(fā)消息，注意這里雖然填了routingKey 但是是無(wú)效的
    @Scheduled(fixedRate = 6000)
    private void directMessageScheduleFirst() {
        Message message = new Message();
        message.setTitle("hello");
        message.setContent("how are you for direct first");
        rabbitTemplate.convertAndSend("direct-exchange","first-key",message);
    }
}
//  消費(fèi)者，兩個(gè)隊(duì)列都能收到同一份消息
@Component
public class QueueMessageHandler {
    @RabbitListener(queues = { "first-fanout-queue"})
    public void firstFanoutQueue(Message msg){
        System.out.println(msg);
    }

    @RabbitListener(queues = { "second-fanout-queue"})
    public void secondFanoutQueue(Message msg){
        System.out.println(msg);
    }
}

主題交換機(jī) Topic

前文介紹了主題交換機(jī)的路由方式，注意我代碼中的路由鍵設(shè)置，這里我設(shè)置兩個(gè)bingding-key 分別是 com.muggle.first 和 com.# 我用 routing-key 為 com.muggle.test 發(fā)消息這兩個(gè)隊(duì)列都能接收到

    @Bean("topicQueueFirst")
    public Queue topicQueueFirst() {
        return new Queue("first-topic-queue");
    }

    @Bean("topicQueueSecond")
    public Queue topicQueueSecond() {
        return new Queue("second-topic-queue");
    }

    @Bean
    public Binding bindTopicFirst(@Qualifier("topicQueueFirst") Queue queue, @Qualifier("topicExchange") TopicExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("com.muggle.first");
    }

    @Bean
    public Binding bindTopicSecond(@Qualifier("topicQueueFirst") Queue queue, @Qualifier("topicExchange") TopicExchange exchange) {
        return BindingBuilder.bind(topicQueueFirst()).to(topicExchange()).with("com.#");
    }
    
@Component
public class ScheduleHandler {
    @Scheduled(fixedRate = 6000)
    private void topicMessage() {
        Message message = new Message();
        message.setTitle("hello");
        message.setContent("how are you for topic test");
        rabbitTemplate. convertAndSend("topic-exchange","com.muggle.test",message);
    }
    
}

@Component
public class QueueMessageHandler {
    @RabbitListener(queues = { "first-topic-queue"})
    public void firstTopicMessageQueue(Message msg){
        System.out.println(msg);
    }

    @RabbitListener(queues = { "second-topic-queue"})
    public void secondTopicMessageQueue(Message msg){
        System.out.println(msg);
    }
}

現(xiàn)在我們學(xué)習(xí)了 rabbitMQ 的各類(lèi)交換機(jī)的用法，這些只是 rabbitMQ 的基礎(chǔ)特性，下文我們將介紹一些 rabbitMQ 的更復(fù)雜的使用方法。

作者：muggle 點(diǎn)我關(guān)注作者

出處：https://muggle-book.gitee.io/

版權(quán)：本文版權(quán)歸作者所有

轉(zhuǎn)載：歡迎轉(zhuǎn)載，但未經(jīng)作者同意，必須保留此段聲明；必須在文章中給出原文連接；否則必究法律責(zé)任

點(diǎn)擊關(guān)注我的博客

?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請(qǐng)聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時(shí)請(qǐng)結(jié)合常識(shí)與多方信息審慎甄別。
平臺(tái)聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點(diǎn)，簡(jiǎn)書(shū)系信息發(fā)布平臺(tái)，僅提供信息存儲(chǔ)服務(wù)。

友情鏈接更多精彩內(nèi)容