RabbitMQ讲解与整合

RabbitMq安装

类型概念

租户
RabbitMQ 中有一个概念叫做多租户，每一个 RabbitMQ 服务器都能创建出许多虚拟的消息服务器，这些虚拟的消息服务器就是我们所说的虚拟主机（virtual host），一般简称为 vhost。
每一个 vhost 都是一个独立的小型 RabbitMQ 服务器，这个 vhost 中会有自己的消息队列、消息交换机以及相应的绑定关系等等，并且拥有自己独立的权限，不同的 vhost 中的队列和交换机不能互相绑定，这样技能保证运行安全又能避免命名冲突
交换机
在这里插入图片描述

交换机
属性	意义	值	意义
type	类型	direct	默认的直接交换机根据交换机下队列绑定的routingKey直接匹配
		fanout	扇形交换机简单来说就是发布订阅队列直接绑定在交换机下，统一发布消息
		headers	头部交换机，通过message header头部信息进行比对可以根据定义全匹配、部分匹配等规则
		topic	主题交换机通过绑定routingKey进行模糊匹配
Durability	耐用（持久化）	durable	持久化，数据存放于硬盘
Durability	耐用（持久化）	transient	瞬态，数据存放于内存
Auto delete	自动删除	Yes	没有绑定队列时自动删除，针对的是曾经有过但后来没有的事物
Auto delete	自动删除	No	不自动删除
Internal	内部使用	Yes	该路由绑定的队列不会被用户消费
Internal	内部使用	No	不自动删除

队列

在这里插入图片描述

队列
属性	意义	值	意义
type	类型	Default for virtual host	租户配置的默认选项，下列三种其一默认Classic无需设置
		Classic	传统的队列类型数据存储在单个节点上不具备quorum队列的高可用性和数据保护特性 ps:单机时使用
		Quorum	高可用性队列数据会被复制到多个节点提供更好的数据可靠性和持久性 ps:部署多节点时使用
		Stream	特殊类型的队列用于支持事件流处理（event streaming）具有类似于Kafka的流式处理特性 ps:听说不成熟，暂时用不上
Durability	耐用（持久化）	durable	持久化，数据存放于硬盘
Durability	耐用（持久化）	transient	瞬态，数据存放于内存

参数：

显示参数	实际参数	作用
Auto expire	x-expires	设置队列的过期时间，单位为毫秒。当队列在指定时间内未被使用，将会被自动删除
Message TTL	x-message-ttl	设置队列中消息的过期时间（Time-To-Live），单位为毫秒。消息在队列中存放的时间超过设定的过期时间后会被自动删除
Overflow behaviour	x-overflow	设置队列溢出行为，可选值为 drop-head（删除最旧的消息）或 reject-publish（拒绝发布新消息）
Single active consumer	x-single-active-consumer	配置队列是否只允许单个消费者消费消息。当设置了x-single-active-consumer参数时，表示队列只允许有一个消费者活跃地消费消息，其他消费者将被阻塞，直到当前的消费者停止消费或断开连接
Dead letter exchange	x-dead-letter-exchange	设置队列中的死信消息转发到的交换机名称。当消息成为死信时，将会被转发到指定的交换机
Dead letter routing key	x-dead-letter-routing-key	设置死信消息转发时的路由键。死信消息将通过指定的路由键转发到目标交换机
Max length	x-max-length	设置队列的最大长度，即队列中消息的最大数量。当队列中消息数量达到设定的最大长度后，新消息将无法入队
Max length bytes	x-max-length-bytes	设置队列消息的最大总字节数。当队列中消息的总字节数达到设定的最大值后，新消息将无法入队
Leader locator	x-queue-leader-locator	配置队列的领导者（Leader）定位器，集群中使用

SpringBoot整合

引入依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
    <version>2.6.3</version>
</dependency>

配置数据源

spring: 
  rabbitmq:
    addresses: xxx.xxx.xx.xx:5672
    username: admin
    password: xxxxxx
    virtual-host: /

配置交换机和队列

