消息中间件（六）之-springBoot集成RabbitMQ

缺乏、安全感 2022-05-29 01:22 191阅读 0赞

### 概述 ###

RabbitMQ 即一个消息队列，主要是用来实现应用程序的异步和解耦，同时也能起到消息缓冲，消息分发的作用。消息中间件最主要的作用是解耦，中间件最标准的用法是生产者生产消息传送到队列，消费者从队列中拿取消息并处理，生产者不用关心是谁来消费，消费者不用关心谁在生产消息，从而达到解耦的目的。在分布式的系统中，消息队列也会被用在很多其它的方面，比如：分布式事务的支持，RPC的调用等等。RabbitMQ是流行的开源消息队列系统，用erlang语言开发。RabbitMQ是AMQP（高级消息队列协议）的标准实现。如果不熟悉AMQP，直接看RabbitMQ的文档会比较困难。不过RabbitMQ也就几个关键概念，关键词理解更有助于理解其工作方式和原理，RabbitMQ的结构图如下：

![JCccc-RabbitMq][]

### 关键词 ###

Broker：简单来说就是消息队列服务器实体；  
　　Exchange：消息交换机，它指定消息按什么规则，路由到哪个队列；  
　　Queue：消息队列载体，每个消息都会被投入到一个或多个队列；  
　　Binding：绑定，它的作用就是把exchange和queue按照路由规则绑定起来；  
　　Routing Key：路由关键字，exchange根据这个关键字进行消息投递；  
　　vhost：虚拟主机，一个broker里可以开设多个vhost，用作不同用户的权限分离；  
　　producer：消息生产者，就是投递消息的程序；  
　　consumer：消息消费者，就是接受消息的程序；  
　　channel：消息通道，在客户端的每个连接里，可建立多个channel，每个channel代表一个会话任务；

消息队列的使用过程步骤归纳如下：

1.  client连接到消息队列server，打开一个channel;
2.  client声明一个exchange，并设置相关属性;
3.  client声明一个queue，并设置相关属性;
4.  client使用routing key，在exchange和queue之间建立好绑定关系;
5.  client投递消息到exchange;

exchange接收到消息后，就根据消息的key和配置的binding，进行消息路由，将消息投递到一个或多个队列里。exchange也有几个类型：

①Direct：仅依据 key进行消息投递的叫做Direct交换机，例如，绑定时设置了routing key为”test”，那么客户端提交的消息，只有设置了key为”test”的才会投递到队列；

②Topic：依据key进行模式匹配后进行投递的叫做Topic交换机，符号”\#”匹配一个或多个词，符号”\*”匹配正好一个词。例如”abc.\#”匹配”abc.def.ghi”，”abc.\*”只匹配”abc.def”；

③Fanout：存在一种模式下不需要key叫做Fanout交换机，它采取广播模式，一个消息进来时，投递到与该交换机绑定的所有队列，就类似微信公众号的推送，你只要订阅了某个公众号，所有关注这个公众号的微信都会收到推送消息；

④Headers：是根据一个规则进行匹配,在消息队列和交换机绑定的时候会指定一组键值对规则,而发送消息的时候也会指定一组键值对规则,当两组键值对规则相匹配的时候,消息会被发送到匹配的消息队列中。

### 消息持久化 ###

RabbitMQ支持消息的持久化，也就是数据写在磁盘上，为了数据安全考虑，我想大多数用户都会选择持久化。消息队列持久化包括3个部分：  
　　（1）exchange持久化，在声明时指定durable => 1  
　　（2）queue持久化，在声明时指定durable => 1  
　　（3）消息持久化，在投递时指定delivery\_mode => 2（1是非持久化）

如果exchange和queue都是持久化的，那么它们之间的binding也是持久化的。如果exchange和queue两者之间有一个持久化，一个非持久化，就不允许建立绑定。

### RabbitMQ使用 ###

**基本设置**

