weixin_40725706

这个屌丝很懒，什么也没留下！

热门标签

深入理解RabbitMQ高级特性_channel.basicack

作者：weixin_40725706 | 2024-02-21 22:33:53

踩

channel.basicack

前言

在微服务的架构下，对于服务调用产生的分布式事务问题，比较主流的解决方案有：基于XA协议的两阶段提交协议（2PC）、事务补偿、消息队列实现最终一致性（柔一致性）及阿里的GTS分布式事务中间件。本篇文章主要采用最终一致性解决方案RabbitMQ消息中间件结合实例来深入解读RabbitMQ的几个高级特性

RabbitMQ介绍

RabbitMQ 即一个消息队列遵循 AMQP（高级消息队列协议）服务器端用Erlang语言编写，主要是用来实现应用程序的异步和解耦，同时也能起到消息缓冲，消息分发的作用，消息的发送者无需知道消息使用者的存在，反之亦然。AMQP 的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全>>>官网传送门

RabbitMQ安装

RabbitMQ安装及配置

erlang下载
链接：https://pan.baidu.com/s/1EKoKLnT_0zTEWhc5JmICLg
提取码：xk1e
可以通过访问http://localhost:15672进行测试，默认的登陆账号为：guest，密码为：guest

管理界面可谓非常友好，在上面我们可以创建用户、分配权限、创建交换机、创建队列等、还有查看队列消息，消费效率，推送效率等

RabbitMQ基本概念

RabbitMQ

生产者(Producer) > 交换器(Exchange) > 队列(Queue) > 消费者(Consumer)

Message

消息，消息是不具名的，它由消息头和消息体组成。消息体是不透明的，而消息头则由一系列的可选属性组成，这些属性包括 routing-key（路由键）、priority（相对于其他消息的优先权）、delivery-mode（指出该消息可能需要持久性存储）等。

Publisher

消息的生产者，也是一个向交换器发布消息的客户端应用程序。

Exchange

交换器，用来接收生产者发送的消息并将这些消息路由给服务器中的队列。

Binding

绑定，用于消息队列和交换器之间的关联。一个绑定就是基于路由键将交换器和消息队列连接起来的路由规则，所以可以将交换器理解成一个由绑定构成的路由表。

Queue

消息队列，用来保存消息直到发送给消费者。它是消息的容器，也是消息的终点。一个消息可投入一个或多个队列。消息一直在队列里面，等待消费者连接到这个队列将其取走。

Connection

网络连接，比如一个TCP连接。

Channel

信道，多路复用连接中的一条独立的双向数据流通道。信道是建立在真实的TCP连接内地虚拟连接，AMQP命令都是通过信道发出去的，不管是发布消息、订阅队列还是接收消息，这些动作都是通过信道完成。因为对于操作系统来说建立和销毁 TCP 都是非常昂贵的开销，所以引入了信道的概念，以复用一条 TCP 连接。

Consumer

消息的消费者，表示一个从消息队列中取得消息的客户端应用程序。

Virtual Host

虚拟主机，表示一批交换器、消息队列和相关对象。虚拟主机是共享相同的身份认证和加密环境的独立服务器域。每个 vhost本质上就是一个 mini 版的 RabbitMQ 服务器，拥有自己的队列、交换器、绑定和权限机制。vhost 是 AMQP 概念的基础，必须在连接时指定，RabbitMQ 默认的 vhost 是 / 。

Broker

表示消息队列服务器实体。

RabbitMQ三种模式

Direct

消息中的路由键（routing key）如果和 Binding 中的 binding key 一致，交换器就将消息发到对应的队列中。路由键与队列名完全匹配，如果一个队列绑定到交换机要求路由键为“dog”，则只转发 routing key 标记为“dog”的消息，不会转发“dog.puppy”，也不会转发“dog.guard”等等。它是完全匹配、单播的模式。

Fanout

