您的位置: 首页  >  文章  >  SpringBoot集成RabbitMQ入门篇

SpringBoot集成RabbitMQ入门篇

分类: 文章 • 2023-12-11 10:07:10

RabbitMQ 即一个消息队列，主要是用来实现应用程序的异步和解耦，同时也能起到消息缓冲，消息分发的作用。

消息中间件最主要的作用是解耦，中间件最标准的用法是生产者生产消息传送到队列，消费者从队列中拿取消息并处理，生产者不用关心是谁来消费，消费者不用关心谁在生产消息，从而达到解耦的目的。在分布式的系统中，消息队列也会被用在很多其它的方面，比如：分布式事务的支持，RPC的调用等等。

消息队列

对于消息队列，我们一般知道有三个概念：发消息者、队列、收消息者，RabbitMQ 在这个基本概念之上，多做了一层抽象，在发消息者和队列之间,加入了交换器(Exchange)这样发消息者和队列就没有直接联系，转而变成发消息者把消息给交换器,交换器根据调度策略再把消息再给队列。

                                               SpringBoot集成RabbitMQ入门篇

左侧P代表生产者，也就是往RabbitMQ发消息的程序。

中间即是RabbitMQ，其中绿色的X代表交换机，红色的通道代表队列。

右侧C代表消费者，也就是往RabbitMQ 拿消息的程序。

交换机

交换机的功能主要是接收消息并且转发到绑定的队列。

交换机类型: Direct类型、Topic类型、Headers类型和Fanout类型。 

Direct 是RabbitMQ默认的交换机模式,也是最简单的模式.即创建消息队列的时候,指定一个BindingKey.当发送者发送消息的时候,指定对应的Key.当Key和消息队列的BindingKey一致的时候,消息将会被发送到该消息队列中。

Topic 转发信息主要是依据通配符,队列和交换机的绑定主要是依据一种模式(通配符+字符串),而当发送消息的时候,只有指定的Key和该模式相匹配的时候,消息才会被发送到该消息队列中。

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

Fanout 是路由广播的形式,将会把消息发给绑定它的全部队列,即便设置了key,也会被忽略。

RabbitMQ

RabbitMQ 即一个消息队列，主要是用来实现应用程序的异步和解耦，同时也能起到消息缓冲，消息分发的作用。

聊聊RabbitMQ

传统模式

SpringBoot集成RabbitMQ入门篇

这个流程，全部在主线程完成，注册-》入库-》发送邮件-》发送短信，由于都在主线程，所以要等待每一步完成才能继续执行。由于每一步的操作时间响应时间不固定，所以主线程的请求耗时可能会非常长，如果请求过多，会导致IIS站点巨慢，排队请求，甚至宕机，严重影响用户体验。

 常用方式

SpringBoot集成RabbitMQ入门篇

这个流程是主线程只做耗时非常短的入库操作，发送邮件和发送短信，会开启2个异步线程，扔进去并行执行，主线程不管，继续执行后续的操作，这种处理方式要远远好过第一种处理方式，极大的增强了请求响应速度，用户体验良好。缺点是，由于异步线程里的操作都是很耗时间的操作，一个请求要开启2个线程，而一台标准配置的ECS服务器支撑的并发线程数大概在800左右，假设一个线程在10秒做完，这个单个服务器最多能支持400个请求的并发，后面的就要排队。出现这种情况，就要考虑增加服务器做负载，尴尬的增加成本。

RabbitMQ

SpringBoot集成RabbitMQ入门篇

这个流程是，主线程依旧处理耗时低的入库操作，然后把需要处理的消息写进消息队列中，这个写入耗时可以忽略不计，非常快，然后，独立的发邮件子系统，和独立的发短信子系统，同时订阅消息队列，进行单独处理。处理好之后，向队列发送ACK确认，消息队列整条数据删除。这个流程也是现在各大公司都在用的方式，以SOA服务化各个系统，把耗时操作，单独交给独立的业务系统，通过消息队列作为中间件，达到应用解耦的目的，并且消耗的资源很低，单台服务器能承受更大的并发请求。

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

3、队列配置

@Configuration
 public class RabbitConfig {

      @Bean
      public Queue helloQueue() {
             return new Queue("hello");
      }

 }

4、发送者

@Component
public class HelloSender {

     @Autowired
     private AmqpTemplate rabbitTemplate;

     public void send() {
             String context = "hello " + new Date();
             System.out.println("Sender : " + context);
             this.rabbitTemplate.convertAndSend("hello", context);
     }
}

5、接收者

@Component
@RabbitListener(queues = "hello")
public class HelloReceiver {

     @RabbitHandler
     public void process(String hello) {
            System.out.println("Receiver  : " + hello);
     }

}

6、测试

@RunWith(SpringRunner.class)
@SpringBootTest
public class Test {

