浏阳网站建设公司,其中包括,wordpress手机页面没有注册,建筑模板规格MQ消息队列
其实队列JDK中本身就有#xff0c;不过这种队列也只能单体服务可能会使用#xff0c;一旦项目使用的分布式架构#xff0c;那么一定还是需要用到一个消息中间件的。我们引入消息队列的原因就是对我们的页面相应速度再优化#xff0c;让用户的体验更好#xff…MQ消息队列
其实队列JDK中本身就有不过这种队列也只能单体服务可能会使用一旦项目使用的分布式架构那么一定还是需要用到一个消息中间件的。我们引入消息队列的原因就是对我们的页面相应速度再优化让用户的体验更好原来下订单可能需要1s等待时间引入队列之后可能只需要50ms。
消息中间件的好处 异步处理 最开始我们执行任务时都是同步的比如下图的第一种模式。我们必须等各个操作的做完才能返回响应这样效率就会很慢。例如发送邮件、发送短信但它们能不能收到其实并不是侧重点。因此可以启动两个线程来执行也就是第二种模式异步执行但是使用消息中间件mq可以让效率更上一层楼我们可以把要处理的任务放进mq中然后直接返回结果至于任务则可以慢慢在后面进行处理。 应用解耦 最开始我们如果需要调用不同服务直接的代码时需要在我们的代码中加上调用其他服务方法的逻辑如果方法需要修改比如逻辑要修改参数要修改我们就要修改源代码有了mq之后我们可以把这些调用都交给mq进行处理。即使下单时库存系统不能正常使用也不影响正常下单。因为下单后订单系统写入消息队列就不再关心其他的后续操作了实现订单系统与库存系统的应用解耦 流量控制 服务器接收用户的请求后先写入消息队列。假如消息队列长度超过最大数量则直接抛弃用户请求或跳转到错误页面。 比如对于大并发量的情况秒杀我们可以先把请求放进mq中不需要立刻处理让服务根据能力处理mq中的请求就可以了达到流量削峰的目的。
MQ的相关概念
消息中间件概述
1.大多应用中可通过消息服务中间件来提升系统异步通信、扩展解耦能力
2.消息服务中两个重要概念
消息代理message broker就是运行消息中间件的服务器,这个服务器替我们接收、发送消息和目的地destination当消息发送者发送消息以后将由消息代理接管消息代理保证消息传递到指定目的地。
3.消息队列主要有两种形式的目的地
队列queue 点对点消息通信point-to-point主题topic 发布publish/订阅subscribe消息通信
4.点对点式
消息发送者发送消息消息代理将其放入一个队列中消息接收者从队列中获 取消息内容消息读取后被移出队列消息只有唯一的发送者和接受者但并不是说只能有一个接收者即点对点可以有很多的消息的接收者但消息的接受者只能有一个谁能拿到消息需要靠抢
5.发布订阅式
发送者发布者发送消息到主题多个接收者订阅者监听订阅这个 主题那么就会在消息到达时同时收到消息
6.JMSJava Message ServiceJAVA消息服务
基于JVM消息代理的规范。ActiveMQ、HornetMQ是JMS实现
7.AMQPAdvanced Message Queuing Protocol
高级消息队列协议也是一个消息代理的规范兼容JMSRabbitMQ是AMQP的实现
8.Spring支持
spring-jms提供了对JMS的支持spring-rabbit提供了对AMQP的支持需要ConnectionFactory的实现来连接消息代理提供JmsTemplate、RabbitTemplate来发送消息JmsListenerJMS、RabbitListenerAMQP注解在方法上监听消息代理发布的消息EnableJms、EnableRabbit开启支持
9.Spring Boot自动配置
JmsAutoConfigurationRabbitAutoConfiguration
10.市面上的MQ产品
ActiveMQ、RabbitMQ、RocketMQ、Kafka
消息代理规范
JMSJava Message ServiceJAVA消息服务 基于JVM消息代理的规范。ActiveMQ、HornetMQ是 JMS 实现AMQPAdvanced Message Queuing Protocol 高级消息队列协议也是一个消息代理的规范兼容JMS RabbitMQ 是 AMQP 的实现
下面我们来看一看JMS和AMQP两个规范协议 RabbitMQ是基于AMQP协议实现的并且兼容JMSActiveMQ是基于JMS实现的。
JMS和AMQP的区别在于JMS面向纯java平台不不支持跨平台而AMQP是可以跨平台假如后台服务有用PHP编写则可以兼容。
JMS和AMQP的简单对比
①.AMQP的消息模型中direct exchange是类比JMS中P2P(Queue),AMQP的其它四种消息模型则是类比于JMS的Topic
②.JMS支持的各种消息类型AMQP只支持byte[]但也无妨最后都可以json序列化后传输
RabbitMQ概念
RabbitMQ是一个由erlang开发的遵循AMQP(Advanved Message Queue Protocol)协议的开源消息队列实现。
核心概念
Message
消息消息是不具名的它由消息头和消息体组成。消息体是不透明的而消息头则由一系列的可选属性组成 这些属性包括routing-key路由键、priority相对于其他消息的优先权、delivery-mode指出该消息可 能需要持久性存储等。
Publisher
消息的生产者也是一个向交换器发布消息的客户端应用程序。
Exchange
交换器用来接收生产者发送的消息并将这些消息路由给服务器中的队列。
Exchange有4种类型direct(默认)fanout, topic, 和headers不同类型的Exchange转发消息的策略有所区别
Queue
消息队列用来保存消息直到发送给消费者。它是消息的容器也是消息的终点。一个消息可投入一个或多个队列。消息一直 在队列里面等待消费者连接到这个队列将其取走。
Binding
绑定用于消息队列和交换器之间的关联。一个绑定就是基于路由键将交换器和消息队列连接起来的路由规则所以可以将交 换器理解成一个由绑定构成的路由表。
Exchange 和Queue的绑定可以是多对多的关系。
Connection
网络连接比如一个TCP连接。
Channel
信道多路复用连接中的一条独立的双向数据流通道。信道是建立在真实的TCP连接内的虚拟连接AMQP 命令都是通过信道 发出去的不管是发布消息、订阅队列还是接收消息这些动作都是通过信道完成。因为对于操作系统来说建立和销毁 TCP 都 是非常昂贵的开销所以引入了信道的概念以复用一条 TCP 连接。
Consumer
消息的消费者表示一个从消息队列中取得消息的客户端应用程序。
Virtual Host
虚拟主机表示一批交换器、消息队列和相关对象。虚拟主机是共享相同的身份认证和加密环境的独立服务器域。每个 vhost 本质上就是一个 mini 版的 RabbitMQ 服务器拥 有自己的队列、交换器、绑定和权限机制。vhost 是 AMQP 概念的基础必须在连接时 指定RabbitMQ 默认的 vhost 是 / 。
Broker
表示消息队列服务器实体。 工作流程 首先生产者客户端会向消息中间件发送MessageMessage由消息头和消息体组成消息头中有一个route-key属性用于标识存储的队列位置消息中间件接收到消息之后会由相应的交换机将消息存储到指定的消息队列中交换机和队列具有绑定关系无论生产者还是消费者客户端想发送或者接收消息都需要使用connnection去创建一个长连接长连接类似于高速公路信道类似于高速公路中的每个车道。RabbitMQ还有一个虚拟主机即类似于Docker中的容器彼此互不干扰不需要创建多个RabbitMQ只需要创建多个虚拟机即可实现向java后台、PHP后台发送消息也可以用虚拟主机实现生产和开发环境其提供一个隔离的RabbitMQ环境。
长连接的好处是当客户端宕机之后RabbitMQ将不会向消费者客户端发送消息而是将消息持久化保证消息不会丢失。 docker安装RabbitMQ
这里不需要下载镜像直接安装。默认会帮你下载
docker run -d --name rabbitmq -p 5671:5671 -p 5672:5672 -p 4369:4369 -p 25672:25672 -p 15671:15671 -p 15672:15672 rabbitmq:management修改RabbitMQ配置为只要启动docker自动重启rabbitMQ
docker update rabbitmq --restartalways下面为RabbitMQ中涉及的端口号
对应端口号解释
4369, 25672 (Erlang发现集群端口)5672, 5671 (AMQP端口)15672 (web管理后台端口)61613, 61614 (STOMP协议端口)1883, 8883 (MQTT协议端口)
可访问的可视化RabbitMQ端口号 : 15672 访问RabbitMQ控制页面
下面关于在RabbitMQ可视化界面的操作就不记笔记了因为大多需要粘贴图片排版出来会很难看相信小伙伴们看视频也是轻松get。
AMQP中消息的路由过程和 Java 开发者熟悉的 JMS 存在一些差别 AMQP 中增加了 Exchange 和 Binding 的角色 生产者把消息发布 到 Exchange 上消息最终到达队列并被消费者接收而 Binding 决定交 换器的消息应该发送到那个队列。 因此这里我们简要说明一下RabbitMQ中的自动创建出的四种消息模型的交换机即Exchange的类型 Direct是点对点模式一个消息只能发送给一个队列且被一个消息接收者接收。只有消息中的路由键routing key和 Binding 中的 binding key 完全一致时 交换器才会将消息发到对应的队列中即路由键与队列名完全匹配如果一个队列绑定到交换机要求其路由键为“dog”那么交换机只会给这个队列转发 routing key 为“dog”的消息不会转发 “dog.puppy”也不会转发“dog.guard” 等等。它是完全匹配、单播的模式。 headers这种也是点对点的同direct。但headers 匹配 AMQP 消息的 header 而不是路由键 headers 交换器和 direct 交换器完全一致但性能差很多目前几乎用不到了。 Fanout广播式只要跟这个交换机绑定的队列都会收到发送到这个交换机的消息。Fanout 交换器不处理路由键只是简单的将队列 绑定到交换器上每个发送到交换器的 消息都会被转发到与该交换器绑定的所 有队列上。它很像子网广播每台子网内 的主机都获得了一份复制的消息。Fanout 类型转发消息是最快的。 topic这种也是广播式不过它会根据路由键进行匹配可以通配符模糊匹配只有匹配成功的队列才会接收到消息。将路由键和某个模式进行匹配此时队列需要绑定到一个模式上。它将路由键和绑定键的字符串切分成单词这些单词之间用点隔开。 识别通配符 #匹配 0 个或多个单词 *匹配一个单词 SpringBoot整合RabbitMQ
导入amqp依赖 dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId/dependency我们引入这个依赖之后RabbitAutoConfiguration这个类也就自然而然引入了我们的项目中这个配置类也就会自动注册进容器将其配置信息生效比如这个类就给容器中自动配置了RabbitTemplate、AmqpAdmin、CachingConnectionFactory、RabbitMessagingTemplate等类
Configuration
ConditionalOnClass({RabbitTemplate.class, Channel.class})
EnableConfigurationProperties({RabbitProperties.class})
Import({RabbitAnnotationDrivenConfiguration.class})
public class RabbitAutoConfiguration {添加RabbitMQ的配置信息看到上面配置类中加载了一个配置文件点进去就会发现这个类绑定了我们spring默认的属性配置文件 ConfigurationProperties(prefix spring.rabbitmq
)
public class RabbitProperties {由此我们就可以去application.properties文件中配置我们rabbitMQ的信息了
spring.rabbitmq.host192.168.10.10
spring.rabbitmq.port5672
spring.rabbitmq.virtual-host/
#其他信息就不需要我们配置了大部分都在RabbitProperties中配置了默认值主启动类添加EnableRabbit注解
EnableRabbit
SpringBootApplication
public class GulimallOrderApplication {public static void main(String[] args) {SpringApplication.run(GulimallOrderApplication.class, args);}
}RabbitMQ在Java中的使用
CachingConnectionFactory
RabbitTemplate使用CachingConnectionFactory作为连接工厂
配置类上标有这样的注解EnableConfigurationProperties(RabbitProperties.class)
向容器中注入CachingConnectionFactory的代码中是从配置文件中加载配置信息的。
spring.rabbitmq为配置的前缀可以指定一些端口号ip地址等信息。
AmqpAdmin
AmqpAdmin是org.springframework.amqp.core下的类这个类主要是用来进行一些资源的创建的如创建交换机队列绑定信息等。通过此类可以用代码的方式创建Exchange、Queue还有Binding
Autowired
AmqpAdmin amqpAdmin;/*** 创建绑定*/
Test
public void createBinding() {// String destination 目的地// DestinationType destinationType 绑定类型队列/交换机// String exchange 交换机名称// String routingKey 路由键//、MapString, Object arguments 参数Binding binding new Binding(hello.queue , Binding.DestinationType.QUEUE, hello, hello.queue,null);amqpAdmin.declareBinding(binding);
}/*** 创建队列*/
Test
public void createMQ() {/*** param name 队列的名称* param durable 是否持久化队列* param exclusive 是否声明为一个独占队列* param autoDelete 如果服务不在使用时是否自动删除队列*/Queue queue new Queue(hello.queue, true, false, false);String s amqpAdmin.declareQueue(queue);log.info(创建queue成功... {}, queue);
}/*** 创建交换机* TopicExchange* FanoutExchange* DirectExchange*/
Test
public void createExchange() {// String name 交换机名称// boolean durable 是否持久化// boolean autoDelete 是否自动删除Exchange exchange new DirectExchange(hello, true, false);amqpAdmin.declareExchange(exchange);log.info(创建exchange成功...);
}RabbitTemplate
这个类就是用来控制消息的收发了。通过RabbitTemplate类中的方法可以像使用Rabbit客户端一样向队列发送消息以及更多其他的操作并且多个重载的”send“(发送消息)方法。
Autowired
RabbitTemplate rabbitTemplate;/**
* convertAndSend(String exchange, String routingKey, Object object)
* String exchange, 交换器
* String routingKey, 路由值
* Object object 消息如果发送的消息是对象我们会使用序列化机制将对象写出去。*/
Test
public void test() {// 发送消息rabbitTemplate.convertAndSend(hello, hello.queue ,msg);
}发送消息如果发送的消息是个对象我们会使用序列化机制将对象写出去。对象必须实现Serializable或者我们想要发送的对象序列化为JSON格式
通过指定不同的MessageConverter来实现可以向容器中注入我们想要的MessageConverter从而使用。 配置MyRabbitConfig让发送的对象类型的消息可以是一个json
添加“com.atguigu.gulimall.order.config.MyRabbitConfig”类代码如下
Configuration
public class MyRabbitConfig {Beanpublic MessageConverter messageConverter() {return new Jackson2JsonMessageConverter();}
}注意
配置MyRabbitConfig配置类向容器中添加一个Json的消息转换器让发送的对象类型的消息可以是一个json如果我们不添加RabbitMQ就会用它配置类中自带的一个对象序列化的转换器因为对象无法直接在网络中传输需要转换成字符串。如果使用RabbitMQ自带的不要忘了在传输对象的类上实现Serialize序列化接口。这里我们添加Json消息转换器代码如下
RabbitListener和RabbitHandler注解
监听消息使用RabbitListener和RabbitHandler主启动类必须有EnableRabbit。
RabbitListener 类方法上监听哪些队列即可RabbitHandler: 标在方法上(重载区分不同的消息)
RabbitListener注解和RabbitHandler都可以接受消息队列中的消息并进行处理。
RabbitListener注解
使用RabbitListener时主启动类必须有EnableRabbit其可以标记方法或类上进行使用
自定义方法的参数可以为以下类型
1、Message message原生消息详细信息。头 体
2、T 发送的消息的类型 可以是我们自定义的对象
3、Channel channel 当前传输数据的信道。
RabbitListener(queues {hello.queue})
public String receiveMessage(Message message, OrderEntity content) {//消息体信息byte[] body message.getBody();// 消息头信息MessageProperties messageProperties message.getMessageProperties();log.info(收到的消息 {}, content);return ok;
}同时要注意Queue可以由很多方法来监听只要收到消息队列就删除消息并且只能有一个方法收到消息。并且一个方法接收消息是一个线性的操作只有处理完一个消息之后才能接收下条消息。
RabbitHandler注解
RabbitHandler标在方法上用于接受不同类型的消息对象。
RabbitHandler标记的方法结合RabbitListenerRabbitHandler使用可以变得更加灵活采用在类上加 RabbitListener 注解标识监听哪些消息队列。在方法上添加RabbitHandler注解重载区分不同的消息。
比如说当两个方法对一个消息队列进行监听时用于监听的两个方法用于接收消息内容的参数不同根据消息的内容可以自动的确定使用那个方法。
RestController
public class RabbitController {AutowiredRabbitTemplate rabbitTemplate;GetMapping(/sendMq)public String sendMq(RequestParam(value num,defaultValue 10) Integer num){for (int i 0; i num; i){//向一个队列中发送两种不同类型的消息if (i%20){OrderReturnApplyEntity orderReturnApplyEntity new OrderReturnApplyEntity();orderReturnApplyEntity.setId(1L);orderReturnApplyEntity.setCreateTime(new Date());orderReturnApplyEntity.setReturnName(哈哈哈);//配置MyRabbitConfig让发送的对象类型的消息可以是一个jsonrabbitTemplate.convertAndSend(hello-java-exchange,hello.java,orderReturnApplyEntity, new CorrelationData(UUID.randomUUID().toString()));}else {OrderEntity entity new OrderEntity();entity.setOrderSn(UUID.randomUUID().toString());rabbitTemplate.convertAndSend(hello-java-exchange,hello.java,entity, new CorrelationData(UUID.randomUUID().toString()));}}return OK;}
}修改“com.atguigu.gulimall.order.service.impl.OrderItemServiceImpl”类代码如下
RabbitListener(queues {hello-java-queue})//queues:声明需要监听的所有队列
Service(orderItemService)
public class OrderItemServiceImpl extends ServiceImplOrderItemDao, OrderItemEntity implements OrderItemService {Overridepublic PageUtils queryPage(MapString, Object params) {IPageOrderItemEntity page this.page(new QueryOrderItemEntity().getPage(params),new QueryWrapperOrderItemEntity());return new PageUtils(page);}/*** 以下参数是我们自定义的spring会自动帮我们解析* 参数1、Message message原生消息详细信息。头体* 参数2、T发送的消息类型 OrderReturnApplyEntity content* 参数3、Channel channel 当前传输数据的通道* Queue:可以很多人都来监听。只要收到消息队列删除消息而且只能有一个收到此消息* 场景* 1、订单服务启动多个同一个消息只能有一个客户端收到* 2)、只有一个消息完全处理完方法运行结束才可以接收到下一个消息*///RabbitListener(queues {hello-java-queue})RabbitHandlerpublic void receiverMessage(Message message,OrderReturnApplyEntity content,Channel channel) throws InterruptedException {//消息体byte[] body message.getBody();//消息头属性信息MessageProperties properties message.getMessageProperties();System.out.println(接收到消息...内容: content);//Thread.sleep(3000);System.out.println(消息处理完成》content.getReturnName());}RabbitHandlerpublic void receiverMessage(OrderEntity orderEntity){System.out.println(接收到消息...内容: orderEntity);}
}可靠投递-发送端确认 服务器收到消息 p-bConfirmCallback pring.rabbitmq.publisher-confirmstrue设置确认回调 ConfirmCallback 消息抵达队列就回调 e-qReturnCallback spring.rabbitmq.publisher-returns: truespring.rabbitmq.template.mandatory: true设置确认回调 ReturnCallback
ConfirmCallback ConfirmCallback和RetruhnCallback一样都是RabbitTemplate内部的接口。
消息只要被 broker 接收到就会执行 confirmCallback如果是 cluster 模式需要所有 broker 接收到才会调用 confirmCallback。
也就是说当消息到达RabbitMQ的服务器就会执行回调方法。但是被 broker 接收到只能表示 message 已经到达服务器并不能保证消息一定会被投递到目标 queue 里所以需要用到接下来的 returnCallback。
首先需要修改配置文件
#开启发送端确认
spring.rabbitmq.publisher-confirmstrue然后准备一个发送消息使用的接口和两个用来监听消息队列并接收消息的方法
发送消息接口
RestController
public class SendMsgController {AutowiredRabbitTemplate rabbitTemplate;GetMapping(/sendMsg)public String sendMsg() {for (int i 0; i 10; i) {if (i % 2 0) {OrderEntity orderEntity new OrderEntity();orderEntity.setId(1L);orderEntity.setMemberUsername(Tom);orderEntity.setReceiveTime(new Date());rabbitTemplate.convertAndSend(hello-java-exchange, hello.news, orderEntity, new CorrelationData(UUID.randomUUID().toString()));} else {OrderReturnReasonEntity orderReturnReasonEntity new OrderReturnReasonEntity();orderReturnReasonEntity.setCreateTime(new Date());orderReturnReasonEntity.setId(2L);orderReturnReasonEntity.setName(test);orderReturnReasonEntity.setSort(1);rabbitTemplate.convertAndSend(hello-java-exchange, hello.news, orderReturnReasonEntity, new CorrelationData(UUID.randomUUID().toString()));}}return ok;}
}监听消息队列并接收消息的方法
RabbitListener(queues {hello.news})
Slf4j
Service(orderItemService)
public class OrderItemServiceImpl extends ServiceImplOrderItemDao, OrderItemEntity implements OrderItemService {RabbitHandlerpublic void receiveMessage1(Message message, OrderReturnReasonEntity content, Channel channel) {//消息体信息byte[] body message.getBody();// 消息头信息MessageProperties messageProperties message.getMessageProperties();System.out.println(receiveMessage1 接收消息: content);}RabbitHandlerpublic void receiveMessage2(Message message, OrderEntity content, Channel channel) {//消息体信息byte[] body message.getBody();// 消息头信息MessageProperties messageProperties message.getMessageProperties();System.out.println(receiveMessage2 接收消息: content);}
}第三步在配置类中定制RedisTemplate
Configuration
public class MyRabbitConfig {AutowiredRabbitTemplate rabbitTemplate;PostConstruct // 该注解表示在初始化构造器之后就调用初始化定制 RabbitTemplatepublic void initRabbitTemplate() {// 设置确认回调rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {/**** param correlationData 当前消息的唯一相关数据 这个是消息的唯一id* param ack 消息是否成功收到* param cause 失败的原因*/Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {System.out.println(ConfirmCallback... correlationData: [ correlationData ] ack: [ ack ] cause: [ cause ]);}});}
}那么一旦消息抵达消息队列服务器就会调用我们自己定义的配置类中实现的回调方法打印输出消息信息。
ReturnCallback
首先修改application.properties配置上回调确认
#开启发送端抵达队列确认
spring.rabbitmq.publisher-returnstrue
#只要抵达队列以异步发送优先回调我们这个returnConfirm
spring.rabbitmq.template.mandatorytrue被 broker 接收到只能表示 message 已经到达服务器并不能保证消息一定会被投递到目标 queue 里。所以需要用到接下来的 returnCallback 。
如果在交换机将消息投递到queue的过程中发生了某些问题最终导致消息投递失败就会触发这个方法。
为定制的RabbitTemplate添加这个方法
rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {/*** param message 投递失败的消息的详细信息* param replyCode 回复的状态码* param replyText 回复的文本内容* param exchange 但是这个消息发给哪个交换机* param routingKey 当时这个消息使用哪个路由键*/Overridepublic void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {System.out.println(FailMessage: [ message ] replyCode: [ replyText ] exchange: [ exchange ] routingKey: [ routingKey ]);}
});我们在发送消息的一端故意写错路由键致使exchange投递消息失败。最后会看到回调方法ReturnCallback中打印的内容
FailMessage: [(Body:{id:2,name:test,sort:1,status:null,createTime:1641608721639} MessageProperties [headers{spring_returned_message_correlationb6b21f2d-73ad-473d-9639-feec76953c7b, __TypeId__com.atguigu.gulimall.order.entity.OrderReturnReasonEntity}, contentTypeapplication/json, contentEncodingUTF-8, contentLength0, receivedDeliveryModePERSISTENT, priority0, deliveryTag0])] replyCode: [NO_ROUTE] exchange: [hello-java-exchange] routingKey: [hello.news1]补充在发送消息的时候还可以指定一个CorrelationData类型的参数可以回顾上文的发送消息的方法这个CorrelationData类的构造器参数可以填一个UUID代表消息的唯一id在重写ConfirmCallback中的方法的第一个参数就是这个通过这个参数就可以获取消息的唯一id。
注意监听方法返回值必须为void否则控制台会不断打印报错信息。血的教训
可靠投递-消费端确认 ACK(Acknowledge)消息确认机制
消费者获取到消息成功处理可以回复Ack给Broker
basic.ack用于肯定确认broker可以删除此消息basic.nack用于否定确认可以指定broker是否丢弃此消息可以批量basic.reject用于否定确认同上但不能批量
在默认状况下ACK消息确认机制是当消息一旦抵达消费方客户端会自动确认服务端就会删除这个消息出队但是如果在消息消费过程中服务器宕机了这些消息也会被删除这就造成了消息丢失的问题。 问题 我们收到很多消息客户端会自动回复给服务器ack进行确认,但如果只有一个消息处理成功然后客户端宕机了。就会发生消息丢失 这时我们改为手动确认模式。只要我们没有明确告诉MQ,消息被签收了也就是没有ACK,消息就一直unacked状态 即使Consumer宕机。消息也不会丢失状态会重新变为Ready,下一次有新的Consumer连接进来就发给他
通过配置可以开启消息需要经过手动确认才能从队列中删除消息
#手动ack消息
spring.rabbitmq.listener.simple.acknowledge-modemanual修改我们的RabbitMQ的配置类MyRabbitConfig代码如下
RabbitHandler
public void receiveMessage2(Message message, OrderEntity content, Channel channel) {//消息体信息byte[] body message.getBody();// 消息头信息MessageProperties messageProperties message.getMessageProperties();long deliveryTag messageProperties.getDeliveryTag();//手动接收消息//long deliveryTag相当当前消息派发的标签从messageProperties中获取并且在Channel中自增的//boolean multiple 是否批量确认try {channel.basicAck(deliveryTag, false);} catch (Exception e) {e.printStackTrace();}System.out.println(receiveMessage2 接收消息: content);
}我们在上方的代码打上断点并观察RabbitMQ客户端的状况 对中总共有5条消息并且进入了Unacked即未被确认的状态。
但是这里使用debug模式启动然后关掉服务模拟服务器宕机会发生一个问题就是在关闭服务之前idea会将未执行完的方法先执行完再关闭服务。
所以可以在cmd杀掉进程模拟宕机。
这时由于打了断点没有走到消息确认的那一行代码随机服务器宕机所有没有确认的消息都会从Unacked的状态回调Ready的状态。
有接收消息的方法就有拒绝消息的方法basicNack和basicReject
//long deliveryTag 当前消息派发的标签
//boolean multiple 是否批量处理
//boolean requeue 拒绝后是否将消息重新入队
channel.basicNack(deliveryTag, false, true);
channel.basicReject(deliveryTag, true);basicNack和basicReject都可以用来拒绝消息但是basicNack比basicReject多了一个参数boolean multiple(是否批量处理)
如果将requeue设置为true被拒绝的消息就会重新入队等待消费false则拒绝消息就相当于丢弃此消息。
修改配置文件application.properties
#手动确认收货ack
spring.rabbitmq.listener.simple.acknowledge-modemanualRabbitHandler
public void receiverMessage(Message message,OrderReturnApplyEntity content,Channel channel) throws InterruptedException {//消息体byte[] body message.getBody();//消息头属性信息MessageProperties properties message.getMessageProperties();System.out.println(接收到消息...内容: content);//Thread.sleep(3000);System.out.println(消息处理完成》content.getReturnName());//channel内按顺序自增的long deliveryTag message.getMessageProperties().getDeliveryTag();System.out.println(deliveryTag:deliveryTag);//签收货物非批量模式try{if (deliveryTag % 2 0){//收货channel.basicAck(deliveryTag,false);System.out.println(签收了货物。。。deliveryTag);}else {//退货第二个参数是multiple批量处理第三个参数是丢弃的消息是否重新入队requeuefalse 丢弃 requeuetrue发回服务器服务器重新将消息入队。channel.basicNack(deliveryTag,false,true);System.out.println(没有签收货物...deliveryTag);}}catch (Exception e){//网络中断}
}如何签收信息 业务成功就应该签收channel.basicAck(deliveryTag,false); 业务处理失败就应该拒签让别人处理channel.basicNack(deliveryTag,false,true); 感谢耐心看到这里的同学觉得文章对您有帮助的话希望同学们不要吝啬您手中的赞动动您智慧的小手您的认可就是我创作的动力 之后还会勤更自己的学习笔记感兴趣的朋友点点关注哦。
