RabbitMQ学习笔记（下）：延迟队列，发布确认高级，备份交换机

十、延迟队列

延迟队列

概念：

延迟队列使用场景：

流程图：

延迟队列整合Springboot

导入依赖：

<dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>1.2.47</version></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId></dependency><dependency><groupId>io.springfox</groupId><artifactId>springfox-swagger2</artifactId><version>2.9.2</version></dependency><dependency><groupId>io.springfox</groupId><artifactId>springfox-swagger-ui</artifactId><version>2.9.2</version></dependency><dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit-test</artifactId><scope>test</scope></dependency></dependencies>

在java/com/atguigu/rabbitmq下创建config创建类SwaggerConfig，写入代码：

@Configuration
@EnableSwagger2
public class SwaggerConfig {@Beanpublic Docket webApiConfig(){return new Docket(DocumentationType.SWAGGER_2).groupName("webApi").apiInfo(webApiInfo()).select().build();}private ApiInfo webApiInfo() {return new ApiInfoBuilder().title("rabbitmq接口文档").description("本文档描述了rabbitmq微服务接口定义").version("1.0").contact(new Contact("enjoy6288","http://atguigu.com","1551388580@qq.com")).build();}
}

队列TTL代码框架图：

队列TTL（配置类代码）：

@Configuration
public class TtlQueueConfig {//普通交换机的名称public static final String X_EXCHANGE = "X";//死信交换机的名称public static final String Y_DEAD_LETTER_EXCHANGE = "Y";//普通队列的名称public static final String QUEUE_A = "QA";public static final String QUEUE_B = "QB";//死信队列的名称public static final String DEAD_LETTER_QUEUE = "QD";//声明xExchange@Bean("xExchange")public DirectExchange xExchange(){return new DirectExchange(X_EXCHANGE);}//声明yExchange@Bean("yExchange")public DirectExchange yExchange(){return new DirectExchange(Y_DEAD_LETTER_EXCHANGE);}//声明普通队列A的TTL为10s@Bean("queueA")public Queue queueA(){Map<String,Object> arguments = new HashMap<>(3);//设置死信交换机arguments.put("x-dead-letter-exchange",Y_DEAD_LETTER_EXCHANGE);//设置死信RoutingKeyarguments.put("x-dead-letter-routing-key","YD");//设置TTLarguments.put("x-message-ttl",10000);return QueueBuilder.durable(QUEUE_A).withArguments(arguments).build();}//声明普通队列B的TTL为40s@Bean("queueB")public Queue queueB(){Map<String,Object> arguments = new HashMap<>(3);//设置死信交换机arguments.put("x-dead-letter-exchange",Y_DEAD_LETTER_EXCHANGE);//设置死信RoutingKeyarguments.put("x-dead-letter-routing-key","YD");//设置TTLarguments.put("x-message-ttl",40000);return QueueBuilder.durable(QUEUE_B).withArguments(arguments).build();}//死信队列@Bean("queueD")public Queue queueD() {return QueueBuilder.durable(DEAD_LETTER_QUEUE).build();}//绑定A-X@Beanpublic Binding queueABindingX(@Qualifier("queueA") Queue queueA,@Qualifier("xExchange") DirectExchange xExchange){return BindingBuilder.bind(queueA).to(xExchange).with("XA");}//绑定B-x@Beanpublic Binding queueBBindingX(@Qualifier("queueB") Queue queueB,@Qualifier("xExchange") DirectExchange xExchange){return BindingBuilder.bind(queueB).to(xExchange).with("XB");}//绑定D-y@Beanpublic Binding queueDBindingY(@Qualifier("queueD") Queue queueD,@Qualifier("yExchange") DirectExchange yExchange){return BindingBuilder.bind(queueD).to(yExchange).with("YD");}
}

队列TTL（生产者）：

@Slf4j
@RestController
@RequestMapping("/ttl")
public class SendMsgController {@Autowiredprivate RabbitTemplate rabbitTemplate;@GetMapping("/sendMsg/{message}")public void sendMsg(@PathVariable String message){log.info("当前时间：{},发送一条信息给两个TTL队列:{}",new Date().toString(),message);rabbitTemplate.convertAndSend("X","XA","消息来自ttl为10s的队列:"+message);rabbitTemplate.convertAndSend("X","XB","消息来自ttl为40s的队列:"+message);}
}

