RabbitMQ

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

进入rabbitmqweb管理页面

SpringBoot整合RabbitMQ

在pom文件中导入AMQP依赖

      org.springframework.boot     spring-boot-starter-amqp  

模型：

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

• 数据隔离 ：用户只能操作自己的虚拟主机

• 交换机与队列有绑定关系

• AMQP一种消息通讯协议，该协议与语言和平台无关

• 同一条消息只被一个消费者处理（Fanout交换机为发送多条相同内容的消息到不同绑定队列中）

• AMQP提供三个类 :

  • Queue ： 用于声明队列，可由工厂类QueueBuilder构建
  • Exchange：用于声明交换机，可由工厂类ExcahngeBuilder构建
  • Binding：用于声明队列和交换机的绑定关系，可由工厂类BindingBuilder构建

Fanout

@Configuration public class FanoutConfiguration {      @Bean     public FanoutExchange fanoutExchange(){ 		//创建fanout交换机         return new FanoutExchange("ChenJJ.fanout2");     }           @Bean     public Queue fanoutQueue3(){         return new Queue("fanout.queue3");//创建队列     }      @Bean     public Binding fanoutBinding3(){ 		//队列绑定与fanout交换机         return BindingBuilder.bind(fanoutQueue3()).to(fanoutExchange());     } } 

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

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

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

Direct

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

@Configuration public class DirectConfiguration {      @Bean     public DirectExchange directExchange(){ 		//创建Direct交换机         return new DirectExchange("ChenJJ.direct2");     }      @Bean     public Queue directQueue3(){         return new Queue("direct.queue3");//创建队列     }      @Bean     public Binding directBinding3Red(){         return	BindingBuilder.bind(directQueue3()).to(directExchange()).with("red");     }//绑定队列和交换机并设置routingkey为“red”     @Bean     public Binding directBinding3Blue(){          return BindingBuilder.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"} )) public  void directQueueTest3(String msg) {     System.out.println("消费者3接收到消息" + msg); } 

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

void testSend2Direct1(){     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 都符合要求

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

消息转换器

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

@Test void testSendObject(){     Map msg = new HashMap<>(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依赖：

      com.fasterxml.jackson.dataformat     jackson-dataformat-xml  

配置MessageConverter

@Bean public MessageConverter jacksonMessageConverter() {     return new Jackson2JsonMessageConverter(); } 

此时再发送消息 队列中接受到消息为 {“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 @Configuration public class MqConfirmConfig implements ApplicationContextAware {      @Override     public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {         RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);         //配置回调         rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {             @Override             public void returnedMessage(ReturnedMessage returnedMessage) {                 log.error("收到消息的return callback，exchange:{},key:{},msg；{}，code:{},replyText:{}",                         returnedMessage.getMessage(),                         returnedMessage.getRoutingKey(),                         returnedMessage.getMessage(),                         returnedMessage.getReplyCode(),                         returnedMessage.getReplyText());             }         });         {          };     } } 

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

@Test void testConfirmCallBack() throws InterruptedException {     //创建CorrelationData     CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());     //给Future添加ConfirmCallBack     correlationData.getFuture().addCallback(new ListenableFutureCallback() {         @Override         public void onFailure(Throwable ex) {             //Future发生异常时的处理逻辑，基本不会触发             log.error("消息回调失败",ex);         }          @Override         public void onSuccess(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：数据持久化（旧解决方法）

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

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

@Test void testPageOut(){     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")     ))     public void listenLazyQueue(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") public class ErrorConfiguration {      @Bean     public DirectExchange errorExchanged(){         return new DirectExchange("error.direct");//创建接受失败消息的交换机     }      @Bean     public Queue errorQueue(){         return new Queue("error.queue");//创建接受失败消息的队列     }      @Bean     public Binding errorBinding(Queue errorQueue,DirectExchange errorExchanged){         return BindingBuilder.bind(errorQueue).to(errorExchanged).with("error");//绑定     }      @Bean     public MessageRecoverer messageRecoverer(RabbitTemplate rabbitTemplate){         //消息恢复器         return new RepublishMessageRecoverer(rabbitTemplate,"error.direct","error");     } } 

@ConditionalOnProperty(prefix = "spring.rabbitmq.listener.simple.retry",name = "enabled",havingValue = "true") 该注解表示查询配置类中 该属性是否符合要求，若符合则该配置类生效,此处即开启重试机制 

业务的幂等性

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

解决方案：

唯一消息ID

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

@Bean public MessageConverter jacksonMessageConverter() {     Jackson2JsonMessageConverter jjmc = new Jackson2JsonMessageConverter();//配置消息转换器     jjmc.setCreateMessageIds(true);//配置自动创建消息ID     return 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的插件安装，先到这里了）