每个发到 fanout 类型交换器的消息都会分到所有绑定的队列上去。fanout 交换器不处理路由键，只是简单的将队列绑定到交换器上，每个发送到交换器的消息都会被转发到与该交换器绑定的所有队列上。很像子网广播，每台子网内的主机都获得了一份复制的消息。fanout 类型转发消息是最快的。

Topic

topic 交换器通过模式匹配分配消息的路由键属性，将路由键和某个模式进行匹配，此时队列需要绑定到一个模式上。它将路由键和绑定键的字符串切分成单词，这些单词之间用点隔开。它同样也会识别两个通配符：符号“#”和符号“”。#匹配0个或多个单词，匹配不多不少一个单词。

headers

不常用，和direct功能接近，不讨论。

上面只是简单的介绍了一下RabbitMQ的三种模式，接下来结合代码实例来看看

SpringBoot整合RabbitMQ

pom文件中导入amqp包


<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
    <version>1.5.8.RELEASE</version>
</dependency>

application.properties中配置如下信息这样


#rabbitmq
spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.publisher-confirms=true
spring.rabbitmq.publisher-returns=true

这样RabbitMQ就引入成功了，接下来以微服务下商品秒杀场景为例，结合Direct模式来深入了解RabbitMQ的高级特性

创建一个RabbitMQConfig类，Direct就是一对一模式，从上面可以知道，RabbitMQ有发送者，交换机，队列，接收者。Direct就是一个发送者对应一个接收者。如果有多个，只会有一个接收到消息


package com.giantfind.common.mq;
 
import org.springframework.amqp.core.*;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
import java.util.HashMap;
import java.util.Map;
 
/**
 * @Package: com.giantfind.common.mq
 * @ClassName: RabbitMQConfig
 * @Author: liuyaolong
 * @Description: 配置rabbitmq
 * @Version: 1.0
 */
@Configuration
public class RabbitMQConfig {
 
    /**业务交换机*/
    public static final String BUSINESS_EXCHANGE = "business.exchange";
 
    /**常规超时时间*/
    public static Long QUEUE_EXPIRATION = 20000L;
 
    /**生成订单队列*/
    public static final String ORDER_CREATE_QUEUE = "order.create.queue";
 
    /**生成订单路由键*/
    public static final String ORDER_CREATE_ROUTING_KEY = "order.create.routing.key";
 
    @Bean
    public MessageConverter messageConverter(){
        return new Jackson2JsonMessageConverter();
    }
 
    /**业务交换机*/
    @Bean
    public Exchange getBusinessExchange(){
        return ExchangeBuilder.directExchange(BUSINESS_EXCHANGE).durable(true).build();
    }
 
    /**生成订单队列*/
    @Bean
    public Queue getOrderQueue(){
        return new Queue(ORDER_CREATE_QUEUE);
    }
 
    /**绑定业务交换机和生成订单队列*/
    @Bean
    public Binding bindOrder(){
        return BindingBuilder.bind(getOrderQueue()).to(getBusinessExchange()).with(ORDER_CREATE_ROUTING_KEY).noargs();
    }
 
}

模拟秒杀下单接口（消息生产者），因为并发性很高所以创建订单信息放入消息队列里解耦，异步下单，常见的抢票软件等等都是类似功能


  
    @Autowired
    private RabbitTemplate rabbitTemplate;
    /**
     * 商品秒杀
     * @param orderInfo
     * @return
     */
    @PostMapping("/seckill")
    public String seckill(@RequestBody RequestMsg<OrderInfoIn> orderInfo) throws CommonRuntimeException {
        //验证库存
        //减库存（乐观锁机制）
        //创建订单消息放入创建订单队列...
        rabbitTemplate.convertAndSend(RabbitMQConfig.BUSINESS_EXCHANGE,RabbitMQConfig.ORDER_CREATE_ROUTING_KEY,orderInfoIn);
        return null;
 
}

