0
0

学习笔记 ： RabbitMQ基础

RabbitMQ

浏览器输入 : http://localhost:15672/

进入rabbitmqweb管理页面

SpringBoot整合RabbitMQ

在pom文件中导入AMQP依赖

<!--AMQP--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency>

模型：

生产者 ——> 交换机（可以跳过交换机） ——>队列（一个或多个）——>消费者

  • 数据隔离 ：用户只能操作自己的虚拟主机
  • 交换机与队列有绑定关系
  • AMQP一种消息通讯协议，该协议与语言和平台无关
  • 同一条消息只被一个消费者处理（Fanout交换机为发送多条相同内容的消息到不同绑定队列中）
  • AMQP提供三个类 :- Queue ： 用于声明队列，可由工厂类QueueBuilder构建- Exchange：用于声明交换机，可由工厂类ExcahngeBuilder构建- Binding：用于声明队列和交换机的绑定关系，可由工厂类BindingBuilder构建

Fanout

@ConfigurationpublicclassFanoutConfiguration{@BeanpublicFanoutExchangefanoutExchange(){//创建fanout交换机returnnewFanoutExchange("ChenJJ.fanout2");}@BeanpublicQueuefanoutQueue3(){returnnewQueue("fanout.queue3");//创建队列}@BeanpublicBindingfanoutBinding3(){//队列绑定与fanout交换机returnBindingBuilder.bind(fanoutQueue3()).to(fanoutExchange());}}

fanout交换机会将交换机中消息发送给所有绑定的队列，所有绑定的队列都会收到消息。

@TestvoidtestSendMessage1(){String queueName ="fanout.queue3";//指定队列名String message ="hello，queue";//编写消息体
    rabbitTemplate.convertAndSend(queueName, message);}

@RabbitListener(queues ="fanout.queue2")//指定监听该队列publicvoidfanoutQueueTest2(String msg){System.out.println("消费者接收到消息"+msg);}

Direct

俩种java中创建队列方式 都可。

@ConfigurationpublicclassDirectConfiguration{@BeanpublicDirectExchangedirectExchange(){//创建Direct交换机returnnewDirectExchange("ChenJJ.direct2");}@BeanpublicQueuedirectQueue3(){returnnewQueue("direct.queue3");//创建队列}@BeanpublicBindingdirectBinding3Red(){returnBindingBuilder.bind(directQueue3()).to(directExchange()).with("red");}//绑定队列和交换机并设置routingkey为“red”@BeanpublicBindingdirectBinding3Blue(){returnBindingBuilder.bind(directQueue3()).to(directExchange()).with("blue");}//绑定队列和交换机并设置routingkey为“blue”}

@RabbitListener(bindings =@QueueBinding(
        value =@Queue(name ="direct.queue5",durable ="true"),
        exchange =@Exchange(name ="ChenJJ.direct3",type =ExchangeTypes.DIRECT),
        key ={"red","blue"}))publicvoiddirectQueueTest3(String msg){System.out.println("消费者3接收到消息"+ msg);}

Dire交换机在和队列绑定中，需要绑定routingkey。发送消息时携带routingkey，在交换机识别key后发送到对应的队列中，若key不存在则发送失败。

voidtestSend2Direct1(){String exchangeName ="ChenJJ.direct";String msg ="红色是毁灭";
    rabbitTemplate.convertAndSend(exchangeName,"red", msg);}

Topic

Topic交换机机制和Direct交换机机制类似，绑定队列时需绑定Bindingkey，区别是routingkey可以是多个单词的列表且以 . 分割

  • #: 发表0个或多个单词
  • *：代指一个单词

例：queue1的bandingkey为china.#，则消息携带key中有”china“时接收。 如china.news , china.weather ，china 。

​ queue2的bandingkey为#.weather,同理， japan.weather , china.weather , country.china.weather 都符合要求

@TestvoidtestSend2Topic1(){String exchangeName ="ChenJJ.topic";//该队列绑定了key：yg.#和#.newsString msg ="yg的黑色蕾丝内裤到货";
    rabbitTemplate.convertAndSend(exchangeName,"yg.news", msg);}@TestvoidtestSend2Topic2(){String exchangeName ="ChenJJ.topic";String msg ="紧急通知：yg的黑色蕾丝内裤已售罄";
    rabbitTemplate.convertAndSend(exchangeName,".news", msg);}

消息转换器

消息传输时，会自动传化为字节传输。若传输消息主题不为String类型时，

@TestvoidtestSendObject(){Map<String,String> msg =newHashMap<>(2);
    msg.put("name","yg");
    msg.put("age","18");
    rabbitTemplate.convertAndSend("object.queue",msg);}

传输结果将是一串乱码。因为Spring的消息对象处理是基于JDK的ObjectOutPutStream完成序列化。

  • JDK的序列化有安全风险（反序列）
  • JDK序列化的消息太大 （如刚刚的消息 足足有180字节）
  • JDK序列化的消息可读性差

消息转换器，即用JSON序列化带起默认的JDK序列化

在pom文件中引入jackson依赖：

<!--Jackson--><dependency><groupId>com.fasterxml.jackson.dataformat</groupId><artifactId>jackson-dataformat-xml</artifactId></dependency>

配置MessageConverter

@BeanpublicMessageConverterjacksonMessageConverter(){returnnewJackson2JsonMessageConverter();}

此时再发送消息 队列中接受到消息为 {“name”:“yg”,“age”:"18}，仅仅占了20多字节。

发送者的可靠性

发送者的可靠性保证 生产者重连

由于网络波动，可能出现客户端连接MQ丢失的情况，可通过在yml配置文件配置连接失败后的重试机制。

spring:
  rabbitmq:
    host: localhost
    port: 5672                  #MQ端口
    username: ChenJJ            #MQ账户
    password: 123456
    virtual-host: /ChenJJ
    
    #消费者
    listener:
      simple:
        prefetch: 1             #一次接收一个消息，处理完成后接受下一个消息
        retry:
          max-attempts: 3       #最大重试次数
          
    #生产者      
    template:
      retry:
        enabled: true           #开启连接重试
        multiplier: 1           #下次重试的间隔时长倍数
        max-attempts: 3         #最大重试次数
    connection-timeout: 1s      #设置MQ的连接超时时间

生产者的可靠性保证 生产者确认

RabbitMQ有Publisher Confirm 和Publishs Return俩种确认机制，开启确认机制后，在MQ成功接收到消息后会返回确认消息给生产者。有以下几种情况：

  • 消息投递到了MQ，但是路由失败。（routingkey错误）会通过PublishReturn返回路由异常原因，然后返回ACK，告知投递成功。
  • 临时投递到了MQ，成功入队，返回ACK，告知投递成功。
  • 持久消息投递到了MQ，成功入队并完成持久化，返回ACK，告知投递成功。（入队成功但是持久化失败不会返回ACK)
  • 其他情况都返回NACK，告知投递失败

在yml文件中添加配置

spring:
  rabbitmq:
    publisher-confirm-type: correlated    #开启publisher confirm机制 
                                #消息确认机制会使效率变低
                                #none：关闭confirm机制，
                                #simple：同步阻塞等待MQ的回执消息
                                #correlated：MQ异步回趟方式返回回执消息
    publisher-returns: true               #开启publisher return机制 会降低性能

每个RabbitTemplate只能配置一个ReturnCallback，因此需要再项目启动中配置

@Slf4j@ConfigurationpublicclassMqConfirmConfigimplementsApplicationContextAware{@OverridepublicvoidsetApplicationContext(ApplicationContext applicationContext)throwsBeansException{RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);//配置回调
        rabbitTemplate.setReturnsCallback(newRabbitTemplate.ReturnsCallback(){@OverridepublicvoidreturnedMessage(ReturnedMessage returnedMessage){
                log.error("收到消息的return callback，exchange:{},key:{},msg；{}，code:{},replyText:{}",
                        returnedMessage.getMessage(),
                        returnedMessage.getRoutingKey(),
                        returnedMessage.getMessage(),
                        returnedMessage.getReplyCode(),
                        returnedMessage.getReplyText());}});{};}}

发送消息，指定消息ID，消息ConfirmCallBack

@TestvoidtestConfirmCallBack()throwsInterruptedException{//创建CorrelationDataCorrelationData correlationData =newCorrelationData(UUID.randomUUID().toString());//给Future添加ConfirmCallBack
    correlationData.getFuture().addCallback(newListenableFutureCallback<CorrelationData.Confirm>(){@OverridepublicvoidonFailure(Throwable ex){//Future发生异常时的处理逻辑，基本不会触发
            log.error("消息回调失败",ex);}@OverridepublicvoidonSuccess(CorrelationData.Confirm result){//result为回执内容
            log.debug("收到消息confirm callback回执");if(result.isAck()){
                log.debug("消息发送成功,收到ack");}else{
                log.error("消息发送失败，收到nack，原因：{}",result.getReason());}}});
    rabbitTemplate.convertAndSend("ChenJJ.direct","red","红色是毁灭",correlationData);Thread.sleep(2000);}

注意：生产者消费机制性能极低，尽可能不用

MQ的可靠性

默认情况下，RabbitMQ中收到的消息保存在内存中以较低消息收发的延迟。会导致俩个问题：

  • 一旦MQ宕机，内存中消息会丢失
  • 内存空间有限当消费者故障或处理信息过慢时，会导致消息挤压，引发MQ阻塞

解决方法1：数据持久化（旧解决方法）

  • 交换机持久化
  • 队列持久化
  • 消息持久化

此方法不推荐使用，仅仅演示发送持久化消息:

@TestvoidtestPageOut(){Message msg =MessageBuilder.withBody("yg的黑色蕾丝内裤登上热销榜一".getBytes(StandardCharsets.UTF_8))//转字节.setDeliveryMode(MessageDeliveryMode.PERSISTENT).build();// 持久化，存入磁盘 此处理方式不会发生MQ阻塞问题，但是性能不好//先存入内存 再存入磁盘
    rabbitTemplate.convertAndSend("object.queue",msg);}

解决方法2：LazyQueue（推荐)

//LazyQueue 消息直接存入磁盘而非内存，（内存中保留最近的2048条消息）//消费者要消费消息时才会从磁盘读取并加载到内存 支持数百万条消息的存储 3.12版本后，所有消息都是lazy       queue模式，无法更改//开启持久化和生产者确认时，MQ只有在消息持久化完成后才给生产者ACK回执@RabbitListener(queuesToDeclare =@Queue(
            name ="lazy.queue",
            durable ="true",
            arguments =@Argument(name ="x-queue-mode",value ="lazy")))publicvoidlistenLazyQueue(String msg){
        log.info("接受到lazyqueue的消息：{}",msg);}}