队列TTL（消费者）

@Slf4j
@Component
public class DeadLetterQueueConsumer {//接收消息@RabbitListener(queues="QD")public void receiveD(Message message, Channel channel) throws Exception{String msg = new String(message.getBody());log.info("当前时间：{}，收到死信队列的消息：{}",new Date().toString(),msg);}
}

延迟队列优化

不能为需求增加队列，

写一个通用队列作为延迟队列：

配置类，在上面配置类中加入如下代码：

public static final String QUEUE_C = "QC";
//和死信交换机连接
@Bean("queueC")
public Queue queue(){Map<String,Object> arguments = new HashMap<>(3);//设置死信交换机arguments.put("x-dead-letter-exchange",Y_DEAD_LETTER_EXCHANGE);//设置死信RoutingKeyarguments.put("x-dead-letter-routing-key","YD");return QueueBuilder.durable(QUEUE_C).withArguments(arguments).build();
}
//和普通交换机绑定
@Bean
public Binding queueCBindingX(@Qualifier("queueC") Queue queueC,@Qualifier("xExchange") DirectExchange xExchange){return BindingBuilder.bind(queueC).to(xExchange).with("XC");
}

在前面生产者的基础上写入如下代码：

@Slf4j
@RestController
@RequestMapping("/ttl")
public class SendMsgController {@Autowiredprivate RabbitTemplate rabbitTemplate;@GetMapping("/sendExpirationMsg/{message}/{ttlTime}")public void sendMsg(@PathVariable String message,@PathVariable String ttlTime){log.info("当前时间:{},发送一条时长{}毫秒TTL信息给队列QC:{}",new Date().toString(),ttlTime,message);rabbitTemplate.convertAndSend("X","XC",message,msg->{//发送消息的时候延迟时长msg.getMessageProperties().setExpiration(ttlTime);return msg;});}
}

点击启动类重新启动，在网页端输入：localhost:8080/ttl/sendExpirationMsg/你好1/20000和localhost:8080/ttl/sendExpirationMsg/你好2/2000。

问题：延迟队列是排队的，当队列中有多条消息时，延迟队列的消息会根据前面最长时间发送。

安装延迟队列插件：

在https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases下载rabbitmq_delayed_message_exchange插件，解压放置到RabbitMQ的插件目录。

rabbitmq的插件在：/usr/lib/rabbitmq/lib/rabbitmq_server-3.8.8/plugins目录下，上传插件到目录下，如果上传失败用sudo rz先获得权限。

输入：rabbitmq-plugins enable rabbitmq_delayed_message_exchange

重启rabbitmq：systemctl restart rabbitmq-server

进入rabbitmq的交换机界面，查看下面是否出现，如果出现代表安装成功：

基于插件的延迟队列：

配置类

@Configuration
public class DelayedQueueConfig {//队列public static final String DELAYED_QUEUEE_NAME="delayed.queue";//交换机public static final String DELAYED_EXCHANGE_NAME="delayed.exchange";//routingKeypublic static final String DELAYED_ROUTING_KEY="delayed.routingkey";//声明队列@Beanpublic Queue delayedQueue(){return new Queue(DELAYED_QUEUEE_NAME);}//声明交换机,基于插件的@Beanpublic CustomExchange delayedExchange(){Map<String,Object> arguments = new HashMap<>();arguments.put("x-delayed-type","direct");//交换机名称，交换机类型,是否需要持久化，是否需要自动删除，其它参数return new CustomExchange(DELAYED_EXCHANGE_NAME,"x-delayed-message",true,false,arguments);}//绑定@Beanpublic Binding delayedQueueBindingDelayedExchange(@Qualifier("delayedQueue") Queue delayedQueue,@Qualifier("delayedExchange") CustomExchange delayedExchange){return BindingBuilder.bind(delayedQueue).to(delayedExchange).with(DELAYED_ROUTING_KEY).noargs();}
}

生产者

//发消息，基于插件的消息及延迟时间
@GetMapping("/sendDelayMsg/{message}/{delayTime}")
public void sendMsg(@PathVariable String message,@PathVariable  Integer delayTime){log.info("当前时间:{},发送一条时长{}毫秒信息给延迟队列delayed.queue:{}",new Date().toString(),delayTime,message);rabbitTemplate.convertAndSend(DelayedQueueConfig.DELAYED_EXCHANGE_NAME,DelayedQueueConfig.DELAYED_ROUTING_KEY,message,msg->{//发送消息时延迟时间（毫秒）msg.getMessageProperties().setDelay(delayTime);return msg;});
}

消费者，创建DelayQueueConsumer类：

@Slf4j
@Component
public class DelayQueueConsumer {//监听消息@RabbitListener(queues = DelayedQueueConfig.DELAYED_QUEUEE_NAME)public void receiveDelayQueue(Message message){String msg = new String(message.getBody());log.info("当前时间：{},收到延迟队列的消息:{}",new Date().toString(),msg);}}

测试：localhost:8080/ttl/sendDelayMsg/come on baby1/20000。localhost:8080/ttl/sendDelayMsg/come on baby2/2000

十一、发布确认高级

加入交换机或者队列两者有其中一者宕掉，消息都会丢失。

配置类：

@Configuration
public class ConfirmConfig {//交换机public static final String CONFIRM_EXCHANGE_NAME="confirm.exchange";//队列public static final String CONFIRM_QUEUE_NAME="confirm.queue";//RoutingKeypublic static final String CONFIRM_ROUTING_KEY="key1";//声明交换机@Bean("confirmExchange")public DirectExchange confirmExchange(){return new DirectExchange(CONFIRM_EXCHANGE_NAME);}//声明队列@Bean("confirmQueue")public Queue confirmQueue(){return QueueBuilder.durable(CONFIRM_QUEUE_NAME).build();}//绑定@Beanpublic Binding queueBindingExchange(@Qualifier("confirmQueue") Queue confirmQueue,@Qualifier("confirmExchange") DirectExchange confirmExchange){return BindingBuilder.bind(confirmQueue).to(confirmExchange).with(CONFIRM_ROUTING_KEY);}
}

生产者：

@RestController
@Slf4j
@RequestMapping("/confirm")
public class ProducerController {@Autowiredprivate RabbitTemplate rabbitTemplate;//发消息@GetMapping("/sendMessage/{message}")public void sendMessage(@PathVariable String message){rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE_NAME,ConfirmConfig.CONFIRM_ROUTING_KEY,message);log.info("发送消息内容：{}",message);}
}

消费者：

@Slf4j
@Component
public class Consumer {@RabbitListener(queues = ConfirmConfig.CONFIRM_QUEUE_NAME)public void receiveConfirmMessagee(Message message){String msg = new String(message.getBody());log.info("接受到的队列confirm.queue消息：{}",msg);}
}

实验步骤：先输入localhost:8080/confirm/sendMessage/大家好1。

MyCallBack是一个实现类，继承了RabbitTemplate.ConfirmCallback接口，因为不再里面，所以到时候掉接口根本调不到实现类，因此需要注入。

回调接口：

先在配置文件中写入如下：

spring.rabbitmq.publisher-confirm-type=correlated

@Slf4j
@Component
public class MyCallBack implements RabbitTemplate.ConfirmCallback {@Autowiredprivate RabbitTemplate rabbitTemplate;@PostConstructpublic void init(){//注入rabbitTemplate.setConfirmCallback(this);}/*交换机确认回调方法* 1.发消息，交换机接收到了，回调*  1.1.correlationData保存回调消息的ID及相关信息*  1.2.交换机收到消息 ack = true*  1.3.cause null* 2.发消息，交换机接收失败了，回调*  2.1.correlationData保存回调信息的ID及相关信息*  2.2.交换机收到消息 ack = false*  2.3.cause 失败的原因*/@Overridepublic void confirm(CorrelationData correlationData,boolean ack,String cause){String id = correlationData!=null?correlationData.getId():"";if(ack){log.info("交换机已经收到了Id为：{}的消息",id);}else{log.info("交换机已经收到了Id为：{}的消息,由于原因：{}",id,cause);}}
}

如果交换机没有确认或者确认是失败的，都证明是失败的，会确认回调。

交换机确认

@RestController
@Slf4j
@RequestMapping("/confirm")
public class ProducerController {@Autowiredprivate RabbitTemplate rabbitTemplate;//发消息@GetMapping("/sendMessage/{message}")public void sendMessage(@PathVariable String message){CorrelationData correlationData = new CorrelationData("1");rabbitTemplate.convertAndSend(ConfirmConfig.CONFIRM_EXCHANGE_NAME,ConfirmConfig.CONFIRM_ROUTING_KEY,message,correlationData);log.info("发送消息内容：{}",message);}
}

输入：localhost:8080/confirm/sendMessage/大家好1，测试回调。

回退消息

退回的情况是当routingkey出现错误，差不多就是队列出现问题时，需要回退错误的消息。

Mandatory参数：

首先在配置文件中加入下面代码：

spring.rabbitmq.publisher-returns=true

代码是在上面MyCallBack基础上进行修改，首先在implements后面加上RabbitTemplate.ReturnCallback，记得在init里注入setReturnCallback，returnedMessage是重写的方法，点击MyCallBack然后按住alt+enter重写。

@Slf4j
@Component
public class MyCallBack implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnCallback {@Autowiredprivate RabbitTemplate rabbitTemplate;@PostConstructpublic void init(){//注入rabbitTemplate.setConfirmCallback(this);rabbitTemplate.setReturnCallback(this);}//可以在当消息传递过程中不可达目的地时将消息返回给生产者//只有不可达目的地的时候，才进行回退@Overridepublic void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {log.error("消息{},被交换机{}退回，退回原因：{}，路由Key：{}",new String(message.getBody()),exchange,replyText,routingKey);}
}

十二、备份交换机

要写一个备份交换机，一个备份队列，一个报警队列。

配置类ConfirmConfig，注意要【修改确认交换机】，使其与备份交换机关联；注意要在绑定的方法上面加上@Bean：

//声明确认交换机
@Bean("confirmExchange")
public DirectExchange confirmExchange(){return ExchangeBuilder.directExchange(CONFIRM_EXCHANGE_NAME).durable(true).withArgument("alternate-exchange",BACKUP_EXCHANGE_NAME).build();
}
public static final String BACKUP_EXCHANGE_NAME = "backup_exchange";
public static final String BACKUP_QUEUE_NAME="backup_queue";
public static final String WARNING_QUEUE_NAME="warning_queue";
//备份交换机
@Bean("backupExchange")
public FanoutExchange backupExchange(){return new FanoutExchange(BACKUP_EXCHANGE_NAME);
}
//备份队列
@Bean("backupQueue")
public Queue backupQueue(){return QueueBuilder.durable(BACKUP_QUEUE_NAME).build();
}
//报警队列
@Bean("warningQueue")
public Queue warningQueue(){return QueueBuilder.durable(WARNING_QUEUE_NAME).build();
}
//绑定
@Bean
public Binding backupQueueBindingBackupExchange(@Qualifier("backupQueue") Queue backupQueue,@Qualifier("backupExchange") FanoutExchange backupExchange){return BindingBuilder.bind(backupQueue).to(backupExchange);
}
@Bean
public Binding warningQueueBindingBackupExchange(@Qualifier("warningQueue") Queue warningQueue,@Qualifier("backupExchange") FanoutExchange backupExchange){return BindingBuilder.bind(warningQueue).to(backupExchange);
}

结果分析：

在consumer下创建一个WarningConsumer类，写入如下代码：

@Component
@Slf4j
public class WarningConsumer {//接收报警消息@RabbitListener(queues = ConfirmConfig.WARNING_QUEUE_NAME)public void receiveWarningMsg(Message message){String msg = new String(message.getBody());log.error("报警发现不可路由消息：{}",msg);}
}

然后要在rabbitmq可视化界面将【confirm交换机删除掉】：

若想实现如下效果需要更改routingkey的值，比如加上一个“2”。

由此看出备份交换机的优先级高于回退消息。

幂等性：

概念：

消息重复消费：

解决思路：

消费端的幂等性保障：

1. 唯一ID+指纹码机制

2.利用Redis原子性

优先级队列使用场景：

优先级队列0到255越大越优先执行

优先级队列代码实现：

生产者代码如下：

public class producer {//队列名称public static final String QUEUE_NAME = "hello";//发消息public static void main( String[] args ) throws IOException, TimeoutException {//第1步：创建一个连接工程ConnectionFactory factory = new ConnectionFactory();//第2步：输入工厂IP，用户名和密码——连接RabbitMQd队列factory.setHost("192.168.182.157");factory.setUsername("admin");factory.setPassword("123");//第3步：创建连接Connection connection = factory.newConnection();//第4步：获取信道Channel channel = connection.createChannel();Map<String,Object> arguments = new HashMap<>();arguments.put("x-max-priority",10);channel.queueDeclare(QUEUE_NAME,true,false,false,arguments);for(int i=1;i<11;i++){String message = "info"+i;if(i==5){AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().priority(5).build();channel.basicPublish("",QUEUE_NAME,properties,message.getBytes());channel.basicPublish("",QUEUE_NAME,null,message.getBytes());}else{channel.basicPublish("",QUEUE_NAME,null,message.getBytes());}}}
}

测试：生产者先将所有消息发到队列当中，然后再由消费者消费。先启动生产者，再启动消费者。最后info5在控制台最上面，是最后打印的，符合逻辑。

十三、集群

惰性队列：

消息保存在内存中还是在磁盘上。正常情况：消息保存在内存中。惰性队列：消息保存在磁盘中。惰性队列消费速度很慢，需要先提取到内存中。使用场景：消息堆积太多比如有100万，但消费者宕机了。

使用场景：

两种模式：

集群原理：

搭建集群：

先关机，克隆出2台机器，然后全部启动。查看ip地址，然后用xshell连接上3台主机。

第1步：修改3台机器的主机名称。在第1台主机上输入hostname，显示node1。在第2台和第3台主机上输入vi /etc/hostname，然后把node1分别改成node2和node3，要输入【reboot】进行重启。

第2步：配置各个节点的hosts文件，让各个节点能互相识别对方。vi /etc/hosts。把3条ip node值，赋值到3个Xshell会话。

第3步：确保各个节点的cookie文件使用的是同一个值。在node1上执行远程操作命令：

scp /var/lib/rabbitmq/.erlang.cookie root@node2:/var/lib/rabbitmq/.erlang.cookie
scp /var/lib/rabbitmq/.erlang.cookie root@node3:/var/lib/rabbitmq/.erlang.cookie

第4步：启动RabbitMQ服务，顺带启动Erlang虚拟机和RabbitMQ应用服务（在三台节点上分别执行下命令）：rabbitmq-server -detached

第5步：在节点2执行：

rabbitmqctl stop_app
rabbitmqctl reset
rabbitmqctl join_cluster rabbit@node1
rabbitmqctl start_app

第6步：在节点3执行：

rabbitmqctl stop_app
rabbitmqctl reset
rabbitmqctl join_cluster rabbit@node2
rabbitmqctl start_app

第7步：查看集群状态：rabbitmqctl cluster_status

第8步：重新设置用户：创建账号：rabbitmqctl add_user admin 123。设置用户角色：rabbitmqctl set_user_tags admin administrator。设置用户权限：rabbitmqctl set_permissions -p "/" admin ".*" ".*" ".*"

第9步：解除集群节点（node2和node3机器分别执行）

镜像队列：

镜像是备份，接收到的消息不应该只在一个节点上存在，否则一个节点宕机，消息就消失了。

ha-mode:备机模式。ha-param:填写备份数量。ha-sync-mode备份模式。

实验：在1号机上创建1个队列，然后会自动备份一份在2号或3号机上。现在节点1关闭，备份的数据会被存储到另外

实现高可用负载均衡：

VIP连接主机，如果主机宕机了，会自动将连接漂移到备机。备机还会时不时询问主机的存在。需要Haproxy以及keepalive软件。

高可用主机宕机备机接管。

十四、Federation

1.首先要开启多个节点，然后让每台节点单独运行。

2.在每台机器上开启federation插件：先输入：rabbitmq-plugins enable rabbitmq_federation再输入：rabbitmq-plugins enable rabbitmq_federation_management

Exchange实现：

下面是部分代码：

public static final String FED_EXCHANGE="fed_exchange";
channel.exchangeDeclare(FED_EXCHANGE,BuiltinExchangeType.DIRECT);
channel.queueDeclare("node2_queue",true,false,false,null);
channel.queueBind("node2_queue",FED_EXCHANGE,"routeKey");

下面是具体的配置：