订单服务进行队列监听（消费者）


 /**
     * 生成订单消息监听
     * @param orderInfoIn
     * @param message
     * @param channel
     * @throws CommonRuntimeException
     */
    @RabbitListener(queues = RabbitMQConfig.ORDER_CREATE_QUEUE)
    public void createOrderListener(OrderInfoIn orderInfoIn,Message message, Channel channel) throws CommonRuntimeException, IOException {
            try {
                //创建订单...
            } catch (Exception e){
                logger.error("消费创建订单消息失败【】error："+ message.getBody());
                logger.error("OrderConsumer  handleMessage {} , error:",message,e);
                
            }
    }

这样一个简单的创建订单的消息通过@RabbitListener（queues=""）监听即可进行消费。这种做法会有什么问题呢？大家可以思考下几个问题：队列是如何知道该消息是否消费掉？如果没有成功消费掉如何重试？假如消费端一直消费失败会不会导致重复消费死循环？如何保证消费消息冥等？有没有对应的补偿机制呢？那么接下来就是本文的重点：RabbitMQ的高级特性介绍

ACK应答机制配置

手动确认消息是否成功消费


#开启ACK 手动确认
spring.rabbitmq.listener.direct.acknowledge-mode=manual
spring.rabbitmq.listener.simple.acknowledge-mode=manual
spring.rabbitmq.listener.simple.default-requeue-rejected=false

如果消费端消费失败我们通常结合场景会重新放入队列，代码实例


     /**
     * 生成订单消息监听
     * @param orderInfoIn
     * @param message
     * @param channel
     * @throws CommonRuntimeException
     */
    @RabbitListener(queues = RabbitMQConfig.ORDER_CREATE_QUEUE)
    public void createOrderListener(OrderInfoIn orderInfoIn,Message message, Channel channel) throws CommonRuntimeException, IOException {
            try {
                //创建订单...
                //成功消费
                channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
            } catch (Exception e){
                logger.error("消费创建订单消息失败【】error："+ message.getBody());
                logger.error("OrderConsumer  handleMessage {} , error:",message,e);
                //处理消息失败，将消息重新放回队列
                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false,true);
            }
    }

重试机制配置

(如果消费端一直消费失败，我们可采用重试机制来限制重复消费，从而避免消费死循环)


#开启消费者重试
spring.rabbitmq.listener.simple.retry.enabled=true
#最大重试次数
spring.rabbitmq.listener.simple.retry.max-attempts=5
#重试间隔时间（单位毫秒）
spring.rabbitmq.listener.simple.retry.initial-interval=3000
#重试最大时间间隔（单位毫秒）
spring.rabbitmq.listener.simple.retry.max-interval=3000
#应用于前一重试间隔的乘法器
spring.rabbitmq.listener.simple.retry.multiplier=5

TTL队列/消息

是 Time To Live 的缩写，也就是生存时间。1、RabbitMQ支持消息的过期时间设置，在消息发送时指定。2、RabbitMQ支持队列的过期时间设置，从消息入队时开始计算，只要超过队列的超时时间配置，消息自动清除
   /**生成订单队列*/
    @Bean
    public Queue getOrderQueue(){
        Map<String, Object> args = new HashMap<>();
        //设置过期时间
        args.put("x-message-ttl", QUEUE_EXPIRATION);
        return QueueBuilder.durable(ORDER_CREATE_QUEUE).withArguments(args).build();
    }

死信队列DLX

全称为 Dead-Letter-Exchange，可以称之为死信交换器，也有人称之为死信邮箱。当消息在一个队列中变成死信 (dead message) 之后，它能被重新被发送到另一个交换器中，这个交换器就是 DLX，绑定 DLX 的队列就称之为死信队列。消息被拒绝、消息过期、无法入队，该何去何从，死信队列这里来，这里可结合场景做数据落库，或者通过管理后台进行人工补偿机制