// 1. 引入pom
      <dependency>
         <groupId>org.springframework.boot</groupId>
         <artifactId>spring-boot-starter-amqp</artifactId>
      </dependency>
    
      // 2.配置文件
      #配置rabbitmq的安装地址、端口以及账户信息
      #注意port 15672是管理端的端口
      spring.application.name=spirng-boot-rabbitmq-sender
      spring.rabbitmq.host=127.0.0.1
      spring.rabbitmq.port=5672
      spring.rabbitmq.username=guest
      spring.rabbitmq.password=guest

**Direct模式简单测试**

producer 和 consumer的**queue ****name**必须一致，否则会收不到消息。**Direct**模式相当于一对一模式，producer 发送一个消息被后，会被转发到绑定的消息队列中，然后被一个 consumer接收。

// 队列配置
    
    @Configuration
    public class RabbitConfig {
    
         @Bean
         public Queue helloQueue() {
                 return new Queue("hello");
         }
    }
    
    // producer
    
    public void send(User user) {
        System.out.println("Sender object: " + user.toString());
        this.rabbitTemplate.convertAndSend("object", user);
    }
    
    // consumer
    
    @RabbitHandler
    public void process(User user) {
        System.out.println("consumer object : " + user);
        System.out.println("username:"+user.getUsername());
        System.out.println("password:"+user.getPassword());
    }
    
    // TEST
    
    @org.junit.Test
    public void sendObj() {
        User user = new User();
        user.setId(1);
        user.setUsername("admin");
        user.setPassword("1234");
        objectSender.send(user);
    }
    
    // TEST RESULT
    
     Sender object: com.example.bean.User@fa5f81c
     2019-01-06 11:13:40.175  INFO 26456 --- [       Thread-2] 
     o.s.a.r.l.SimpleMessageListenerContainer : Waiting for workers to finish.
     consumer object : com.example.bean.User@7b0f425
     username:admin
     password:1234

**Topic模式**

Topic 模式是 RabbitMQ中最灵活的一种方式，可以根据routing\_key自由的绑定不同的队列，这就需要单独配置队列对应关系。

1.Topic模式配置类

@Configuration
    public class TopicRabbitConfig {
    
          // 一般在配置文件
          final static String one= "TOPIC_MESSAGE_ONE";
          final static String two= "TOPIC_MESSAGE_TWO";
    
          // 创建队列
          @Bean
          public Queue queueMessage() {
                return new Queue(TopicRabbitConfig.one);
          }
    
          // 创建队列
          @Bean
          public Queue queueMessages() {
                 return new Queue(TopicRabbitConfig.two);
          }
    
          // 将对列绑定到Topic交换器
          @Bean
          TopicExchange exchange() {
                return new TopicExchange("topicExchange");
          }
    
          // 将对列绑定到Topic交换器
          @Bean
          Binding bindingExchangeMessage(Queue queueMessage, TopicExchange exchange) {
                return BindingBuilder.bind(queueMessage).to(exchange).with("topic.one");
          }
    
          // 将对列绑定到Topic交换器 采用#的方式
          @Bean
          Binding bindingExchangeMessages(Queue queueMessages, TopicExchange exchange) {
                return BindingBuilder.bind(queueMessages).to(exchange).with("topic.#");
          }
    }

2. 测试类

// producer one
    public void send1() {
        String context = "hello test one";
        System.out.println("Sender : " + context);
        this.rabbitTemplate.convertAndSend("topicExchange", "topic.one", context);
    }
    
    // producer two
    public void send2() {
        String context = "hello test two";
        System.out.println("Sender : " + context);
        this.rabbitTemplate.convertAndSend("topicExchange", "topic.two", context);
    }
    
    
    // consumer
    @Component
    @RabbitListener(queues = "topic.one")
    public class TopicReceiver {
           @RabbitHandler
           public void process(String message) {
               System.out.println("Topic consumer one: " + message);
           }
    }
    
    /*
     * producer 发送send1会匹配到topic.#和topic.one 两个consumer都可以收到消息， 
     * producer 发送send2只有topic.#可以匹配所有只有consumer two监听到消息
    */
    
    // test
    @RunWith(SpringRunner.class)
    @SpringBootTest
    public class Test {
    
         @Autowired
         private TopicSender topicSender;
    
        @org.junit.Test
        public void send1() {
           topicSender.send1();
        }
       @org.junit.Test
        public void send2() {
           topicSender.send2();
        }
    
    }