@Component
public class RabbitMqConfig {
	// 定义交换机名称
    public static final String FANOUT_EXCHANGE = "fanout.test";
    @Bean(name = FANOUT_EXCHANGE)
    public FanoutExchange fanoutExchange() {
    	// 交换机类型按需创建，这里用的是Fanout，发布订阅，绑定在该交换机下的队列都会收到消息
    	// 参数2：是否持久化
    	// 参数3：是否自动删除
        return new FanoutExchange(FANOUT_EXCHANGE, true, false);
    }
	
	//  定义队列
    public static final String FANOUT_QUEUE1 = "queue1";
    @Bean(name = FANOUT_QUEUE1)
    public Queue fanoutQueue1() {
    	// 后三个不写也行，这是默认值
    	// 参数2：是否持久化数据到磁盘（防止意外关闭数据丢失）
    	// 参数3：是否具有排他性
    	// 参数4：队列不再使用时是否自动删除
        return new Queue(FANOUT_QUEUE1, true, false, false);
    }
	public static final String FANOUT_QUEUE2 = "queue2";
    @Bean(name = FANOUT_QUEUE2)
    public Queue fanoutQueue2() {
        return new Queue(FANOUT_QUEUE2, true, false, false);
    }
		
    @Bean
    public Binding bindingSimpleQueue1(@Qualifier(FANOUT_QUEUE1) Queue fanoutQueue1,
                                       @Qualifier(FANOUT_EXCHANGE) FanoutExchange fanoutExchange) {
        // 将交换机和队列绑定
        return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
    }
    @Bean
    public Binding bindingSimpleQueue2(@Qualifier(FANOUT_QUEUE2) Queue fanoutQueue2,
                                       @Qualifier(FANOUT_EXCHANGE) FanoutExchange fanoutExchange) {
        // 将交换机和队列绑定
        return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
    }
}

测试发一条消息到队列

@SpringBootTest(classes = TemplateApplication.class)
public class RabbitMQTest {

    @Autowired
    RabbitMessagingTemplate rabbitMessagingTemplate;

    @Test
    public void testSent(){
		//指定交换机->指定队列（因为创建的交换机是FanoutExchange，所以绑定该交换机的队列都会收到一条消息）
        rabbitMessagingTemplate.convertAndSend("fanout.test","发送数据到FanoutExchange");
    	// 如果创建队列不绑定交换机和路由键，那么实际上会有默认的交换机和路由键，均为空，直接将消息发送给队列，队列名则和路由键保持一致，仍然可以成功发送消息。
    }
}

测试接收队列消息

写个监听类接收消息：

@Component
public class RabbitMqListenter {
    @RabbitListener(queues = {RabbitMqConfig.FANOUT_QUEUE1,RabbitMqConfig.FANOUT_QUEUE2})
    public void reciveLogAll(String msg) throws Exception {
        System.out.println("消费到数据：" + msg);
    }
}

-------------基础的使用到这里就结束了-------------

拓展事项

rabbitMqPusher

自己封装一个更加方便使用的发送工具，可有可无，其中可以使用RabbitMessagingTemplate和RabbitTemplate，RabbitMessagingTemplate和RabbitTemplate都是Spring AMQP提供的用于与RabbitMQ进行交互的工具类如果只是简单使用，那么RabbitMessagingTemplate就够用了，如果需要更精细的控制，可以选择使用RabbitTemplate。

，但它们在使用方式和功能上有一些不同点：

RabbitMessagingTemplate：

RabbitMessagingTemplate是MessagingTemplate的子类，用于在Spring应用程序中发送和接收消息。
它提供了一种更高级别的抽象，使得在Spring框架中更容易使用消息发送和接收的功能。
可以直接与Spring的消息通道（MessageChannel）集成，方便进行消息的发送和接收。

RabbitTemplate：

RabbitTemplate是Spring AMQP提供的用于与RabbitMQ进行交互的核心类，提供了丰富的方法来发送和接收消息。
它是一个强大而灵活的工具，可以直接与RabbitMQ的交互进行细粒度的控制。
可以设置消息的属性、监听发送确认、接收确认等功能，更加灵活地处理消息发送和接收的细节。

package com.template.rabbitmq.producer.impl;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;