package com.giantfind.common.mq;
 
import org.springframework.amqp.core.*;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
import java.util.HashMap;
import java.util.Map;
 
/**
 * @Package: com.giantfind.common.mq
 * @ClassName: RabbitMQConfig
 * @Author: liuyaolong
 * @Description: 配置rabbitmq
 * @Version: 1.0
 */
@Configuration
public class RabbitMQConfig {
 
    /**业务交换机*/
    public static final String BUSINESS_EXCHANGE = "business.exchange";
    /**死信交换机*/
    public static final String DEAD_LETTER_EXCHANGE = "dead.letter.exchange";
 
    /**常规超时时间*/
    public static Long QUEUE_EXPIRATION = 20000L;
 
    /**生成订单队列*/
    public static final String ORDER_CREATE_QUEUE = "order.create.queue";
 
    /**生成订单死信队列*/
    public static final String ORDER_CREATE_DEAD_LETTER_QUEUE = "order.create.dead.letter.queue";
 
    /**生成订单路由键*/
    public static final String ORDER_CREATE_ROUTING_KEY = "order.create.routing.key";
 
    /**生成订单死信路由键*/
    public static final String ORDER_CREATE_DEAD_LETTER_ROUTING_KEY = "order.create.dead.letter.routing.key";
 
    @Bean
    public MessageConverter messageConverter(){
        return new Jackson2JsonMessageConverter();
    }
 
    /**业务交换机*/
    @Bean
    public Exchange getBusinessExchange(){
        return ExchangeBuilder.directExchange(BUSINESS_EXCHANGE).durable(true).build();
    }
 
    /**死信交换机*/
    @Bean
    public Exchange getDeadLetterExchange(){return ExchangeBuilder.directExchange(DEAD_LETTER_EXCHANGE).durable(true).build();}
 
    /**生成订单队列*/
    @Bean
    public Queue getOrderQueue(){
        Map<String, Object> args = new HashMap<>();
        //x-dead-letter-exchange 声明当前队列绑定的死信交换机
        args.put("x-dead-letter-exchange", DEAD_LETTER_EXCHANGE);
        //x-dead-letter-routing-key 声明当前队列的死信路由key
        args.put("x-dead-letter-routing-key", ORDER_CREATE_DEAD_LETTER_ROUTING_KEY);
        //设置过期时间
        args.put("x-message-ttl", QUEUE_EXPIRATION);
        return QueueBuilder.durable(ORDER_CREATE_QUEUE).withArguments(args).build();
    }
    /**绑定业务交换机和生成订单队列*/
    @Bean
    public Binding bindOrder(){
        return BindingBuilder.bind(getOrderQueue()).to(getBusinessExchange()).with(ORDER_CREATE_ROUTING_KEY).noargs();
    }
 
    /**生成订单死信队列*/
    @Bean
    public Queue getOrderDeadLetterQueue(){return new Queue(ORDER_CREATE_DEAD_LETTER_QUEUE);}
    /**绑定死信交换机和生成订单死信队列*/
    @Bean
    public Binding bingOrderDeadLetter(){
        return BindingBuilder.bind(getOrderDeadLetterQueue()).to(getDeadLetterExchange()).with(ORDER_CREATE_DEAD_LETTER_ROUTING_KEY).noargs();
    }
 
}

如何保证只成功消费一次（冥等性）