**Headers 模式**

1.headers模式队列配置类

@Configuration
    public class HeadersConfig {
    
         @Bean
         public Queue creditBankQueue() {
                return new Queue("credit.bank");
         }
         @Bean
         public Queue creditFinanceQueue() {
                return new Queue("credit.finance");
         }
         @Bean
         public HeadersExchange creditBankExchange() {
                return new HeadersExchange("creditBankExchange");
         }
         @Bean
         public HeadersExchange creditFinanceExchange() {
                return new HeadersExchange("creditFinanceExchange");
         } 
    
         // 配置一个routingKey为credit.bank的消息队列并绑定在creditBankExchange交换机上
         @Bean
         public Binding bindingCreditAExchange(Queue creditBankQueue, HeadersExchange creditBankExchange) {
              Map<String,Object> headerValues = new HashMap<>();
              headerValues.put("type", "cash");
              headerValues.put("aging", "fast");
              return BindingBuilder.bind(creditBankQueue).to(creditBankExchange).whereAll(headerValues).match();
        }
    
        // 配置一个routingKey为credit.finance的消息队列并绑定在creditFinanceExchange交换机上
         @Bean
         public Binding bindingCreditBExchange(Queue creditFinanceQueue, HeadersExchange creditFinanceExchange) {
              Map<String,Object> headerValues = new HashMap<>();
              headerValues.put("type", "cash");
              headerValues.put("aging", "fast");
              return BindingBuilder.bind(creditFinanceQueue).to(creditFinanceExchange).whereAny(headerValues).match();
         }
    }

2.测试类

// producer
    @Component
    public class ApiCreditSender {
    
                @Autowired
                private AmqpTemplate rabbitTemplate;
    
                public void creditBank(Map<String, Object> head, String msg){
                       System.out.println("credit.bank send message: "+msg);
                       rabbitTemplate.convertAndSend("creditBankExchange", "credit.bank", getMessage(head, msg));
                }
    
                public void creditFinance(Map<String, Object> head, String msg){
                       System.out.println("credit.finance send message: "+msg);
                       rabbitTemplate.convertAndSend("creditFinanceExchange", "credit.finance", getMessage(head, msg));
                }
    
                private Message getMessage(Map<String, Object> head, Object msg){
                        MessageProperties messageProperties = new MessageProperties();
                        for (Map.Entry<String, Object> entry : head.entrySet()) {
                                 messageProperties.setHeader(entry.getKey(), entry.getValue());
                        }
                        MessageConverter messageConverter = new SimpleMessageConverter();
                        return messageConverter.toMessage(msg, messageProperties);
                 }
    
    }
    
    // consumer
    @Component
    public class ApiCreditReceive {
    
        @RabbitHandler
        @RabbitListener(queues = "credit.bank")
        public void creditBank(String msg) {
               System.out.println("credit.bank receive message: "+msg);
        }
    
         @RabbitHandler
         @RabbitListener(queues = "credit.finance")
         public void creditFinance(String msg) {
               System.out.println("credit.bank receive message: "+msg);
         }
     }
    
    // test
    @RunWith(SpringRunner.class)
    @SpringBootTest
    public class Test {
    
            @Autowired
            private ApiCreditSender sender;
    
             @org.junit.Test
             public void test_creditBank_type() {
                  Map<String,Object> head = new HashMap<>();
                  head.put("type", "cash");
                  sender.creditBank(head, "银行授信(部分匹配)");
             }
    
            @org.junit.Test
            public void test_creditBank_all() {
                 Map<String,Object> head = new HashMap<>();
                 head.put("type", "cash");
                 head.put("aging", "fast");
                 sender.creditBank(head, "银行授信(全部匹配)");
            }
    
            @org.junit.Test
            public void test_creditFinance_type() {
                 Map<String,Object> head = new HashMap<>();
                 head.put("type", "cash");
                 sender.creditFinance(head, "金融公司授信(部分匹配)");
            }
    
            @org.junit.Test
            public void test_creditFinance_all() {
                   Map<String,Object> head = new HashMap<>();
                   head.put("type", "cash");
                   head.put("aging", "fast");
                   sender.creditFinance(head, "金融公司授信(全部匹配)");
            }
    }
    
     // result
     // 发送者日志
     credit.bank send message: 银行授信(部分匹配)
     credit.finance send message: 金融公司授信(全部匹配)
     credit.bank send message: 银行授信(全部匹配)
     credit.finance send message: 金融公司授信(部分匹配)
     // 接收者日志
     credit.bank receive message: 银行授信(全部匹配)
     credit.bank receive message: 金融公司授信(全部匹配)
     credit.bank receive message: 金融公司授信(部分匹配)

