Spring Boot 整合 RabbitMQ 之 Direct模式 (一)

妖狐艹你老母 2024-02-18 16:17 47阅读 0赞

摘要：Spring Boot 整合RabbitMQ 实现消息发生和接收。

一：RabbitMQ的介绍：

RabbitMQ是消息中间件的一种,消息中间件即分布式系统中完成消息的发送和接收的基础软件.这些软件有很多,包括ActiveMQ(apache公司的),RocketMQ(阿里巴巴公司的,现已经转让给apache).  
消息中间件的工作过程可以用生产者消费者模型来表示.即,生产者不断的向消息队列发送信息,而消费者从消息队列中消费信息.具体过程如下:

![Center][]

从上图可看出,对于消息队列来说,生产者,消息队列,消费者是最重要的三个概念,生产者发消息到消息队列中去,消费者监听指定的消息队列,并且当消息队列收到消息之后,接收消息队列传来的消息,并且给予相应的处理.消息队列常用于分布式系统之间互相信息的传递.  
对于RabbitMQ来说,除了这三个基本模块以外,还添加了一个模块,即交换机(Exchange).它使得生产者和消息队列之间产生了隔离,生产者将消息发送给交换机,而交换机则根据调度策略把相应的消息转发给对应的消息队列.那么RabitMQ的工作流程如下所示:

![Center 1][]

紧接着说一下交换机.交换机的主要作用是接收相应的消息并且绑定到指定的队列.交换机有四种类型,分别为Direct,topic,headers,Fanout.  
  
  
　　Direct是RabbitMQ默认的交换机模式,也是最简单的模式.即创建消息队列的时候,指定一个BindingKey.当发送者发送消息的时候,指定对应的Key.当Key和消息队列的BindingKey一致的时候,消息将会被发送到该消息队列中.  
  
  
　　topic转发信息主要是依据通配符,队列和交换机的绑定主要是依据一种模式(通配符+字符串),而当发送消息的时候,只有指定的Key和该模式相匹配的时候,消息才会被发送到该消息队列中.  
  
  
　　headers也是根据一个规则进行匹配,在消息队列和交换机绑定的时候会指定一组键值对规则,而发送消息的时候也会指定一组键值对规则,当两组键值对规则相匹配的时候,消息会被发送到匹配的消息队列中.  
  
  
　　Fanout是路由广播的形式,将会把消息发给绑定它的全部队列,即便设置了key,也会被忽略.

二：Spring Boot 整合 RabbitMQ 支持三种模式，分别是：Direct模式(相当于一对一模式,一个消息被发送者发送后,会被转发到一个绑定的消息队列中,然后被一个接收者接收!)，

Topic转发模式，Fanout Exchange形式(广播模式)

三：Spring Boot 整合 RabbitMQ 之Direct模式：

SpringBoot整合RabbitMQ非常简单!感觉SpringBoot真的极大简化了开发的搭建环境的时间..这样我们程序员就可以把更多的时间用在业务上了,下面开始搭建环境:

首先创建1个maven工程,这是为了模拟分布式应用系统中,新建两个类，一个发送者(Sender),一个接收者(Receiver) 如下图：

![Center 2][]

紧接着,配置pom.xml文件,注意其中用到了springboot对于AMQP(高级消息队列协议,即面向消息的中间件的设计)

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
             xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    
        <groupId>com.micai</groupId>
        <artifactId>micai-springboot-rabbitmq-7</artifactId>
        <version>1.0-SNAPSHOT</version>
        <modelVersion>4.0.0</modelVersion>
    
        <packaging>jar</packaging>
    
        <name>micai-springboot-rabbitmq-7</name>
        <url>http://maven.apache.org</url>
    
        
        <parent>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-parent</artifactId>
            <version>1.5.1.RELEASE</version>
        </parent>
    
        <dependencies>
            
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
            </dependency>
            
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-amqp</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-test</artifactId>
                <scope>test</scope>
            </dependency>
        </dependencies>
    
    
    </project>

紧接着,我们编写发送者相关的代码.首先毫无疑问,要书写启动类:

package com.micai.springboot;
    
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    
    /**
     * Spring Boot应用启动类
     *
     * @author zhaoxinguo on 2017/8/21.
     */
    //@EnableScheduling // 定时任务注解
    @SpringBootApplication
    public class Application {
    
        public static void main(String [] args){
    
            SpringApplication.run(Application.class,args);
    
        }
    
    
    }

接着在application.properties中,去编辑和RabbitMQ相关的配置信息,配置信息的代表什么内容根据键就能很直观的看出了.这里端口是5672,不是15672...15672是管理端的端口!

spring.application.name=micai-springboot-rabbitmq-7
    
    # MQ配置
    spring.rabbitmq.host=localhost
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=guest
    spring.rabbitmq.password=guest

随后,配置Queue(消息队列).那注意由于采用的是Direct模式,需要在配置Queue的时候,指定一个键,使其和交换机绑定.

package com.micai.springboot.mq.config;
    
    import com.micai.springboot.base.BaseConfig;
    import org.springframework.amqp.core.*;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    /**
     * 描述：Topic转发模式
     * <p>
     * Author: 赵新国
     * Date: 2017/11/3 18:51
     */
    @Configuration
    public class SenderConf extends BaseConfig {
    
        // ---------------------------------------------------- Direct 形式 -------------------------------------------- //
    
        @Bean
        public Queue queue() {
            return new Queue(QUEUE_KEY);
        }
    
        // ---------------------------------------------------- Topic 形式 -------------------------------------------- //
    
        /*@Bean(name = "message")
        public Queue queueMessage() {
            return new Queue("topic.message");
        }
    
        @Bean(name = "messages")
        public Queue queueMessages() {
            return new Queue("topic.messages");
        }
    
        @Bean
        public TopicExchange exchange() {
            return new TopicExchange("exchange");
        }
    
        @Bean
        Binding bindingExchangeMessage(@Qualifier("message") Queue queueMessage, TopicExchange exchange) {
            return BindingBuilder.bind(queueMessage).to(exchange).with("topic.message");
        }
    
        @Bean
        Binding bindingExchangeMessages(@Qualifier("messages") Queue queueMessages, TopicExchange exchange) {
            return BindingBuilder.bind(queueMessages).to(exchange).with("topic.#");//*表示一个词,#表示零个或多个词
        }*/
    
        // -------------------------------------------- Fanout Exchange形式 ------------------------------------------- //
    
        /*@Bean(name="Amessage")
        public Queue AMessage() {
            return new Queue("fanout.A");
        }
    
    
        @Bean(name="Bmessage")
        public Queue BMessage() {
            return new Queue("fanout.B");
        }
    
        @Bean(name="Cmessage")
        public Queue CMessage() {
            return new Queue("fanout.C");
        }
    
        @Bean
        FanoutExchange fanoutExchange() {
            return new FanoutExchange("fanoutExchange");//配置广播路由器
        }
    
        @Bean
        Binding bindingExchangeA(@Qualifier("Amessage") Queue AMessage,FanoutExchange fanoutExchange) {
            return BindingBuilder.bind(AMessage).to(fanoutExchange);
        }
    
        @Bean
        Binding bindingExchangeB(@Qualifier("Bmessage") Queue BMessage, FanoutExchange fanoutExchange) {
            return BindingBuilder.bind(BMessage).to(fanoutExchange);
        }
    
        @Bean
        Binding bindingExchangeC(@Qualifier("Cmessage") Queue CMessage, FanoutExchange fanoutExchange) {
            return BindingBuilder.bind(CMessage).to(fanoutExchange);
        }*/
    
    }

接着就可以发送消息啦!在SpringBoot中,我们使用AmqpTemplate去发送消息!代码如下:

package com.micai.springboot.mq;
    
    import com.micai.springboot.base.BaseConfig;
    import org.springframework.amqp.core.AmqpTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    /**
     * 描述：消息生产者
     * <p>
     * Author: 赵新国
     * Date: 2017/11/3 15:37
     */
    @Component
    public class Sender extends BaseConfig {
    
        @Autowired
        private AmqpTemplate rabbitTemplate;
    
        public void send() {
            // ---------------------------------------------------- Direct 形式 -------------------------------------------- //
            // 在该生产者，我们会产生一个字符串，并发送到名为hello的队列中
            String context = "Hello " + "Rabbit MQ！";
            System.out.println("发送MQ消息 : " + context);
            this.rabbitTemplate.convertAndSend(QUEUE_KEY, context);
    
            // 发送对象，但是该对象必须实现Serializable接口
            /*User user = new User(); //实现Serializable接口
            user.setId(1L);
            user.setName("张三");
            this.rabbitTemplate.convertAndSend(QUEUE_KEY, user);*/
    
            // ---------------------------------------------------- Topic 形式 -------------------------------------------- //
            /*this.rabbitTemplate.convertAndSend("exchange", "topic.message", "hello, rabbit!");*/
    
            // -------------------------------------------- Fanout Exchange形式 ------------------------------------------- //
            /*this.rabbitTemplate.convertAndSend("fanoutExchange", "", "xixi,hlhdidi");// 参数2将被忽略*/
        }
    
    }

编写测试类!这样我们的发送端代码就编写完了~

package com.micai;
    
    import com.micai.springboot.Application;
    import com.micai.springboot.mq.Sender;
    import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
    
    /**
     * 描述：
     * <p>
     * Author: 赵新国
     * Date: 2017/11/3 17:21
     */
    @SpringBootTest(classes=Application.class)
    @RunWith(SpringJUnit4ClassRunner.class)
    public class RabbitMQTest {
    
        @Autowired
        private Sender sender;
    
        @Test
        public void send (){
           sender.send();
        }
    
    }

接着我们编写接收端.接收端的pom文件,application.properties(修改spring.application.name),Queue配置类,App启动类都是一致的!这里省略不计.主要在于我们需要配置监听器去监听绑定到的消息队列,当消息队列有消息的时候,予以接收,代码如下:

package com.micai.springboot.mq;
    
    import com.micai.springboot.base.BaseConfig;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    /**
     * 描述：消息消费者
     * @RabbitListener注解定义该类对hello队列的监听，
     * 并用@RabbitHandler注解来指定对消息的处理方法。
     * 所以，该消费者实现了对hello队列的消费，消费操作为输出消息的字符串内容
     * Author: 赵新国
     * Date: 2017/11/3 15:42
     */
    @Component
    public class Receiver extends BaseConfig {
    
        // ---------------------------------------------------- Direct 形式 -------------------------------------------- //
        //监听器监听指定的Queue
        @RabbitListener(queues = QUEUE_KEY)
        public void process(String str) {
            System.out.println("接收MQ消息 : " + str);
        }
    
        // //监听器监听指定的Queue
        /*@RabbitListener(queues = QUEUE_KEY)
        public void process(User user) { //用User作为参数
            System.out.println("接收MQ消息 : " + user);
        }
    
        // ---------------------------------------------------- Topic 形式 -------------------------------------------- //
        @RabbitListener(queues="topic.message")    //监听器监听指定的Queue
        public void process1(String str) {
            System.out.println("message:"+str);
        }
    
        @RabbitListener(queues="topic.messages")    //监听器监听指定的Queue
        public void process2(String str) {
            System.out.println("messages:"+str);
        }*/
    
    
        // -------------------------------------------- Fanout Exchange形式 ------------------------------------------- //
        /*@RabbitListener(queues="fanout.A")
        public void processA(String str1) {
            System.out.println("ReceiveA:"+str1);
        }
    
        @RabbitListener(queues="fanout.B")
        public void processB(String str) {
            System.out.println("ReceiveB:"+str);
        }
    
        @RabbitListener(queues="fanout.C")
        public void processC(String str) {
            System.out.println("ReceiveC:"+str);
        }*/
    
    }

接下来就可以测试啦,首先启动接收端的应用,紧接着运行发送端的单元测试,接收端应用打印出来接收到的消息,测试即成功!  
  
需要注意的地方,Direct模式相当于一对一模式,一个消息被发送者发送后,会被转发到一个绑定的消息队列中,然后被一个接收者接收!

运行结果如下图：

![Center 3][]

![Center 4][]

实际上RabbitMQ还可以支持发送对象:当然由于涉及到序列化和反序列化,该对象要实现Serilizable接口.Sender做出如下改写:

package com.micai.springboot.mq;
    
    import com.micai.springboot.base.BaseConfig;
    import com.micai.springboot.entity.User;
    import org.springframework.amqp.core.AmqpTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    /**
     * 描述：消息生产者
     * <p>
     * Author: 赵新国
     * Date: 2017/11/3 15:37
     */
    @Component
    public class Sender extends BaseConfig {
    
        @Autowired
        private AmqpTemplate rabbitTemplate;
    
        public void send() {
            // ---------------------------------------------------- Direct 形式 -------------------------------------------- //
            // 在该生产者，我们会产生一个字符串，并发送到名为hello的队列中
            /*String context = "Hello " + "Rabbit MQ！";
            System.out.println("发送MQ消息 : " + context);
            this.rabbitTemplate.convertAndSend(QUEUE_KEY, context);*/
    
            // 发送对象，但是该对象必须实现Serializable接口
            User user = new User(); //实现Serializable接口
            user.setId(1L);
            user.setName("张三");
            this.rabbitTemplate.convertAndSend(QUEUE_KEY, user);
    
            // ---------------------------------------------------- Topic 形式 -------------------------------------------- //
            /*this.rabbitTemplate.convertAndSend("exchange", "topic.message", "hello, rabbit!");*/
    
            // -------------------------------------------- Fanout Exchange形式 ------------------------------------------- //
            /*this.rabbitTemplate.convertAndSend("fanoutExchange", "", "xixi,hlhdidi");// 参数2将被忽略*/
        }
    
    }

Receiver做出如下改写:

package com.micai.springboot.mq;
    
    import com.micai.springboot.base.BaseConfig;
    import com.micai.springboot.entity.User;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    /**
     * 描述：消息消费者
     * @RabbitListener注解定义该类对hello队列的监听，
     * 并用@RabbitHandler注解来指定对消息的处理方法。
     * 所以，该消费者实现了对hello队列的消费，消费操作为输出消息的字符串内容
     * Author: 赵新国
     * Date: 2017/11/3 15:42
     */
    @Component
    public class Receiver extends BaseConfig {
    
        // ---------------------------------------------------- Direct 形式 -------------------------------------------- //
        //监听器监听指定的Queue
        /*@RabbitListener(queues = QUEUE_KEY)
        public void process(String str) {
            System.out.println("接收MQ消息 : " + str);
        }*/
    
        // //监听器监听指定的Queue
        @RabbitListener(queues = QUEUE_KEY)
        public void process(User user) { //用User作为参数
            System.out.println("接收MQ消息 : " + user);
        }
    
        // ---------------------------------------------------- Topic 形式 -------------------------------------------- //
        /*@RabbitListener(queues="topic.message")    //监听器监听指定的Queue
        public void process1(String str) {
            System.out.println("message:"+str);
        }
    
        @RabbitListener(queues="topic.messages")    //监听器监听指定的Queue
        public void process2(String str) {
            System.out.println("messages:"+str);
        }*/
    
    
        // -------------------------------------------- Fanout Exchange形式 ------------------------------------------- //
        /*@RabbitListener(queues="fanout.A")
        public void processA(String str1) {
            System.out.println("ReceiveA:"+str1);
        }
    
        @RabbitListener(queues="fanout.B")
        public void processB(String str) {
            System.out.println("ReceiveB:"+str);
        }
    
        @RabbitListener(queues="fanout.C")
        public void processC(String str) {
            System.out.println("ReceiveC:"+str);
        }*/
    
    }

改写后的运行结果如下：

![Center 5][]

源代码下载地址： [https://gitee.com/micai/micai-springboot/tree/master/micai-springboot-rabbitmq-7][https_gitee.com_micai_micai-springboot_tree_master_micai-springboot-rabbitmq-7]

[Center]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/0ce57244e1734ed6a03e3996db7f9dc4.png
[Center 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/15ad5fe1049f4782b2a1a849c2df3fad.png
[Center 2]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/b8f0f74e2917498d8375de7f2744c7c1.png
[Center 3]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/21aff854506c4e3798238d2f2ea67871.png
[Center 4]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/0ab4f8bad0f1455b821194b99c57c88f.png
[Center 5]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/ff31fecb198346b094f580cb7cd911df.png
[https_gitee.com_micai_micai-springboot_tree_master_micai-springboot-rabbitmq-7]: https://gitee.com/micai/micai-springboot/tree/master/micai-springboot-rabbitmq-7