     @Autowired
     private HelloSender helloSender;


   // 发送单条消息
    @org.junit.Test
    public void contextLoads() {
           helloSender.send();
    }
 }

7、测试结果

Sender : hello Sun Jan 06 10:58:18 CST 2019
2019-01-06 10:58:18.530  INFO 24784 --- [       Thread-2] 
o.s.a.r.l.SimpleMessageListenerContainer : Waiting for workers to finish.
Receiver  : hello Sun Jan 06 10:58:18 CST 2019

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

RabbitMQ发送对象

1、发送者

public void send(User user) {
    System.out.println("Sender object: " + user.toString());
    this.rabbitTemplate.convertAndSend("object", user);
}

2、接收者

@RabbitHandler
public void process(User user) {
    System.out.println("Receiver object : " + user);
    System.out.println("username:"+user.getUsername());
    System.out.println("password:"+user.getPassword());
}

3、测试方法

// 发送对象
@org.junit.Test
public void sendObj() {
    User user = new User();
    user.setId(1);
    user.setUsername("admin");
    user.setPassword("1234");
    objectSender.send(user);
}

4、测试结果

Sender object: [email protected]
 2019-01-06 11:13:40.175  INFO 26456 --- [       Thread-2] 
 o.s.a.r.l.SimpleMessageListenerContainer : Waiting for workers to finish.
 Receiver object : [email protected]
 username:admin
 password:1234

Topic

Topic 是 RabbitMQ中最灵活的一种方式，可以根据routing_key自由的绑定不同的队列。

1、Topic配置

@Configuration
public class TopicRabbitConfig {

       final static String message = "topic.message";
       final static String messages = "topic.messages";
      // 创建队列
      @Bean
      public Queue queueMessage() {
            return new Queue(TopicRabbitConfig.message);
      }
      // 创建队列
      @Bean
      public Queue queueMessages() {
             return new Queue(TopicRabbitConfig.messages);
      }
      // 将对列绑定到Topic交换器
      @Bean
      TopicExchange exchange() {
            return new TopicExchange("topicExchange");
      }
      // 将对列绑定到Topic交换器
      @Bean
      Binding bindingExchangeMessage(Queue queueMessage, TopicExchange exchange) {
            return BindingBuilder.bind(queueMessage).to(exchange).with("topic.message");
      }
      // 将对列绑定到Topic交换器 采用#的方式
      @Bean
      Binding bindingExchangeMessages(Queue queueMessages, TopicExchange exchange) {
            return BindingBuilder.bind(queueMessages).to(exchange).with("topic.#");
      }
}

2、发送者

public void send1() {
    String context = "hi, i am message 1";
    System.out.println("Sender : " + context);
    this.rabbitTemplate.convertAndSend("topicExchange", "topic.message", context);
}

public void send2() {
    String context = "hi, i am messages 2";
    System.out.println("Sender : " + context);
    this.rabbitTemplate.convertAndSend("topicExchange", "topic.messages", context);
}

3、接收者

@Component
@RabbitListener(queues = "topic.message")
public class TopicReceiver {
       @RabbitHandler
       public void process(String message) {
           System.out.println("Topic Receiver1  : " + message);
       }
}

发送send1会匹配到topic.#和topic.message 两个Receiver都可以收到消息，发送send2只有topic.#可以匹配所有只有Receiver2监听到消息

4、测试类

 @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();
    }

}

发送send1会匹配到topic.#和topic.message 两个Receiver都可以收到消息，发送send2只有topic.#可以匹配所有只有Receiver2监听到消息。

Headers

1、队列配置

配置一个routingKey为credit.bank的消息队列并绑定在creditBankExchange交换机上

配置一个routingKey为credit.finance的消息队列并绑定在creditFinanceExchange交换机上

@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");
     } 

     @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();
    }

     @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、发送者

@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);
             }

}

3、接收者

@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);
     }
 }

4、测试类

@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, "金融公司授信(全部匹配)");
        }
}

5、测试结果

// 发送者日志
 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

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

1、配置队列

 @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、发送者

@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);
       }

}

3、接收者

这里使用了A、B、C三个队列绑定到Fanout交换机上面，发送端的routing_key写任何字符都会被忽略。

@RabbitHandler
public void process(String message) {
    System.out.println("fanout Receiver A  : " + message);
}

4、测试类

@RunWith(SpringRunner.class)
@SpringBootTest
public class Test {

       @Autowired
       private FanoutSender fanoutSender;

       @org.junit.Test
       public void sendFanout () {
           fanoutSender.send();
       }
}

5、测试结果

fanout Receiver B: hi, fanout msg 
fanout Receiver A  : hi, fanout msg 
fanout Receiver C: hi, fanout msg

GitHub地址:

https://github.com/xiaonongOne/springboot-rabbitmq

 

SpringBoot集成RabbitMQ入门篇

相关推荐