场景：消费端成功消费信息，但是在ack时发生网络抖动等原因，导致消息已被消费掉，然而还存在于队列当中。主流解决方案：唯一ID+指纹码机制或利用Redis的原子性去实现


 
    /**
     * 生成订单消息监听
     * @param orderInfoIn
     * @param message
     * @param channel
     * @throws CommonRuntimeException
     */
    @RabbitListener(queues = RabbitMQConfig.ORDER_CREATE_QUEUE)
    public void createOrderListener(OrderInfoIn orderInfoIn,Message message, Channel channel) throws CommonRuntimeException, IOException {
        //消息生产端加入雪花算法
        //消费端消费前先验证该消息是否被消费
        String messageId = (String)redisTemplate.opsForValue().get(orderInfoIn.getMessageId().toString());
        if(messageId == null){//保证冥等性
            try {
                //创建订单
                redisTemplate.opsForValue().set(orderInfoIn.getMessageId().toString(), "ack");
                //成功消费
                channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
 
            } catch (Exception e){
                logger.error("消费创建订单消息失败【】error："+ message.getBody());
                logger.error("OrderConsumer  handleMessage {} , error:",message,e);
                //处理消息失败，将消息重新放回队列
                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false,true);
            }
        }else{
            //已消费
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
        }
 
    }

延迟队列

延迟队列指的就是可以在固定时间长度之后才可以被消费到，比如在订单系统中，一个用户下单之后通常有 30 分钟的时间进行支付，如果 30 分钟之内没有支付成功，那么这个订单将进行异常处理，这时就可以使用延迟队列来处理这些订单，延迟队列的实现就是通过TTL + DLX来实现 å¨è¿éæå¥å¾çæè¿°


package com.giantfind.common.mq;
 
import org.springframework.amqp.core.*;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
import java.util.HashMap;
import java.util.Map;
 
/**
 * @Package: com.giantfind.common.mq
 * @ClassName: RabbitMQConfig
 * @Author: liuyaolong
 * @Description: 配置rabbitmq
 * @Version: 1.0
 */
@Configuration
public class RabbitMQConfig {
 
    /**业务交换机*/
    public static final String BUSINESS_EXCHANGE = "business.exchange";
    /**死信交换机*/
    public static final String DEAD_LETTER_EXCHANGE = "dead.letter.exchange";
    /**常规超时时间*/
    public static Long QUEUE_EXPIRATION = 20000L;
    /**支付超时时间*/
    public static Long QUEUE_PAID_EXPIRATION = 7200000L;//2小时
 
 
    /**订单待支付队列*/
    public static final String UNPAID_QUEUE = "unpaid.queue";
 
    /**订单待支付死信队列*/
    public static final String UNPAID_DEAD_LETTER_QUEUE = "unpaid.dead.letter.queue";
 
    /**订单待支付路由键*/
    public static final String UNPAID_ROUTING_KEY = "unpaid.routing.key";
 
    /**订单待支付死信路由键*/
    public static final String UNPAID_DEAD_LETTER_ROUTING_KEY = "unpaid.dead.letter.routing.key";
 
 
    @Bean
    public MessageConverter messageConverter(){
        return new Jackson2JsonMessageConverter();
    }
 
    /**业务交换机*/
    @Bean
    public Exchange getBusinessExchange(){
        return ExchangeBuilder.directExchange(BUSINESS_EXCHANGE).durable(true).build();
    }
 
    /**死信交换机*/
    @Bean
    public Exchange getDeadLetterExchange(){return ExchangeBuilder.directExchange(DEAD_LETTER_EXCHANGE).durable(true).build();}
 
    /**订单未支付队列*/
    @Bean
    public Queue getUnpaidQueue(){
        Map<String,Object> args = new HashMap<>();
        //x-dead-letter-exchange 声明当前队列绑定的死信交换机
        args.put("x-dead-letter-exchange", DEAD_LETTER_EXCHANGE);
        //x-dead-letter-routing-key 声明当前队列的死信路由key
        args.put("x-dead-letter-routing-key", UNPAID_DEAD_LETTER_ROUTING_KEY);
        //设置过期时间
        args.put("x-message-ttl", QUEUE_PAID_EXPIRATION);
        return QueueBuilder.durable(UNPAID_QUEUE).withArguments(args).build();
    }
 
    /**绑定业务交换机和订单未支付队列*/
    @Bean
    public Binding bindUnpaid(){
        return BindingBuilder.bind(getUnpaidQueue()).to(getBusinessExchange()).with(UNPAID_ROUTING_KEY).noargs();
    }
 
