高端网站开放,阿里云虚拟主机做多个网站,重庆百度seo排名,宁波职业能力建设网2.9日学习打卡
一.RabbitMQ 死信队列 在MQ中#xff0c;当消息成为死信#xff08;Dead message#xff09;后#xff0c;消息中间件可以将其从当前队列发送到另一个队列中#xff0c;这个队列就是死信队列。而在RabbitMQ中#xff0c;由于有交换机的概念#xff0c;实…2.9日学习打卡
一.RabbitMQ 死信队列 在MQ中当消息成为死信Dead message后消息中间件可以将其从当前队列发送到另一个队列中这个队列就是死信队列。而在RabbitMQ中由于有交换机的概念实际是将死信发送给了死信交换机Dead Letter Exchange简称DLX。死信交换机和死信队列和普通的没有区别。 消息成为死信的情况
队列消息长度到达限制。消费者拒签消息并且不把消息重新放入原队列。消息到达存活时间未被消费。 死信队列 死信交换机和普通交换机只是名字不同其余的都一样就是说死信交换机绑定死信队列普通交换机绑定普通队列。
并且普通队列需要在创建的同时绑定死信交换机和死信队列的路由关键字就是说当普通队列的消息达到死信的条件时会通过绑定的死信交换机根据绑定的死信队列路由关键字发送该普通队列的消息这个就是成为死信的过程。
除了拒签成为死信的需要监听普通队列其余俩个不需要到达消费者端就会成为死信。 死信会有专门的死信消费者去监听。
创建死信队列
package com.jjy.rabbitproducer;import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;Configuration
public class RabbitmqConfig4 {private final String DEAD_EXCHANGE dead_exchange;private final String DEAD_QUEUE dead_queue;private final String NORMAL_EXCHANGE normal_exchange;private final String NORMAL_QUEUE normal_queue;// 死信交换机Bean(DEAD_EXCHANGE)public Exchange deadExchange(){return ExchangeBuilder.topicExchange(DEAD_EXCHANGE).durable(true).build();}// 死信队列Bean(DEAD_QUEUE)public Queue deadQueue(){return QueueBuilder.durable(DEAD_QUEUE).build();}// 死信交换机绑定死信队列Beanpublic Binding bindDeadQueue(Qualifier(DEAD_EXCHANGE) Exchange exchange, Qualifier(DEAD_QUEUE)Queue queue){return BindingBuilder.bind(queue).to(exchange).with(dead_routing).noargs();}// 普通交换机Bean(NORMAL_EXCHANGE)public Exchange normalExchange(){return ExchangeBuilder.topicExchange(NORMAL_EXCHANGE).durable(true).build();}// 普通队列Bean(NORMAL_QUEUE)public Queue normalQueue(){return QueueBuilder.durable(NORMAL_QUEUE).deadLetterExchange(DEAD_EXCHANGE) // 绑定死信交换机.deadLetterRoutingKey(dead_routing) // 死信队列路由关键字.ttl(10000) // 消息存活10s.maxLength(10) // 队列最大长度为10.build();}// 普通交换机绑定普通队列Beanpublic Binding bindNormalQueue(Qualifier(NORMAL_EXCHANGE) Exchange exchange,Qualifier(NORMAL_QUEUE)Queue queue){return BindingBuilder.bind(queue).to(exchange).with(my_routing).noargs();}
}
测试死信队列
生产者发送消息 Testpublic void TestDlx(){// 存活时间过期后变成死信rabbitTemplate.convertAndSend(normal_exchange,my_routing,测试死信);// 超过队列长度后变成死信for (int i 0; i 20; i) {rabbitTemplate.convertAndSend(normal_exchange,my_routing,测试死信);}// 消息拒签但不返回原队列后变成死信rabbitTemplate.convertAndSend(normal_exchange,my_routing,测试死信);}消费者拒收消息
package com.jjy.rabbitconsumer;import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;import java.io.IOException;Component
public class DlxConsumer {RabbitListener(queues normal_queue)public void listenMessage(Message message, Channel channel) throws IOException {// 拒签消息channel.basicNack(message.getMessageProperties().getDeliveryTag(),true,false);}
}
二. RabbitMQ 延迟队列 延迟队列即消息进入队列后不会立即被消费只有到达指定时间后才会被消费。
例如用户下单后30分钟后查询订单状态未支付则会取消订单 但RabbitMQ中并未提供延迟队列功能我们可以使用死信队列实现延迟队列的效果。
延迟队列_死信队列实现 编写配置文件 spring:rabbitmq:host: 192.168.66.100port: 5672username: jjypassword: jjyvirtual-host: /# 日志格式
logging:pattern:console: %d{HH:mm:ss.SSS} %clr(%-5level) --- [%-15thread] %cyan(%-50logger{50}):%msg%n创建队列和交换机
package com.jjy.yanchirabbitmqdemo;import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;Configuration
public class RabbitmqConfig {// 订单交换机和队列private final String ORDER_EXCHANGE order_exchange;private final String ORDER_QUEUE order_queue;// 过期订单交换机和队列private final String EXPIRE_EXCHANGE expire_exchange;private final String EXPIRE_QUEUE expire_queue;// 过期订单交换机Bean(EXPIRE_EXCHANGE)public Exchange deadExchange(){return ExchangeBuilder.topicExchange(EXPIRE_EXCHANGE).durable(true).build();}// 过期订单队列Bean(EXPIRE_QUEUE)public Queue deadQueue(){return QueueBuilder.durable(EXPIRE_QUEUE).build();}// 将过期订单队列绑定到交换机Beanpublic Binding bindDeadQueue(Qualifier(EXPIRE_EXCHANGE) Exchange exchange,Qualifier(EXPIRE_QUEUE) Queue queue){return BindingBuilder.bind(queue).to(exchange).with(expire_routing).noargs();}// 订单交换机Bean(ORDER_EXCHANGE)public Exchange normalExchange(){return ExchangeBuilder.topicExchange(ORDER_EXCHANGE).durable(true).build();}// 订单队列Bean(ORDER_QUEUE)public Queue normalQueue(){return QueueBuilder.durable(ORDER_QUEUE).ttl(10000) // 存活时间为10s,模拟30min.deadLetterExchange(EXPIRE_EXCHANGE) // 绑定死信交换机.deadLetterRoutingKey(expire_routing) // 死信交换机的路由关键字.build();}// 将订单队列绑定到交换机Beanpublic Binding bindNormalQueue(Qualifier(ORDER_EXCHANGE) Exchange exchange,Qualifier(ORDER_QUEUE) Queue queue){return BindingBuilder.bind(queue).to(exchange).with(order_routing).noargs();}
}
编写下单的控制器方法下单后向订单交换机发送消息
RestController
public class OrderController {Autowiredprivate RabbitTemplate rabbitTemplate;GetMapping(/place/{orderId})public String placeOrder(PathVariable String orderId){System.out.println(处理订单数据);//将订单id发送到订单队列rabbitTemplate.convertAndSend(order_exchange,order_routing,orderId);return 下单成功修改库存;}}
编写监听死信队列的消费者
package com.jjy.yanchirabbitmqdemo.consumer;import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;Component
public class ExpireOrderConsumer {//监听过期订单队列RabbitListener(queues expire_queue)public void listenMessage(String orderId){System.out.println(查询orderId订单状态如果已支付无需处理未支付回退库存);}RabbitListener(queues delayed_queue)public void listenMessage2(String orderId){System.out.println(查询orderId号订单的状态如果已支付则无需处理如果未支付则需要回退库存);}}
RabbitMQ延迟队列_插件实现 在使用死信队列实现延迟队列时会遇到一个问题RabbitMQ只会移除队列顶端的过期消息如果第一个消息的存活时长较长而第二个消息的存活时长较短则第二个消息并不会及时执行。 RabbitMQ虽然本身不能使用延迟队列但官方提供了延迟队列插 件安装后可直接使用延迟队列。 安装延迟队列插件
使用rz将rabbitmq_delayed_message_exchange-3.10.2.ez 插件上传到虚拟机安装插件
# 将插件放入RabbitMQ插件目录中
mv rabbitmq_delayed_message_exchange-
3.9.0.ez /usr/local/rabbitmq/plugins/
# 启用插件
rabbitmq-plugins enable
rabbitmq_delayed_message_exchange重启RabbitMQ服务
#停止rabbitmq
rabbitmqctl stop
#启动rabbitmq
rabbitmq-server restart -detached此时登录管控台可以看到交换机类型多了延迟消息 管控台地址https://自己地址:15672默认账号和密码都是 guest 此时交换机是x-delayed-message类型的延迟队列就是说发送延迟交换机及其绑定延迟队列的路由关键字和MessagePostProcessor对象MessagePostProcessor对象是用来设置延迟时长等属性的。
延迟队列只是创建延迟交换机时跟普通不同其他基本相同。
执行流程就是浏览器通过url映射到Controller的RequestMapping注解的value中然后执行方法体执行到发送MQ然后结束方法体MQ此时延迟方法体中设置的时长再发送给消费端消费端监听到再处理延迟队列的MQ。 使用延迟队列
package com.jjy.yanchirabbitmqdemo;import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import java.util.HashMap;
import java.util.Map;Configuration
public class RabbitmqConfig2 {public final String DELAYED_EXCHANGE delayed_exchange;public final String DELAYED_QUEUE delayed_queue;//1.延迟交换机Bean(DELAYED_EXCHANGE)public Exchange delayedExchange() {// 创建自定义交换机MapString, Object args new HashMap();args.put(x-delayed-type, topic); // topic类型的延迟交换机return new CustomExchange(DELAYED_EXCHANGE, x-delayed-message, true, false, args);}//2.延迟队列Bean(DELAYED_QUEUE)public Queue delayedQueue() {return QueueBuilder.durable(DELAYED_QUEUE).build();}// 3.绑定Beanpublic Binding bindingDelayedQueue(Qualifier(DELAYED_QUEUE) Queue queue, Qualifier(DELAYED_EXCHANGE) Exchange exchange) {return BindingBuilder.bind(queue).to(exchange).with(order_routing).noargs();}
}
编写下单的控制器方法
package com.jjy.yanchirabbitmqdemo.controller;import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;RestController
public class OrderController {Autowiredprivate RabbitTemplate rabbitTemplate;GetMapping(/place2/{orderId})public String placeOrder2(PathVariable String orderId) {System.out.println(处理订单数据...);// 设置消息延迟时间为10秒MessagePostProcessor messagePostProcessor new MessagePostProcessor() {Overridepublic Message postProcessMessage(Message message) throws AmqpException {message.getMessageProperties().setDelay(10000);return message;}};// 将订单id发送到订单队列rabbitTemplate.convertAndSend(delayed_exchange, order_routing, orderId, messagePostProcessor);return 下单成功修改库存;}}
编写消费者
package com.jjy.yanchirabbitmqdemo.consumer;import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;Component
public class ExpireOrderConsumer {//监听过期订单队列RabbitListener(queues expire_queue)public void listenMessage(String orderId){System.out.println(查询orderId订单状态如果已支付无需处理未支付回退库存);}RabbitListener(queues delayed_queue)public void listenMessage2(String orderId){System.out.println(查询orderId号订单的状态如果已支付则无需处理如果未支付则需要回退库存);}}
三. RabbitMQ集群 在生产环境中当单台RabbitMQ服务器无法满足消息的吞吐量及 安全性要求时需要搭建RabbitMQ集群。
搭建RabbitMQ集群
设置两个RabbitMQ服务
# 关闭RabbitMQ服务
rabbitmqctl stop
# 设置服务一
RABBITMQ_NODE_PORT5673 RABBITMQ_NODENAMErabbit1 rabbitmq-server
start -detached
# 设置服务二
RABBITMQ_NODE_PORT5674 RABBITMQ_SERVER_START_ARGS-rabbitmq_management listener
[{port,15674}] RABBITMQ_NODENAMErabbit2 rabbitmq-server start -detached将两个服务设置到同一集群中
# 关闭服务2
rabbitmqctl -n rabbit2 stop_app
# 重新设置服务2
rabbitmqctl -n rabbit2 reset
# 将服务2加入服务1中
rabbitmqctl -n rabbit2 join_cluster rabbit1localhost(自己的主机名)
# 启动服务2
rabbitmqctl -n rabbit2 start_app
这个主机名可以到管控台看 后面的就是主机名
镜像队列 搭建了集群后虽然多个节点可以互相通信但队列只保存在了一个节点中如果该节点故障则整个集群都将丢失消息。
# 关闭服务1
rabbitmqctl -n rabbit1 stop_app 此时我们需要引入镜像队列机制它可以将队列消息复制到集群中的其他节点上。如果一个节点失效了另一个节点上的镜像可以保证服务的可用性。
在管此时某个节点故障则不会影响整个集群。控台点击Admin—Policies设置镜像队列
Name为镜像策略的名字;Pattern模式的值为^即应用于所有交换机和队列 ^xxx则是应用到以xxx开头交换机和队列。 ha-modeall则是镜像模式所有即将队列镜像到集群的所有服务器中。 ha-sync-mode: automatic设置为自动同步即当某一服务器宕机而有新的消息发送到来当该服务器再次启动后它已经拿不到该消息了但是通过设置该参数可以自动同步消息。
负载均衡 无论是生产者还是消费者只能连接一个RabbitMQ节点而在我们使用RabbitMQ集群时如果只连接一个RabbitMQ节点会造成该节点的压力过大。我们需要平均的向每个RabbitMQ节点发送请求此时需要一个负载均衡工具帮助我们分发请求接下来使用Haproxy做负载均衡 安装Haproxyyum -y install haproxy 配置Haproxyvim /etc/haproxy/haproxy.cfg 添加下面内容
# 以下为修改内容
defaults# 修改为tcpmode tcp# 以下为添加内容
listen rabbitmq_cluster# 对外暴露端口bind 0.0.0.0:5672mode tcpbalance roundrobin# 代理RabbitMQ的端口server node1 127.0.0.1:5673 check inter 5000 rise 2 fall 2server node2 127.0.0.1:5674 check inter 5000 rise 2 fall 2listen stats# Haproxy控制台路径bind 自己地址:8100mode httpoption httplogstats enablestats uri /rabbitmq-statsstats refresh 5s访问Haproxy控制台http://自己地址/rabbitmq-stats
生产者连接Haproxy发送消息
// 生产者
public class Producer {public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory connectionFactory new ConnectionFactory();connectionFactory.setHost(192.168.66.100);connectionFactory.setPort(5672);connectionFactory.setUsername(guest);connectionFactory.setPassword(guest);connectionFactory.setVirtualHost(/);Connection conn connectionFactory.newConnection();Channel channel conn.createChannel();channel.queueDeclare(simple_queue, false, false, false, null);channel.basicPublish(, simple_queue, null, hello!rabbitmq!.getBytes());channel.close();conn.close();}
}
总结
由Springboot整合RabbitMQ实现通配符模式通配符模式使用topic交换机展示消息从生产者到消费者的一个过程。 生产者由Config配置类yml配置文件还有发送消息的一个调用操作组成而消费者由yml配置文件和多个监听指定队列方法的类组成。 RabbitMQ的Config配置类用来创建交换机、创建队列和将队列绑定到交换机此时yml配置文件的内容是用来连接RabbitMQ的信息和一些模式的开关。 模式有确认模式Confirm回退模式(Return)消费者消息确认Consumer Ack 三种前俩种用在生产者第三种用在消费者。
确认模式是用以消息发送到交换机此时不论成功或失败都会调用setConfirmCallback返回发送的结果成功或失败在失败时打印结果并处理失败的业务成功时处理成功后的业务。回退模式则是消息从交换机发到队列的过程当失败时调用setReturnsCallback方法输出失败的原因等信息和失败后的处理操作。消费者消息确认可以设置手动和自动确认自动确认因消费者监听方法内出现问题或者其他原因会导致消息丢失所以一般是手动确认在正常处理后确认签收在出现异常时捕获异常处理异常并拒签。 整个过程总而言之即从生产者发送消息到指定交换机交换机再根据生产者给的路由键发送到指定队列队列再发送到消费者。
在RabbitMQ的高级属性中有消费端限流、公平与不公平分发、消息存活时间、优先级队列、死信队列、延迟队列。
消费端限流就是说从队列到消费者的消息消费者最多只能存在x条未处理的消息。公平分发就是每个消费者都得到平均分发的消息一般不设置限流不公平分发则是处理快的继续处理没有数额限制但是需要设置限流且值为1。消息存活时间可以设置队列消息的存活时间时需要在配置类在创建队列时指定也可以设置单条消息的存活时间但是需要在发送消息时创建并添加MessageProperties对象设置存活时间作为参数俩这都是存活时间一过消息等到队列顶端后移除。优先级队列即需要在创建队列时指定队列的最大优先级然后在发送消息时创建并添加MessageProperties对象设置优先级数作为参数。死信队列即分别创建俩个交换机和俩个队列并将队列绑定到交换机一个作为死信队列和交换机一个作为普通队列和交换机但是普通队列创建时需要绑定死信交换机和死信队列的路由键并且当普通队列消息过期、拒收并且不重回普通队列、超过普通队列设置的最大消息长度时将消息发送到死信交换机死信交换机再发到死信队列死信队列有专门的消费者进行监听。延迟队列有2种实现一种是通过死信队列设置普通队列的存活时间为延迟的时长通过死信消费者处理延迟的消息第二种是通过rabbitmq的延迟队列插件实现延迟队列。
3. RabbitMQ搭建集群及其实现镜像队列、负载均衡
搭建集群就是多个独立的服务器在一个系统中组成一个群体用以提高系统的可用性、处理能力、响应能力防止宕机出现数据丢失和解决服务器无法承受高并发的等问题。镜像队列就是复制主节点的数据到其他的从节点中防止因为主节点宕机导致数据丢失。负载均衡就是将请求访问平均分发到集群的每个节点中从而提高系统的性能不会导致某个节点因为访问量过大从而出现系统性能降低或者该节点宕机。
关于RabbitMQ的学习就告一段落了 下次就是学习第二个消息中间件RocketMQ
如果我的内容对你有帮助请点赞评论收藏。创作不易大家的支持就是我坚持下去的动力
