java-zh

java-zh

关注
发信
关注(28)粉丝(399)

单个springboot整合rabbitmq

一、rabbitmq的搭建

centos7搭建rabbitmq:centos7安装rabbitmq_java-zh的博客-CSDN博客

二、在什么情况下选择rabbitMQ

 常见的四种MQ比较

三、案例(rabbitMQ常见的五种模式讲解)

3.1 导包

<modelVersion>4.0.0</modelVersion>
<parent>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-parent</artifactId>
   <version>2.4.1</version>
   <relativePath/>
</parent>
<properties>
   <java.version>1.8</java.version>
</properties>
<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>
   <version>3.0.4</version>
 </dependency>
</dependencies>

yml初始配置

spring:
 rabbitmq:
    #如果是集群，用,隔开
    host: 192.168.139.128
    #端口，不能写默认端口15672
    port: 5672
    connection-timeout: 15000
    username: zhonghui
    password: 123456
    virtual-host: /
    #none值是禁用发布确认模式，是默认值
    #correlated值是发布消息成功到交换器后会触发回调方法
    #simple有两种效果，一种是和correlated值一样会触发回调方法,
    #另一种是发布消息成功后使用rabbitTemplate调用waitForConfirms或waitForConfirmsOrDie方法等待broker节点返回发送结果，根据返回结果来判定下一步逻辑,
    #要注意的点是waitForConfrimsOrDie方法如果返回false则会关闭Channel,则接下来无法发送消息到broker
    publisher-confirm-type: correlated
    #消息回退确认机制
    publisher-returns: true
    listener:
      simple:
        acknowledge-mode: auto

3.2 简单模式

单个springboot整合rabbitmq-LMLPHP

  • P:生产者
  • C:消费者 

Producer

@RestController
@AllArgsConstructor
@RequestMapping("/test/mq")
public class SimpleQueueProducer {

    private RabbitTemplate rabbitTemplate;

    // 发送到的队列名称
    public static final String AMQP_SIMPLE_QUEUE = "amqp.simple.queue";

    @GetMapping("/simple/{context}")
    public void sendMessage(@PathVariable String context) {
        for (int i = 0; i < 10; i++) {
            rabbitTemplate.convertAndSend(AMQP_SIMPLE_QUEUE, context + i);
        }
    }
}

Consumer

@Component
@Slf4j
public class SimpleQueueConsumer {

    @RabbitListener(queuesToDeclare = @Queue(name = SimpleQueueProducer.AMQP_SIMPLE_QUEUE))
    public void consumerSimpleMessage(String context){
        // 通过Message解析对象消息
        log.info("简单模式内容为:{}",context);
    }
}

结果