@Component
@Slf4j
public class RabbitMqPusher {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * 发送消息
     * @param quene   队列名称 或 交换机名称
     * @param message 消息内容
     */
    public void send(String quene, String message) {
        rabbitTemplate.send(quene, MessageBuilder.withBody(message.getBytes()).build());
        log.info("发送消息---> quene:{} ---> message:{}", message, quene);
    }

    /**
     * 直接发送消息到队列
     * 超过有效期丢弃,如果队列没有声明x-message-ttl属性则无效
     *
     * @param quene      队列名称
     * @param message    消息内容
     * @param expiration 有效期(毫秒)
     */
    public void send(String quene, String message, Integer expiration) {
        rabbitTemplate.send(quene, MessageBuilder.withBody(message.getBytes()).setExpiration(String.valueOf(expiration)).build());
        log.info("发送消息---> quene:{} ---> message:{} ---> expiration:{}", quene, message, expiration);
    }


    /**
     * 发送消息
     * 超过有效期丢弃,如果队列没有声明x-message-ttl属性则无效
     *
     * @param exchange   交换机名称
     * @param routingKey 路由键
     * @param message    消息内容
     * @param expiration 有效期(毫秒)
     */
    public void send(String exchange, String routingKey, String message, Integer expiration) {
        rabbitTemplate.send(exchange, routingKey, MessageBuilder.withBody(message.getBytes()).setExpiration(String.valueOf(expiration)).build());
        log.info("发送消息---> exchange:{} ---> routingKey:{} ---> message:{} ---> expiration:{}", exchange, routingKey, message, expiration);
    }

    /**
     * 发送消息
     *
     * @param exchange   交换机名称
     * @param routingKey 路由键
     * @param message    消息内容
     */
    public void send(String exchange, String routingKey, String message) {
        rabbitTemplate.send(exchange, routingKey, MessageBuilder.withBody(message.getBytes()).build());
        log.info("发送消息---> exchange:{} ---> routingKey:{} ---> message:{}", exchange, routingKey, message);
    }


}

在RabbitMQ中，如果队列没有设置过期时间（即没有声明x-message-ttl属性），那么即使在发送消息时设置了消息的过期时间也会失效。消息的过期时间只有在队列设置了过期时间的情况下才会生效。
实测以上列代码的方式直接对消息设置有效期是生效的。

死信队列

和普通队列一样，只不过是对其他队列进行配置，将过期的消息路由到死信队列中。
创建死信交换机和死信路由

	// 配置交换机的文件中继续增加配置
	public static final String DIRECT_GP_DEAD_LETTER_EXCHANGE = "DIRECT_GP_DEAD_LETTER_EXCHANGE";
    public static final String DIRECT_GP_DEAD_LETTER_QUEUE = "DIRECT_GP_DEAD_LETTER_QUEUE";
    @Bean(DIRECT_GP_DEAD_LETTER_EXCHANGE)
    public DirectExchangedirectDeadLetterExchange() {
        return new DirectExchange(DIRECT_GP_DEAD_LETTER_EXCHANGE, true, false, new HashMap<>());
    }
    @Bean(DIRECT_GP_DEAD_LETTER_QUEUE)
    public Queue directDeadLetterQueue() {
        return new Queue(DIRECT_GP_DEAD_LETTER_QUEUE, true, false, false, new HashMap<>());
    }

设置队列消息有效期并绑定死信队列

	@Bean(name = DIRECT_QUEUE1)
    public Queue directQueue1() {
        HashMap<String, Object> headers = new HashMap<>();
        // 配置消息有效期，消息发送到队列10秒后如果未被消费者消费，则过期
        headers.put("x-message-ttl",10000);
        // 配置超期交换机，消息过期后会发送到此交换机
        headers.put("x-dead-letter-exchange",DIRECT_GP_DEAD_LETTER_EXCHANGE);
        // 配置超期routingKey，消息过期后转移消息时指定的routingKey
        headers.put("x-dead-letter-routing-key",DIRECT_GP_DEAD_LETTER_QUEUE);
        // 如果只配置了有效期，未配置交换机和routingKey，则消息会被直接丢弃
        return new Queue(DIRECT_QUEUE1, true, false, false,headers);
    }