    /**订单待支付死信队列*/
    @Bean
    public Queue getUnpaidDeadLetterQueue(){return new Queue(UNPAID_DEAD_LETTER_QUEUE);}
    /**绑定死信交换机和待支付死信队列*/
    @Bean
    public Binding bingUnpaidDeadLetter(){
        return BindingBuilder.bind(getUnpaidDeadLetterQueue()).to(getDeadLetterExchange()).with(UNPAID_DEAD_LETTER_ROUTING_KEY).noargs();
    }
 
}


    /**
     * 生成订单消息监听
     * @param orderInfoIn
     * @param message
     * @param channel
     * @throws CommonRuntimeException
     */
    @RabbitListener(queues = RabbitMQConfig.ORDER_CREATE_QUEUE)
    public void createOrderListener(OrderInfoIn orderInfoIn,Message message, Channel channel) throws CommonRuntimeException, IOException {
 
        String messageId = (String)redisTemplate.opsForValue().get(orderInfoIn.getMessageId().toString());
        if(messageId == null){//保证冥等性
            try {
                this.createOrder(orderInfoIn);
                redisTemplate.opsForValue().set(orderInfoIn.getMessageId().toString(), "ack");
                channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
 
                //存入订单待支付队列
                orderInfoIn.setMessageId(idGenerator.getGlobalId());//雪花算法保证冥等性
                rabbitTemplate.convertAndSend(RabbitMQConfig.BUSINESS_EXCHANGE,RabbitMQConfig.UNPAID_ROUTING_KEY,orderInfoIn);
 
            } catch (Exception e){
                logger.error("消费创建订单消息失败【】error："+ message.getBody());
                logger.error("OrderConsumer  handleMessage {} , error:",message,e);
                //处理消息失败，将消息重新放回队列
                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false,true);
            }
        }else{
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
        }
 
    }
 
    /**
     * 订单未支付消息监听
     * @param message
     * @param channel
     * @throws CommonRuntimeException
     *
     */
    @RabbitListener(queues = RabbitMQConfig.UNPAID_DEAD_LETTER_QUEUE)
    public void OrderDeadLetterListener(OrderInfoIn orderInfoIn,Message message, Channel channel) throws CommonRuntimeException, IOException {
        String messageId = (String)redisTemplate.opsForValue().get(orderInfoIn.getMessageId().toString());
        if(messageId == null){//保证冥等性
            try {
                //检测订单状态是否为已支付，否则回滚库存、订单支付超时等操作...
                redisTemplate.opsForValue().set(orderInfoIn.getMessageId().toString(), "ack");
                //消费成功
                channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
 
            } catch (Exception e){
                logger.error("消费订单未支付消息失败【】error："+ message.getBody());
                logger.error("handleMessage {} , error:",message,e);
                //处理消息失败，将消息重新放回队列
                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false,true);
            }
        }else{
            //已消费
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
        }
 
    }

到这里为止，基于SpringBoot的微服务项目里集成RabbitMQ应用就差不多搭建完成，本文结合一些网上的理论知识概念，及实例给大家介绍其中几个常用特性，后面有时间再介绍其他的高级特性，比如：消息100%投递成功方案、消息持久化、消息限流等功能，以上就是本文的全部内容，希望对大家的学习有所帮助，欢迎评论交流。能get到知识点不要忘了关注点赞~~~

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/weixin_40725706/article/detail/127016

深入理解RabbitMQ高级特性_channel.basicack

前言

目录

RabbitMQ介绍

RabbitMQ安装

RabbitMQ基本概念

RabbitMQ三种模式

SpringBoot整合RabbitMQ

ACK应答机制配置

重试机制配置

TTL队列/消息

死信队列DLX

如何保证只成功消费一次（冥等性）

延迟队列