直接选择队列为lazyqueue属性.

消费者的可靠性

消费者确认机制

为确认消费者是否成功处理消息，RabbitMQ提供了消费者确认机制（Consumer Ackowledgement）

消费者处理消息结束后，应该向RabbitMQ发送一个回执，告知RabbitMQ自己消息处理状态。

回执有三种可选值：

  • ACK : 成功处理消息，RabbitMQ从队列中删除该消息
  • nack ： 消息处理失败，RabbitMQ需要再次投递消息
  • reject ： 消息处理失败并拒绝该消息，RabbitMQ从队列中删除该消息

SpringAMQP已经实现了消息确认功能。并允许我们通过配置文件选择ACK处理方式，有三种方式:

  • none ： 不处理。即消息投递给消费者后立刻ACK，消息立刻从MQ删除。很不安全
  • manual : 手动模式。需要自己在业务代码调用api，发送ack或者reject，存在业务入侵，但是更灵活
  • auto ： 自动模式，SpringAMQP利用AOP对我们的消息处理逻辑做了环绕增强，当业务正常执行时则自动返回ack。出现异常就根据异常判断返回不同结果: - 业务异常，返回nack- 消息处理或检验异常，返回reject
spring:
  rabbitmq:
    #消费者
    listener:
      simple:
        prefetch: 1             #一次接收一个消息，处理完成后接受下一个消息
        retry:
          max-attempts: 3           #最大重试次数
          enabled: true             #开启消费者重试机制
          initial-interval: 1000ms  #重试失败等待时长时间 1秒
          stateless: true           #true无状态，false有状态，若业务中包含事务，改为false
        acknowledge-mode: auto  #消息确认机制

