SpringBoot整合RabbitMQ

Spring-AMQP是Spring框架的AMQP消息解决方案，提供模板化的发送和接收消息的抽象层，提供基于消息驱动的POJO的消息监听等。

提供不依赖于任何特定的AMQP代理实现或客户端库通用的抽象，最终用户代码将很容易实现更易替换、添加和删除AMQP，因为它可以只针对抽象层来开发
总之就是提高我们的框架整合消息队列的效率,SpringBoot为更方便开发RabbitMQ推出了starter
我们使用 spring-boot-starter-amqp 进行开发

在SpringBoot项目中添加依赖：

xml

<!--引入AMQP-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

基本配置

web管控台添加虚拟主机

我们在同一个项目中，可能会出现开发、测试包括上线用的都是同一个消息队列，如果不进行隔离，很可能会出现开发环境不小心把线上环境的消息进行消费了，因此添加虚拟主机，达到一个消息隔离的效果。

SpringBoot配置RabbitMQ

在application.yml中进行配置

yaml

spring:
  rabbitmq:
    host: 1.5.1.26
    port: 5672
    virtual-host: /dev #这是我上面添加的虚拟主机
    password: password
    username: admin

创建配置类RabbitMQConfig

首先定义交换机和队列的名称，然后使用Bean注入的方式，注入交换机和队列对象，最后再绑定二者关系，注意导包

java

package com.xk857.config;

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @author cv大魔王
 * @version 1.0
 * @description
 * @date 2022/8/7
 */
@Configuration
public class RabbitMQConfig {

    /**
     * 交换机名称
     */
    public static final String EXCHANGE_NAME = "order_exchange";

    /**
     * 队列名称
     */
    public static final String QUEUE = "order_queue";


    @Bean
    public Exchange orderExchange() {
        // 创建交换机，durable代表持久化，使用Bean注入
        return ExchangeBuilder.topicExchange(EXCHANGE_NAME).durable(true).build();
    }

    @Bean
    public Queue orderQueue() {
        // 创建队列，使用Bean注入
        return QueueBuilder.durable(QUEUE).build();
    }

    /**
     * 交换机和队列绑定关系
     * @param queue 上面注入的队列Bean，如果你的项目又多个，记得给Bean取名字
     * @param exchange 上面注入的交换机Bean
     */
    @Bean
    public Binding orderBinding(Queue queue, Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("order.#").noargs();
    }

}

消息生产者发送消息

java

@SpringBootTest
class RabbitmqDemoApplicationTests {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    void send() {
        rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME,"order.new","新订单来啦！！");
    }
}

消息消费者监听消息

消息发送使用SpringBoot测试类进行发送，消息接收我们创建消息监听类，进行消息接收。

java

package com.xk857.mq;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.io.IOException;

/**
 * @author cv大魔王
 * @description 消息消费者监听消息
 * @date 2022/8/7
 */
@Slf4j
@Component
@RabbitListener(queues = "order_queue") // 监听的队列名称
public class OrderMQListener {

    /**
     * RabbitHandler会自动匹配消息类型（消息自动确认）
     * @param msg 我们发送的是String类型，这里用String进行接收，RabbitHandler会自动进行匹配
     * @param message
     * @throws IOException
     */
    @RabbitHandler
    public void releaseCouponRecord(String msg, Message message) throws IOException {
        long msgTag = message.getMessageProperties().getDeliveryTag();
        log.info("监听到消息：消息内容，msg={}",msg); // 监听到消息：消息内容，msg=新订单来啦！！
        log.info("msgTag={}",msgTag); // msgTag=1
        log.info("message={}",message.toString()); // message=(Body:'新订单来啦！！' MessageProperties [headers={}, ……
    }
}

RabbitMQ实现延时队列

什么是延迟队列？Producer将消息发送到消息队列服务端，但并不期望这条消息立马投递，而是推迟到在当前时间点之后的某一个时间投递到Consumer进行消费，该消息即定时消息。例如超时关单、优惠券回收等场景会用到。

RabbitMQ本身是不支持延迟队列的，我们可以通过死信队列的特性来实现延时队列。

创建延时队列配置类

说明：创建死信队列和创建延时队列没什么不同，当成正常的队列创建就行了。

java

@Configuration
public class RabbitMQTTLConfig {

    /**
     * 死信队列
     */
    public static final String LOCK_MERCHANT_DEAD_QUEUE = "lock_merchant_dead_queue";

    /**
     * 死信交换机
     */
    public static final String LOCK_MERCHANT_DEAD_EXCHANGE = "lock_merchant_dead_exchange";

