做长直播的房地产网站宁波seo优化服务
RabbitMq安装
类型概念
租户
RabbitMQ 中有一个概念叫做多租户，每一个 RabbitMQ 服务器都能创建出许多虚拟的消息服务器，这些虚拟的消息服务器就是我们所说的虚拟主机（virtual host），一般简称为 vhost。
每一个 vhost 都是一个独立的小型 RabbitMQ 服务器，这个 vhost 中会有自己的消息队列、消息交换机以及相应的绑定关系等等，并且拥有自己独立的权限，不同的 vhost 中的队列和交换机不能互相绑定，这样技能保证运行安全又能避免命名冲突
交换机
| 交换机 | |||
| 属性 | 意义 | 值 | 意义 |
| type | 类型 | direct | 默认的直接交换机 根据交换机下队列绑定的routingKey直接匹配 |
| fanout | 扇形交换机 简单来说就是发布订阅 队列直接绑定在交换机下，统一发布消息 | ||
| headers | 头部交换机，通过message header头部信息进行比对 可以根据定义全匹配、部分匹配等规则 | ||
| topic | 主题交换机 通过绑定routingKey进行模糊匹配 | ||
| Durability | 耐用 （持久化） | durable | 持久化，数据存放于硬盘 |
| transient | 瞬态，数据存放于内存 | ||
| Auto delete | 自动删除 | Yes | 没有绑定队列时自动删除，针对的是曾经有过但后来没有的事物 |
| No | 不自动删除 | ||
| Internal | 内部使用 | Yes | 该路由绑定的队列不会被用户消费 |
| No | 不自动删除 | ||
队列
| 队列 | |||
| 属性 | 意义 | 值 | 意义 |
| type | 类型 | Default for virtual host | 租户配置的默认选项，下列三种其一 默认Classic无需设置 |
| Classic | 传统的队列类型 数据存储在单个节点上 不具备quorum队列的高可用性和数据保护特性 ps:单机时使用 | ||
| Quorum | 高可用性队列 数据会被复制到多个节点 提供更好的数据可靠性和持久性 ps:部署多节点时使用 | ||
| Stream | 特殊类型的队列 用于支持事件流处理（event streaming） 具有类似于Kafka的流式处理特性 ps:听说不成熟，暂时用不上 | ||
| Durability | 耐用 （持久化） | durable | 持久化，数据存放于硬盘 |
| 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:5672username: adminpassword: xxxxxxvirtual-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);}@Beanpublic Binding bindingSimpleQueue1(@Qualifier(FANOUT_QUEUE1) Queue fanoutQueue1,@Qualifier(FANOUT_EXCHANGE) FanoutExchange fanoutExchange) {// 将交换机和队列绑定return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);}@Beanpublic 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 {@AutowiredRabbitMessagingTemplate rabbitMessagingTemplate;@Testpublic 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的交互进行细粒度的控制。
可以设置消息的属性、监听发送确认、接收确认等功能，更加灵活地处理消息发送和接收的细节。
public interface RabbitMqPublish {void send(String quene, String message);void send(String exchange, String routingKey, String message);void send(String quene, String message, Integer expiration);void send(String exchange, String routingKey, String message, Integer expiration);}
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 RabbitMqPublishImpl implements RabbitMqPublish {@Autowiredprivate 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);}/*** 直接发送消息到队列* 超过有效期丢弃** @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);}/*** 发送消息* 超过有效期丢弃** @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，消息过期后转移消息时指定的routingKeyheaders.put("x-dead-letter-routing-key",DIRECT_GP_DEAD_LETTER_QUEUE);// 如果只配置了有效期，未配置交换机和routingKey，则消息会被直接丢弃return new Queue(DIRECT_QUEUE1, true, false, false,headers);}
配置完成后，尝试向DIRECT_QUEUE1发送一条消息，不启动消费者，10秒后消息会自动转移到死信队列中，可在可视化管理界面进行验证。
延时队列
延时队列场景举例：
预定一个会议室，两个小时后开始，要求提前十分钟通知参会人员进行开会。
如果不使用延时队列，那么就需要不断轮询，查看是否到达需要通知的时间，进行消息通知。
延时队列的实现方式：
死信队列+消息有效期
预定时间到提前十分钟通知中间有110分钟，那么创建一条通知消息，设置有效期110分钟丢入队列，不用消费者去监听，等待消息过期后路由到指定的死信队列，再去消费死信队列中的消息即可。
所以延时队列实际上是一种实现方案，而不是一种特定的队列类型。