在yml文件中修改配置

失败消息处理策略

开启重试模式后，重试次数耗尽，如果消息依然失败则要有MessageRecoverer接口来处理，包含三种不同实现:

  • RejectAndDontRequeueRecoverer ： 重试耗尽后，直接reject丢弃消息（默认模式）尽量不选
  • ImmediateRequeueMessageRcoverer ： 重试耗尽后，返回nack，消息重新入队
  • RepublishMessageRecoverer : 重试耗尽后，将失败消息投递到指定的交换机（error.direct -> error.queue)

RepublishMessageRecoverer作为失败处理策略:

  • 首先，定义接受失败消息的交换机，队列，及其绑定关系
  • 定义RepublishMessageRecoverer:@Configuration@ConditionalOnProperty(prefix ="spring.rabbitmq.listener.simple.retry",name ="enabled",havingValue ="true")publicclassErrorConfiguration{@BeanpublicDirectExchangeerrorExchanged(){returnnewDirectExchange("error.direct");//创建接受失败消息的交换机}@BeanpublicQueueerrorQueue(){returnnewQueue("error.queue");//创建接受失败消息的队列}@BeanpublicBindingerrorBinding(Queue errorQueue,DirectExchange errorExchanged){returnBindingBuilder.bind(errorQueue).to(errorExchanged).with("error");//绑定}@BeanpublicMessageRecoverermessageRecoverer(RabbitTemplate rabbitTemplate){//消息恢复器returnnewRepublishMessageRecoverer(rabbitTemplate,"error.direct","error");}}
@ConditionalOnProperty(prefix = "spring.rabbitmq.listener.simple.retry",name = "enabled",havingValue = "true")
该注解表示查询配置类中 该属性是否符合要求，若符合则该配置类生效,此处即开启重试机制

业务的幂等性

同一个业务，执行一次或多次队业务状态的影响是一致的。如 ： 查询业务。非幂等如下单业务，会扣减库存,可通过令牌机制解决重复下单的问题，使业务幂等。

解决方案：

唯一消息ID

给每一个消息设置一个唯一ID，与消息一起投递给消费者，处理完成保存ID到数据库若下次又收到相同的消息，去数据库查询是否存在，存在即重复消息放弃处理。

@BeanpublicMessageConverterjacksonMessageConverter(){Jackson2JsonMessageConverter jjmc =newJackson2JsonMessageConverter();//配置消息转换器
    jjmc.setCreateMessageIds(true);//配置自动创建消息IDreturn jjmc;}

这样可以在发送消息时为消息生成ID，缺点是消耗性能，但是方便，通用。

业务判断

结合业务逻辑，基于业务本身判断。如：在支付服务中，查询订单支付状态，处理订单时判断订单状态。类似乐观锁。

延迟消息

生产者发送消息时指定一个时间，消费者不会立刻收到消息，而是在指定时间之后才收到消息。

死信交换机（DLX）

当一个队列中的消息满足下列情况之一时候，就会成为死信。

  • 消费者回执为reject或nack声明消费失败，并且消息的requeue参数设置为false
  • 消息是一个过期的消息，超时无人消费
  • 要投递的队列满了，最早的消息可能成为死信

例 ： 指定simple.direct交换机绑定一个simple.queue队列，但是该队列未绑定消费者。

​ simple.queue队列绑定dlx.direct死信交换机。

​ 再指定dlx.direct和dlx.queue绑定死信消息队列，队列绑定消费者。

发送消息 TTL=30s -> simple.direct -> simple.queue(无消费者监听消费，30秒后进入死信交换机) -> dlx.direct -> dlx.queue -> consumer

​ 比较繁琐，需要定义较多交换机和队列。

延迟消息插件

该插件设计了一种支持延迟消息功能的交换机，当消息投递到交换机后可以暂存一定时间，到期后再投递到队列。
（暂时不会搞Windows版下RaibbtMQ的插件安装，先到这里了）

标签： java intellij-idea rabbitmq
本文转载自: https://blog.csdn.net/OldOrange114514/article/details/140028730
版权归原作者 OldOrange114514 所有， 如有侵权，请联系我们删除。
登录后发布评论

“学习笔记 ： RabbitMQ基础”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

计算机网络基础全攻略：探秘网络构建块（1/10）

JT808-2019协议测试工具：高效开发与测试的利器

Appium: 手机应用自动化测试(一)

【选题指导】大数据专业毕业设计选题分享 实用版 2025

flink+kafka 如何保证精准一次

RabbitMQ常用管理命令及管理后台

【Rabbitmq篇】高级特性----TTL，死信队列，延迟队列

文章导航