根据日志中可以看出4个测试方法均已成功发送消息。但是credit.bank监听的队列有一条消息没有接收到。ApiCreditSenderTests.test\_creditBank\_type()为什么发送的消息，没有被处理？因为在HeadersConfig配置类中，creditBankExchange交换机的匹配规则是完全匹配，即header attribute参数必须完成一致。

### Fanout模式 ###

1.Fanout模式队列配置类：Fanout Exchange交换机是转发消息到所有绑定队列（和routingKey没有关系）,即我们熟悉的广播模式或者订阅模式，给Fanout交换机发送消息，绑定了这个交换机的所有队列都收到这个消息。

@Configuration
     public class FanoutRabbitConfig {
          // 创建队列
         @Bean
         public Queue AMessage() {
              return new Queue("fanout.A");
         }
         // 创建队列
         @Bean
         public Queue BMessage() {
              return new Queue("fanout.B");
         }
         // 创建队列
         @Bean
         public Queue CMessage() {
              return new Queue("fanout.C");
         }
         // 创建Fanout交换器
         @Bean
         FanoutExchange fanoutExchange() {
              return new FanoutExchange("fanoutExchange");
         }
         // 将对列绑定到Fanout交换器
         @Bean
         Binding bindingExchangeA(Queue AMessage, FanoutExchange fanoutExchange) {
              return BindingBuilder.bind(AMessage).to(fanoutExchange);
         }
         // 将对列绑定到Fanout交换器
         @Bean
         Binding bindingExchangeB(Queue BMessage, FanoutExchange fanoutExchange) {
              return BindingBuilder.bind(BMessage).to(fanoutExchange);
         }
         // 将对列绑定到Fanout交换器
         @Bean
         Binding bindingExchangeC(Queue CMessage, FanoutExchange fanoutExchange) {
              return BindingBuilder.bind(CMessage).to(fanoutExchange);
         }
    }

2.发送者

// producer
    @Component
    public class FanoutSender {
    
           @Autowired
           private AmqpTemplate rabbitTemplate;
    
           public void send() {
                 String context = "hi, fanout msg ";
                 System.out.println("Sender : " + context);
                 this.rabbitTemplate.convertAndSend("fanoutExchange","", context);
           }
    
    }
    
    //这里使用了A、B、C三个队列绑定到Fanout交换机上面，发送端的routing_key写任何字符都会被忽略。
    
    // consumer
    @RabbitHandler
    public void process(String message) {
        System.out.println("fanout Receiver A  : " + message);
    }
    ......
    
    // test
    @RunWith(SpringRunner.class)
    @SpringBootTest
    public class Test {
    
           @Autowired
           private FanoutSender fanoutSender;
    
           @org.junit.Test
           public void sendFanout () {
               fanoutSender.send();
           }
    }
    
    // result
    fanout Receiver B: hi, fanout msg 
    fanout Receiver A  : hi, fanout msg 
    fanout Receiver C: hi, fanout msg

以上是RabbitMQ的基本配置和使用，还需要大家继续深入学习理解相关知识。

**如有披露或问题欢迎留言或者入群探讨**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5NDcwNzMz_size_16_color_FFFFFF_t_70][]

[JCccc-RabbitMq]: /images/20220529/de513ffb0892452f810aeb0aced5678c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5NDcwNzMz_size_16_color_FFFFFF_t_70]: /images/20220529/632446a935da4c489a03dfefec3b8b2f.png