    /**
     * 进入死信队列的路由key
     */
    public static final String LOCK_MERCHANT_ROUTING_KEY = "lock_merchant_routing_key";


    /**
     * 创建死信交换机
     */
    @Bean("lockMerchantDeadExchange")
    public Exchange lockMerchantDeadExchange() {
        return new TopicExchange(LOCK_MERCHANT_DEAD_EXCHANGE, true, false);
    }

    /**
     * 创建死信队列
     */
    @Bean("lockMerchantDeadQueue")
    public Queue lockMerchantDeadQueue() {
        return QueueBuilder.durable(LOCK_MERCHANT_DEAD_QUEUE).build();
    }

    /**
     * 绑定死信交换机和死信队列，和创建普通队列没什么区别
     */
    @Bean
    public Binding lockMerchantBinding(@Qualifier("lockMerchantDeadQueue") Queue queue, @Qualifier("lockMerchantDeadExchange") Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with(LOCK_MERCHANT_ROUTING_KEY).noargs();
    }
    
    
    

    /**
     * 普通队列，绑定的个死信交换机
     */
    public static final String NEW_MERCHANT_QUEUE = "new_merchant_queue";

    /**
     * 普通的topic交换机
     */
    public static final String NEW_MERCHANT_EXCHANGE = "new_merchant_exchange";

    /**
     * 路由key
     */
    public static final String NEW_MERCHANT_ROUTTING_KEY = "new_merchant_routing_key";


    /**
     * 创建普通交换机
     */
    @Bean("newMerchantExchange")
    public Exchange newMerchantExchange() {
        return new TopicExchange(NEW_MERCHANT_EXCHANGE, true, false);
    }

    /**
     * 创建普通队列
     */
    @Bean("newMerchantQueue")
    public Queue newMerchantQueue() {
        Map<String, Object> args = new HashMap<>(3);
        //消息过期后，进入到死信交换机
        args.put("x-dead-letter-exchange", LOCK_MERCHANT_DEAD_EXCHANGE);
        //消息过期后，进入到死信交换机的路由key
        args.put("x-dead-letter-routing-key", LOCK_MERCHANT_ROUTING_KEY);
        //过期时间，单位毫秒，10秒
        args.put("x-message-ttl", 10000);
        return QueueBuilder.durable(NEW_MERCHANT_QUEUE).withArguments(args).build();
    }

    /**
     * 绑定交换机和队列
     */
    @Bean
    public Binding newMerchantBinding(@Qualifier("newMerchantQueue") Queue queue, @Qualifier("newMerchantExchange") Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with(NEW_MERCHANT_ROUTTING_KEY).noargs();
    }
}

延时队列-发送消息

思考一下，发送消息是发给谁的？发送消息是发给普通队列的，普通队列的消息过期会进入死信队列，然后我们监听死信队列的消息，但是发送消息是发给普通队列的。这次发消息不使用测试类发送了，换成发送

java

@RestController("/")
public class TestController {
    
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @GetMapping("/send/ttl/{msg}")
    public boolean sendTTLMsg(@PathVariable String msg) {
        rabbitTemplate.convertAndSend(RabbitMQTTLConfig.NEW_MERCHANT_EXCHANGE, RabbitMQTTLConfig.NEW_MERCHANT_ROUTTING_KEY, 
                                      "超时关单信息，10秒后接收订单信息，" + msg);
        return true;
    }

}

延时队列-接收消息

消息消费者写法上没有不同，注意监听的是死信队列即可

java

@Component
@RabbitListener(queues = "lock_merchant_dead_queue") // 监听的队列名称是死信队列的名称
public class TTLMQListener {
    
    @RabbitHandler
    public void releaseCouponRecord(String msg, Message message) throws IOException {
        log.info("监听到延迟消息：消息内容，msg={}", msg);
    }
}

SpringBoot整合RabbitMQ ​

基本配置 ​

web管控台添加虚拟主机 ​

SpringBoot配置RabbitMQ ​

创建配置类RabbitMQConfig ​

消息生产者发送消息 ​

消息消费者监听消息 ​

RabbitMQ实现延时队列 ​

创建延时队列配置类 ​

延时队列-发送消息 ​

延时队列-接收消息 ​

SpringBoot整合RabbitMQ

基本配置

web管控台添加虚拟主机

SpringBoot配置RabbitMQ

创建配置类RabbitMQConfig

消息生产者发送消息

消息消费者监听消息

RabbitMQ实现延时队列

创建延时队列配置类

延时队列-发送消息

延时队列-接收消息