2023-06-20 20:01:43.198  INFO 10440 --- [ntContainer#3-1] c.m.m.controller.SimpleQueueConsumer     : 简单模式内容为:mq的简单模式已经完成1
2023-06-20 20:01:43.198  INFO 10440 --- [ntContainer#3-9] c.m.m.controller.SimpleQueueConsumer     : 简单模式内容为:mq的简单模式已经完成0
2023-06-20 20:01:43.198  INFO 10440 --- [tContainer#3-10] c.m.m.controller.SimpleQueueConsumer     : 简单模式内容为:mq的简单模式已经完成3
2023-06-20 20:01:43.266  INFO 10440 --- [ntContainer#3-8] c.m.m.controller.SimpleQueueConsumer     : 简单模式内容为:mq的简单模式已经完成2
2023-06-20 20:01:43.266  INFO 10440 --- [ntContainer#3-7] c.m.m.controller.SimpleQueueConsumer     : 简单模式内容为:mq的简单模式已经完成4
2023-06-20 20:01:43.266  INFO 10440 --- [ntContainer#3-6] c.m.m.controller.SimpleQueueConsumer     : 简单模式内容为:mq的简单模式已经完成5
2023-06-20 20:01:43.268  INFO 10440 --- [ntContainer#3-4] c.m.m.controller.SimpleQueueConsumer     : 简单模式内容为:mq的简单模式已经完成7
2023-06-20 20:01:43.268  INFO 10440 --- [ntContainer#3-3] c.m.m.controller.SimpleQueueConsumer     : 简单模式内容为:mq的简单模式已经完成6
2023-06-20 20:01:43.268  INFO 10440 --- [ntContainer#3-2] c.m.m.controller.SimpleQueueConsumer     : 简单模式内容为:mq的简单模式已经完成9
2023-06-20 20:01:43.268  INFO 10440 --- [ntContainer#3-5] c.m.m.controller.SimpleQueueConsumer     : 简单模式内容为:mq的简单模式已经完成8

注解含义

  • @RabbitListener:用于类和方法上
  • @queuesToDeclare:将队列绑到默认交换机上
  • @Queue:队列注解，value为队列名称

3.3 Work模式

单个springboot整合rabbitmq-LMLPHP

Producer

@RestController
@Slf4j
@AllArgsConstructor
@RequestMapping("/test/mq")
public class WorkProducer {
    private RabbitTemplate rabbitTemplate;

    // 发送到的队列名称
    public static final String AMQP_WORK_QUEUE = "amqp.work.queue";
    
    @GetMapping("/work/{context}")
    public void sendMessage(@PathVariable String context) {

        for (int i = 0; i < 10; i++) {
            rabbitTemplate.convertAndSend(AMQP_WORK_QUEUE,context+i);
        }
    }
}

Consumer

@Component
@Slf4j
public class WorkConsumer {

    @RabbitListener(queuesToDeclare = @Queue(value = WorkProducer.AMQP_WORK_QUEUE))
    public void consumerSimpleMessage(String context){
        try {
            // 假设业务逻辑执行了一秒
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        log.info("work1模式内容为:{}",context);
    }
}
@Component
@Slf4j
public class WorkConsumer2 {
    @RabbitListener(queuesToDeclare = @Queue(value = WorkProducer.AMQP_WORK_QUEUE))
    public void consumerSimpleMessage2(String context){
        log.info("work2模式内容为:{}",context);
    }

}

结果

2023-06-21 09:06:07.075  INFO 21492 --- [ntContainer#5-1] com.mq.mqcloud.controller.WorkConsumer2  : work2模式内容为:mq的work模式已经完成1
2023-06-21 09:06:07.076  INFO 21492 --- [ntContainer#5-1] com.mq.mqcloud.controller.WorkConsumer2  : work2模式内容为:mq的work模式已经完成3
2023-06-21 09:06:07.076  INFO 21492 --- [ntContainer#5-1] com.mq.mqcloud.controller.WorkConsumer2  : work2模式内容为:mq的work模式已经完成5
2023-06-21 09:06:07.076  INFO 21492 --- [ntContainer#5-1] com.mq.mqcloud.controller.WorkConsumer2  : work2模式内容为:mq的work模式已经完成7
2023-06-21 09:06:07.077  INFO 21492 --- [ntContainer#5-1] com.mq.mqcloud.controller.WorkConsumer2  : work2模式内容为:mq的work模式已经完成9
2023-06-21 09:06:08.053  INFO 21492 --- [ntContainer#4-1] com.mq.mqcloud.controller.WorkConsumer   : work1模式内容为:mq的work模式已经完成0
2023-06-21 09:06:09.054  INFO 21492 --- [ntContainer#4-1] com.mq.mqcloud.controller.WorkConsumer   : work1模式内容为:mq的work模式已经完成2
2023-06-21 09:06:10.055  INFO 21492 --- [ntContainer#4-1] com.mq.mqcloud.controller.WorkConsumer   : work1模式内容为:mq的work模式已经完成4
2023-06-21 09:06:11.057  INFO 21492 --- [ntContainer#4-1] com.mq.mqcloud.controller.WorkConsumer   : work1模式内容为:mq的work模式已经完成6
2023-06-21 09:06:12.059  INFO 21492 --- [ntContainer#4-1] com.mq.mqcloud.controller.WorkConsumer   : work1模式内容为:mq的work模式已经完成8

解析

  1. 消费者1和消费者2接收的内容是不同的，同一个消息只能被一个消费者获取
  2. 消费者1和消费者2获取到的消息数量是相同的，一个是消费奇数号消息，一个是消费偶数号消息

其实，这样是不合理的，因为消费者2线程线程停顿时间短，应该是消费者2要比消费者1获取到的消息更多，这里面涉及到两个知识点，一个是轮询分发，一个是公平分发。

轮询分发:上面的案例就是轮询分发，在默认情况下，RabbitMQ将逐个发送消息到在序列中的下一个消费者(不考虑每个任务的时长等待等，且提前一次性分配，并非一个一个分配)。每个消费者获得相同数量的消息。

公平分发:使用prefetch配置来限制RabbitMQ只发不超过1条的消息给同一个消费者，当消息处理完毕后，有了反馈，才会进行第二次发送

在配置环境中，增加属性prefetch，这个属性的意思是每次领取多少条消息，消费完以后就继续领取

listener:
  simple:
    acknowledge-mode: auto
    #每次领取二个消息，消费以后再领取
    prefetch: 2

增加属性重新运行可以知道，work2消费了8条，work1消费了一条


2023-06-21 09:48:06.838  INFO 21956 --- [ntContainer#5-1] com.mq.mqcloud.controller.WorkConsumer2  : work2模式内容为:mq的simple模式已经完成3
2023-06-21 09:48:06.838  INFO 21956 --- [ntContainer#5-1] com.mq.mqcloud.controller.WorkConsumer2  : work2模式内容为:mq的simple模式已经完成4
2023-06-21 09:48:06.841  INFO 21956 --- [ntContainer#5-1] com.mq.mqcloud.controller.WorkConsumer2  : work2模式内容为:mq的simple模式已经完成5
2023-06-21 09:48:06.841  INFO 21956 --- [ntContainer#5-1] com.mq.mqcloud.controller.WorkConsumer2  : work2模式内容为:mq的simple模式已经完成7
2023-06-21 09:48:06.841  INFO 21956 --- [ntContainer#5-1] com.mq.mqcloud.controller.WorkConsumer2  : work2模式内容为:mq的simple模式已经完成8
2023-06-21 09:48:06.843  INFO 21956 --- [ntContainer#5-1] com.mq.mqcloud.controller.WorkConsumer2  : work2模式内容为:mq的simple模式已经完成6
2023-06-21 09:48:06.843  INFO 21956 --- [ntContainer#5-1] com.mq.mqcloud.controller.WorkConsumer2  : work2模式内容为:mq的simple模式已经完成9
2023-06-21 09:48:07.828  INFO 21956 --- [ntContainer#4-1] com.mq.mqcloud.controller.WorkConsumer   : work1模式内容为:mq的simple模式已经完成0
2023-06-21 09:48:08.830  INFO 21956 --- [ntContainer#4-1] com.mq.mqcloud.controller.WorkConsumer   : work1模式内容为:mq的simple模式已经完成1

其实也可以通过增加消费端个数属性来实现,按照下面这样配置，通过增加消费端个数，上面的内容都会被work1给消费掉


    listener:
      simple:
        ##manual:手动处理 auto:自动处理
        acknowledge-mode: auto
        #消费端监听个数(即@RabbitListenter开启几个线程去处理)
        concurrency: 10
        #消费端监听的最大个数
        max-concurrency: 10
        #每次领取二个消息，消费以后再领取
        prefetch: 2
        #消费不成功的消息,拒绝入队
        default-requeue-rejected: true  
        retry:
          #开启消息重试
          enabled: true 
          #重试次数.
          max-attempts: 4
          #重试最大间隔时间
          max-interval: 10000 
          #重试初始间隔时间
          initial-interval: 2000

3.4 订阅模式(Fanout消息模型)

x:交换机

单个springboot整合rabbitmq-LMLPHP

Producer

@RestController
@AllArgsConstructor
@RequestMapping("/test/mq")
public class FanoutExchangeProducer {

    private RabbitTemplate rabbitTemplate;

    public static final String EXCHANGE_NAME = "exchange.fanout";

    public static final String EXCHANGE_QUEUE_1 = "exchange.fanout.queue_1";

    public static final String EXCHANGE_QUEUE_2 = "exchange.fanout.queue_2";


    /**
     * 扇形交换机，订阅模式
     *
     * @param context
     */
    @GetMapping("/fanout/{context}")
    public void sendMessage(@PathVariable String context) {
        rabbitTemplate.convertAndSend(FanoutExchangeProducer.EXCHANGE_NAME, "", context);
    }

}

Consumer 

@Component
@Slf4j
public class FanoutExchangeConsumer {


    /**
     * 订阅模式，扇形交换机
     * @param context
     */
    @RabbitListener(bindings =
    @QueueBinding(exchange = @Exchange(value = FanoutExchangeProducer.EXCHANGE_NAME,type = ExchangeTypes.FANOUT),
                  value = @Queue(value = FanoutExchangeProducer.EXCHANGE_QUEUE_1)
    ))
    @RabbitHandler
    public void exchangeFanoutQueue1(String context){
        log.info("通道1接收到的消息为:{}",context);
    }

    /**
     * 订阅模式，扇形交换机
     *
     * @param context
     */
    @RabbitListener(bindings =
    @QueueBinding(exchange = @Exchange(value = FanoutExchangeProducer.EXCHANGE_NAME,type = ExchangeTypes.FANOUT)  ,
                  value = @Queue(value = FanoutExchangeProducer.EXCHANGE_QUEUE_2)
    ))
    @RabbitHandler
    public void exchangeFanoutQueue2(String context){
        log.info("通道2接收到的消息为:{}",context);
    }
    
}

结果

消息放到了交换机以后，通道1和通道2都接收到了消息(相当于通道1和通道2都跟这个交换机订阅了消息)

2023-06-21 11:20:43.262  INFO 4696 --- [ntContainer#4-1] c.m.m.controller.FanoutExchangeConsumer  : 通道1接收到的消息为:mq的fanout模式(订阅模式)
2023-06-21 11:20:43.262  INFO 4696 --- [ntContainer#3-1] c.m.m.controller.FanoutExchangeConsumer  : 通道2接收到的消息为:mq的fanout模式(订阅模式)

3.5 路由模式(Direct消息模型)

路由消息模型是交换机根据routingKey进行消息投递的，每个队列都有自己专属的routingKey，生产者发送消息时，指定交换机和rountingKey,消息到了交换机之后，交换机通过routingKey将消息投递到指定队列

 单个springboot整合rabbitmq-LMLPHP

 Producer

@RestController
@AllArgsConstructor
@RequestMapping("/test/mq")
public class DirectExchangeProducer {

    private RabbitTemplate rabbitTemplate;

    public static final String EXCHANGE_DIRECT = "exchange.direct";

    public static final String EXCHANGE_DIRECT_QUEUE_1 = "exchange.direct.queue_1";

    public static final String EXCHANGE_DIRECT_QUEUE_2 = "exchange.direct.queue_2";

    public static final String EXCHANGE_DIRECT_ROUTING_KEY_1 = "exchange.direct.routing.1";

    public static final String EXCHANGE_DIRECT_ROUTING_KEY_2 = "exchange.direct.routing.2";


    @GetMapping("/direct")
    public void sendMessageDirect(@RequestParam("context") String context, @RequestParam("routingkey") Integer routingkey) {
        if (1 == routingkey) {
            rabbitTemplate.convertAndSend(EXCHANGE_DIRECT, EXCHANGE_DIRECT_ROUTING_KEY_1, context + routingkey);
        } else if (2 == routingkey) {
            rabbitTemplate.convertAndSend(EXCHANGE_DIRECT, EXCHANGE_DIRECT_ROUTING_KEY_2, context + routingkey);
        } else {
            System.out.println("数据非法");
        }
    }
}

Consumer

@Component
@Slf4j
public class DirectExchangeConsumer {

    @RabbitListener(bindings = @QueueBinding(
                exchange = @Exchange(value = DirectExchangeProducer.EXCHANGE_DIRECT, type = ExchangeTypes.DIRECT),
                value = @Queue(value = DirectExchangeProducer.EXCHANGE_DIRECT_QUEUE_1),
                key = DirectExchangeProducer.EXCHANGE_DIRECT_ROUTING_KEY_1))
    public void exchangeDirectRoutingKey1(String context,  Message message) {
        log.info("key1:" + message.getMessageProperties().getReceivedRoutingKey());
        log.info("路由器模式1 接收到的消息为:{}", context);
    }

    @RabbitListener(bindings = @QueueBinding(
            exchange = @Exchange(value = DirectExchangeProducer.EXCHANGE_DIRECT, type = ExchangeTypes.DIRECT),
            value = @Queue(value = DirectExchangeProducer.EXCHANGE_DIRECT_QUEUE_2),
            key = DirectExchangeProducer.EXCHANGE_DIRECT_ROUTING_KEY_2))
    public void exchangeDirectRoutingKey2(String context,  Message message) {
        log.info("key2:" + message.getMessageProperties().getReceivedRoutingKey());
        log.info("路由器模式2 接收到的消息为:{}", context);
    }
    
}

结果

单个springboot整合rabbitmq-LMLPHP

2023-06-22 14:48:49.642  INFO 22476 --- [ntContainer#2-2] c.m.m.controller.DirectExchangeConsumer  : key2:exchange.direct.routing.2
2023-06-22 14:48:49.643  INFO 22476 --- [ntContainer#2-2] c.m.m.controller.DirectExchangeConsumer  : 路由器模式2 接收到的消息为:内容2
2023-06-22 14:48:56.289  INFO 22476 --- [ntContainer#1-1] c.m.m.controller.DirectExchangeConsumer  : key1:exchange.direct.routing.1
2023-06-22 14:48:56.289  INFO 22476 --- [ntContainer#1-1] c.m.m.controller.DirectExchangeConsumer  : 路由器模式1 接收到的消息为:内容1
2023-06-22 14:48:59.037  INFO 22476 --- [ntContainer#2-2] c.m.m.controller.DirectExchangeConsumer  : key2:exchange.direct.routing.2
2023-06-22 14:48:59.038  INFO 22476 --- [ntContainer#2-2] c.m.m.controller.DirectExchangeConsumer  : 路由器模式2 接收到的消息为:内容2

3.6  主题模式(Topic模式，通配符模式)

在Topic模式中，Routingkey不再时固定的字符，而是有了通配符，交换机可以进行模糊匹配队列

RoutingKey一般由一个或者多个单词组成，多个单词以"."分割

"*":匹配一个单词,就只有一个单词 比如topic.*  可以匹配到topic.AB topic.AC
"#":匹配一个或者多个单词 比如topic.#  可以配到到topic.AB也可以匹配到topic.AB.AC

单个springboot整合rabbitmq-LMLPHP

 Producer

@RestController
@AllArgsConstructor
@RequestMapping("/test/mq")
public class TopicExchangeProducer {

    private RabbitTemplate rabbitTemplate;

    public static final String EXCHANGE_TOPIC = "exchange.topic";

    public static final String EXCHANGE_TOPIC_QUEUE_1 = "exchange.topic.queue_1";

    public static final String EXCHANGE_TOPIC_QUEUE_2 = "exchange.topic.queue_2";

    //只要包含了routingkey都会匹配到
    public static final String EXCHANGE_TOPIC_ROUTING_KEY_1 = "#.routingkey.#";

    //routingkey开头，后面接了一个单词的都会匹配到,比如routing.AB
    public static final String EXCHANGE_TOPIC_ROUTING_KEY_2 = "routingkey.*";

    public static final String EXCHANGE_TOPIC_CASE_KEY_1 = "topic.routingkey.case1";

    //如果case_key_2这样写,那么绑定case_key_1的队列一样会接收到，因为case_key_2也一样和key1匹配上
    public static final String EXCHANGE_TOPIC_CASE_KEY_2 = "routingkey.case2";


    @GetMapping("/topic")
    public void sendMessageDirect(@RequestParam("context") String context, @RequestParam("routingkey") Integer routingkey) {
        if (1 == routingkey) {
            rabbitTemplate.convertAndSend(EXCHANGE_TOPIC, EXCHANGE_TOPIC_CASE_KEY_1, context + routingkey);
        } else if (2 == routingkey) {
            rabbitTemplate.convertAndSend(EXCHANGE_TOPIC, EXCHANGE_TOPIC_CASE_KEY_2, context + routingkey);
        } else {
            System.out.println("数据非法");
        }
    }
}

Consumer

@Component
@Slf4j
public class TopicExchangeConsumer {


    @RabbitListener(bindings = @QueueBinding(
            exchange = @Exchange(value = TopicExchangeProducer.EXCHANGE_TOPIC,  type = ExchangeTypes.TOPIC),
            value = @Queue(value = TopicExchangeProducer.EXCHANGE_TOPIC_QUEUE_1),
            key = TopicExchangeProducer.EXCHANGE_TOPIC_ROUTING_KEY_1))
    @RabbitHandler
    public void exchangeTopicRoutingKey1(String context, Message message) {
        System.out.println("key1:"+message.getMessageProperties().getReceivedRoutingKey());
        log.info("主题模式1:内容为:{}", context);
    }

    @RabbitListener(bindings = @QueueBinding(
            exchange = @Exchange(value = TopicExchangeProducer.EXCHANGE_TOPIC, type = ExchangeTypes.TOPIC),
            value = @Queue(value = TopicExchangeProducer.EXCHANGE_TOPIC_QUEUE_2),
            key = ExchangeProducer.EXCHANGE_TOPIC_ROUTING_KEY_2))
    @RabbitHandler
    public void exchangeTopicRoutingKey2(String context,  Message message) {
        System.out.println("key2:"+message.getMessageProperties().getReceivedRoutingKey());
        log.info("主题模式2:内容为:{}", context);
    }

}

结果1

key1:topic.routingkey.case1
2023-06-22 15:32:10.554  INFO 26060 --- [ntContainer#8-1] c.m.m.controller.TopicExchangeConsumer   : 主题模式1:内容为:内容1

结果2 

因为第二个key也会被第一个匹配到，所以会产生两个结果

key1:routingkey.case2
key2:routingkey.case2
2023-06-22 15:32:25.451  INFO 26060 --- [ntContainer#8-1] c.m.m.controller.TopicExchangeConsumer   : 主题模式1:内容为:内容2
2023-06-22 15:32:25.451  INFO 26060 --- [ntContainer#9-1] c.m.m.controller.TopicExchangeConsumer   : 主题模式2:内容为:内容2
06-23 12:23
Powered by LMLPHP ©2024 java-zh 0.009009