RabbitMQ进阶学习

港控/mmm° 2022-05-12 08:04 204阅读 0赞

**目录**

一、 消息路由失败处理

1.1 mandatory 参数

1.2. immediate参数

1.3. 备份交换器

二、设置过期时间

2.1 .设置消息的TTL

2.2. 设置队列的TTL

三、死信队列

四、延迟队列

五、优先级队列

5.1. 设置队列的x-max-priority参数来实现

5.2. 在发送消息中设置消息当前的优先级

六、RPC实现

七、持久化

八、生产者确认

8.1 事务机制

8.2 发送发消息确认

8.3 批量confirm方法

8.4 异步Confirm方法

九、消费端要点介绍

9.1 消息分发basicQos

9.2 消息顺序性

9.3 弃用QueueingConsumer

十、消息传输保障

--------------------

## 一、 消息路由失败处理 ##

>    消息路由失败的处理方式有以下两种：
> 
>     1） mandatory和immediate是channel.basicPublish方法中的两个参数，它们都有当消息传递过程中不可达目的地时将消息返回给生产者的功能。
> 
>     2）RabbitMQ提供的备份交换器(Altemate  Exchange )可以将未能被交换器路由的消息(没有绑定队列或者没有匹配的绑定〉存储起来，而不用返回给客户端。

### 1.1 mandatory 参数 ###

发送消息的时候，如下设置了mandatory，在没有这个mandatory参数的时候，默认是false

> void basicPublish(String exchange, String routingKey, boolean mandatory, BasicProperties props, byte\[\] body);

***   当mandatory参数设为true时，交换器无法根据自身的类型和路由键找到一个符合条件的队列***，那么RabbitMQ会调用Basic.Return命令将消息返回给生产者。当mandatory参数设置为false时，出现上述情形，则消息直接被丢弃(这是默认情况下)。那么生产者如何获取到没有被正确路由到合适队列的消息呢?这时候可以通过调用channel.addReturnListener来添加ReturnListener监听器实现。代码如下：

channel.addReturnListener(new ReturnListener() {
                @Override
                public void handleReturn(int replyCode, String replyText, String exchange, String routeKey, 
                                AMQP.BasicProperties basicProperties, byte[] bytes) throws IOException {
                    String text = new String(bytes);
                    System.out.println(“返回的消息是：”+text);
                }
            });

### 1.2. immediate参数 ###

*** immediate设为true时，如果交换器在将消息路由到队列时发现队列上并不存在任何消费者***，那么这条消息将不会存入队列中。当与路由键匹配的所有队列都没有消费者时，该消息会通过Basic.Return返回至生产者。

***RabbitMQ3. 0版本开始去掉了对immediate参数的支持，***对此RabbitMQ官方解释是:immediate参数会影响镜像队列的性能，增加了代码复杂性，所以一般用过期时间和死信队列来实现。后面介绍

void basicPublish(String exchange, String routingKey, boolean mandatory, 
            boolean immediate, BasicProperties props, byte[] body);

总结：

概括来说，mandatory参数告诉服务器至少将该消息路由到一个队列中，否则将消息返回给生产者。immediate参数告诉服务器，如果该消息关联的队列上有消费者，则立刻投递:如果所有匹配的队列上都没有消费者，则直接将消息返还给生产者，不用将消息存入队列而等待消费者了。

### 1.3. 备份交换器 ###

备份交换器，英文名称为AlternateExchange，简称AE，或者更直白地称之为"备胎交换器"。生产者在发送消息的时候如果不设置mandatory参数，那么消息在未被路由的情况下将会丢失:如果设置了mandatory参数，那么需要添加ReturnListener的编程逻辑，生产者的代码将变得复杂。如果既不想复杂化生产者的编程逻辑，又不想消息丢失，那么可以使用备份交换器，这样可以将未被路由的消息存储在RabbitMQ中，再在需要的时候去处理这些消息。

可以通过在声明交换器(调用channel.exchangeDeclare方法)的时候添加alternate-exchange参数来实现，也可以通过策略(Policy,后面介绍)的方式实现。如果两者同时使用，则前者的优先级更高，会覆盖掉Policy的设置。

Map<String,Object> ars = new HashMap<>();
            ars.put("alternate-exchange","myAe");
            //这个ars就是把myAE交换器作为normalExchange的备用交换器
            channel.exchangeDeclare("normalExchange","direct",true,false,ars);
            channel.exchangeDeclare("myAe","fanout",true,false,null);
            
            channel.queueDeclare("normalQueue",true,true,false,null);
            channel.queueBind("normalQueue","normalExchange","normalKey");
            
            channel.queueDeclare("unroutedQueue",true,false,false,null);
            channel.queueBind("unroutedQueue","myAe","");

上面的代码中声明了两个交换器normalExchange和myAe，分别绑定了normalQueue和unroutedQueue这两个队列，同时将myAe设置为normalExchange的备份交换器。注意myAe的交换器类型为fanout(它会把所有发送到该交换器的消息路由到所有与该交换器绑定的队列中，不用管routeKey的事),这里的备用交换器如果设置其他类型，同样需要遵循路由规则，没有路由到消息也会丢失。

如果此时发送一条消息到normalExchange上，当路由键等于"normalKey"的时候，消息能正确路由到normalQueue这个队列中。如果路由键设为其他值，比如" errorKey,即消息不能被正确地路由到与normalExchange绑定的任何队列上，此时就会发送给myAe，进而发送到unroutedQueue这个队列。如下图：

![70][]

> 备份交换器的情况：
> 
>  *   如果设置的备份交换器不存在，客户端和RabbitMQ服务端都不会有异常出现，此时消息会失
>  *  如果备份交换器没有绑定任何队列，客户端和RabbitMQ服务端都不会有异常出现，此时消息会丢失
>  *  如果备份交换器没有任何匹配的队列，客户端和RabbitMQ服务端都不会有异常出现，此时消息会丢失
>  *  如果备份交换器和mandatory参数一起使用，那么mandatory参数无效

## 二、设置过期时间 ##

TTL,time to live的简称。即过期时间，RabbitMQ可以对队列和消息设置TTL。

### 2.1 .设置消息的TTL ###

两种方式来设置消息的TTL，通过队列和单独的每条消息来设置，两种都设置的话，那么以过期时间较小的哪个为准。消息一旦超过设置的TTL值时，就会变成“死信（dead message）”。

**1.1 通过队列来设置消息的TTL**

这种通过队列来设置的 消息，队列中的所有消息都有相同的过期时间。这种方式是在channel.queueDeclare（）,队列声明的时候设置x-message-ttl参数来实现单位是：毫秒   。当然可以通过Policy和HTTP API接口设置。

Map<String,Object> ars = new HashMap<>();
    ars.put("x-message-ttl",1000);
    channel.queueDeclare("queueName",true,false,false,ars);

如果不设置TTL，那么此消息就不会过期，如果将TTL设置为0，则表示此时可以直接将消息投递给消费者，否则该消息将被丢失，这种方式可以代替immediate参数，（immediate参数在投递失败时会调用Basic.Return将消息返回），这个功能通过死信队列返回这个失败的消息。

**1.2 对消息进行单独设置TTL**

**             **这个方式设置消息的TTL，每一个消息的TTL可以不同。通过channel.basicPublish()方法实现，通过加入expiration的属相参数。单位：毫秒，同样可以通过HTTP API接口设置； 如下代码：

channel.basicPublish(EXCHANGE_NAME,ROUTING_KEY,
                    new AMQP.BasicProperties().builder()
                        .deliveryMode(2)   //设置为持久化
                        .expiration("60000") //设置为1分钟的过期时间
                    .build(),message.getBytes());

对于第一种设置队列TTL属性的方法，一旦消息过期，就会从队列中抹去，而在第二种方法中，即使消息过期，也不会马上从队列中抹去，因为每条消息是否过期是在即将投递到消费者之前判定的。

为什么这两种方法处理的方式不一样?因为第一种方法里，队列中己过期的消息肯定在队列头部，RabbitMQ只要定期从队头开始扫描是否有过期的消息即可。而第二种方法里，每条消息的过期时间不同，如果要删除所有过期消息势必要扫描整个队列，所以不如等到此消息即将被消费时再判定是否过期，如果过期再进行删除即可。

### 2.2. 设置队列的TTL ###

通过channel.queueDeclare方法中的**x-expires参数可**以控制队列被自动删除前处于未使用状态的时间。未使用的意思是队列上没有任何的消费者，队列也没有被重新声明，并且在过期时间段内也未调用过Basic.Get命令。

设置队列里的TTL可以应用于类似RPC方式的回复队列，在RPC中，许多队列会被创建出来，但是却是未被使用的。RabbitMQ会确保在过期时间到达后将队列删除，但是不保障删除的动作有多及时。在RabbitMQ重启后，持久化的队列的过期时间会被重新计算。

用于表示过期时间的x-expires参数以毫秒为单位，井且服从和x-message-ttl一样的约束条件，不过不能设置为0。比如该参数设置为1000，则表示该队列如果在1秒钟之内未使用则会被删除。

代码如下：

Map<String,Object> ars = new HashMap<>();
    ars.put("x-expires",1000);
    channel.queueDeclare("queueDemo",true,false,false,ars);

## 三、死信队列 ##

DLX，全称为Dead-Letter-Exchange，可以称之为死信交换器，也有人称之为死信邮箱。当消息在一个队列中变成死信( dead message )后，它能被重新被发送到另一个交换器中，这个交换器就是DLX，绑定DLX的队列就称之为死信队列。

>      消息变成死信一般是由于以下几种情况:
> 
>  *  消息被拒绝(Basic.Reject/Basic.Nack)，并且设置requeue参数为fa l s e
>  *   消息过期;
>  *  队列达到最大长度。

通过在channel.queueDeclare方法中设置x-dead-letter-exchange参数来为这个队列添加DLX。

代码：

channel.exchangeDeclare("dlx_exchange","direct");
    Map<String,Object> ars= new HashMap<>();
    ars.put("x-dead-letter-message","dlx_exchange");
    //为队列添加DLX
    channel.queueDeclare("myQueue",false,false,false,ars);
    //也可以为这个DLX指定路由键，如果没有特殊指定，则使用原队列的路由键
    //ars.put("x-dead-letter-routing-key","dlx-routing-key");

对于RabbitMQ来说，DLX是一个非常有用的特性。它可以处理异常情况下，消息不能够被消费者正确消费(消费者调用了Basic.Nack或者Basic.Reject)而被置入死信队列中的情况，后续分析程序可以通过消费这个死信队列中的内容来分析当时所遇到的异常情况，进而可以改善和优化系统。DLX配合TTL使用还可以实现延迟队列的功能，详细请看下一节。

channel.exchangeDeclare("exchange.dlx","direct" ,true);
    channel.exchangeDeclare("exchange.normal", "fanout",true);
    Map <String , Object>  args  =  new  HashMap <String, Object>( );
    args.put("x-message-ttl",  10000);
    args.put("x-dead-letter-ex change"," exchange.dlx ");
    args.put("x-dead-letter-routing-key ", "routingkey");
    channe1.queueDec1are("queue.norma1", true, fa1se, fa1se, args);
    channe1.queueBind("queue.normal",  " exchange.normal", "");
    channe1.queueDec1are("queue.d1x",true,false,false, null) ;
    channel.q ueueBind("queue.dlx","exchange.dlx", Wroutingkey");
    channel.basicPublish("exchange.normal", "rk",
    MessageProperties.PERSISTENT_TEXT_PLAIN, "dlx".getBytes()) ;

这里创建了两个交换器exchange.normal和exchange.dlx，分别绑定两个队列queue.nomal和queue.dlx.如下图：

![70 1][]

## 四、延迟队列 ##

延迟队列存储的对象是对应的延迟消息，所谓"延迟消息"是指当消息被发送以后，并不想让消费者立刻拿到消息，而是等待特定时间后，消费者才能拿到这个消息进行消费。

>        延迟队列的使用场景有很多，比如:
> 
>  *  在订单系统中，一个用户下单之后通常有30分钟的时间进行支付，如果3 0分钟内没有支付成功，那么这个订单将进行异常处理，这时就可以使用延迟队列来处理这些订单了。
>  *  用户希望通过手机远程遥控家里的智能设备在指定的时间进行工作。这时候就可以将用户指令发送到延迟队列，当指令设定的时间到了再将指令推送到智能设备。

AMQP协议中，或者R a b b i t M Q本身没有直接支持延迟队列的功能，但是可以通过前面所介绍的DLX和TTL模拟出延迟队列的功能。

对于queue.dlx这个死信队列来说，同样可以看作延迟队列。假设一个应用中需要将每条消息都设置为10秒的延迟，生产者通过exchange.nomal这个交换器将发送的消息存储在queue.nomal这个队列中。消费者订阅的并非是queue.nomal这个队列，而是queue.dlx这个队列。当消息从queue.nomal这个队列中过期之后被存入queue.dlx这个队列中，消费者就恰巧消费到了延迟1 0秒的这条消息。

## 五、优先级队列 ##

优先级队列，顾名思义，具有高优先级的队列具有高的优先权，优先级高的消息具备优先被消费的特权。

### 5.1. 设置队列的x-max-priority参数来实现 ###

示例代码：

Map<String,Object> ars= new HashMap<>();
    ars.put("x-max-prioriy",10);
    channel.queueDeclare("myQueue",false,false,false,ars);

### 5.2. 在发送消息中设置消息当前的优先级 ###

示例代码:

AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
     AMQP.BasicProperties properties = builder.priority(5).build();
     channel.basicPublish("exchange_priority","rk_priority",properties,"message".getBytes());

上面的代码中设置消息的优先级为5。默认最低为0，最高为队列设置的最大优先级。优先级高的消息可以被优先消费，这个也是有前提的:如果在消费者的消费速度大于生产者的速度且Broker中没有消息堆积的情况下，对发送的消息设置优先级也就没有什么实际意义。因为生产者刚发送完一条消息就被消费者消费了，那么就相当于Broker中至多只有一条消息，对于单条消息来说优先级是没有什么意义的。

## 六、RPC实现 ##

RPC，是Remote Procedure  Call的简称，即远程过程调用。它是一种通过网络从远程计算机上请求服务，而不需要了解底层网络的技术。RPC的主要功用是让构建分布式计算更容易，在提供强大的远程调用能力时不损失本地调用的语义简洁性。

一般在RabbitMQ中进行RPC是很简单。客户端发送请求消息，服务端回复响应的消息为了接收响应的消息，我们需要在请求消息中发送一个回调队列(参考下面代码中的reply To )。使用默认的队列。

callbackQueueName = channel.queueDeclare().getQueue();
    BasicProperties props = new BasicProperties.Builder().replyTo(callbackQueueName).build();
    channel.basicPublish("", "rpc_queue",props,message.getBytes());
    // then code to read a response message from the callback_queue...

**Correlation Id**

在上述方法中为每个RPC请求创建一个回调队列。这是很低效的。幸运的是，一个解决方案：可以为每个客户端创建一个单一的回调队列。

新的问题被提出，队列收到一条回复消息，但是不清楚是那条请求的回复。这是就需要使用correlationId属性了。我们要为每个请求设置唯一的值。然后，在回调队列中获取消息，查看这个属性，关联response和request就是基于这个属性值的。如果我们看到一个未知的correlationId属性值的消息，可以放心的无视它——它不是我们发送的请求。

你可能问道，为什么要忽略回调队列中未知的信息，而不是当作一个失败？这是由于在服务器端竞争条件的导致的。虽然不太可能，但是如果RPC服务器在发送给我们结果后，发送请求反馈前就挂掉了，这有可能会发送未知correlationId属性值的消息。如果发生了这种情况，重启RPC服务器将会重新处理该请求。这就是为什么在客户端必须很好的处理重复响应，RPC应该是幂等的。

这里需要在channel.basicPublish方法中的参数AMQP.BasicProperties设置使用到replyTo和correlationId两个属性，分别表示需要回复的对列和，请求的唯一表示，用来表示回调后的唯一标识符。如图：

![70 2][]

>  *  当客户端启动时，创建一个匿名的回调队列(名称由RabbitMQ自动创建，图中的回调队列为amq . gen-LhQzlgv3 GhDOv8PIDabOXA)。
>  *  客户为RPC请求设置2个属性:replyTo用来告知RPC服务端回复请求时的目的队列，即回调队列;correlationld用来标记一个请求。
>  *  请求被发送到rpc\_queue队列中。
>  *  RPC服务端监听rpc\_queue队列中的请求，当请求到来时，服务端会处理并且把带有结果的消息发送给客户端。接收的队列就是replyTo设定的回调队列。
>  *  客户端监听回调队列，当有消息时，检查correlationld属性，如果与请求匹配，那就是结果了。

**官方DEMO,完成斐波那契数据的获取：**

**服务端代码： **

public class RPCServer {
        private static final String RPC_QUEUE_NAME = "rpc_queue";
    
        public static void main(String args[]) throws Exception {
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost(RabbitConfig.ip);
            factory.setPort(RabbitConfig.port);
            factory.setUsername(RabbitConfig.username);
            factory.setPassword(RabbitConfig.password);
    
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
    
            channel.queueDeclare(RPC_QUEUE_NAME,false,false,false,null);
            channel.basicQos(1);
    
            QueueingConsumer consumer = new QueueingConsumer(channel);
            channel.basicConsume(RPC_QUEUE_NAME, false, consumer);
            System.out.println(" [x] Awaiting RPC requests");
    
            while(true){
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                BasicProperties props = delivery.getProperties();
                BasicProperties replyProps = new BasicProperties.Builder().correlationId(props.getCorrelationId()).build();
                String message = new String(delivery.getBody());
                int n = Integer.parseInt(message);
                System.out.println(" [.] fib("+message+")");
                String repsonse = ""+fib(n);
                channel.basicPublish("", props.getReplyTo(), replyProps, repsonse.getBytes());
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        }
    
        private static int fib(int n) throws Exception {
            if (n == 0) return 0;
            if (n == 1) return 1;
            return fib(n-1) + fib(n-2);
        }
    }

**客户端代码：**

public class RPCClient {
        private Connection connection;
        private Channel channel;
        private String requestQueueName = "rpc_queue";
        private String replyQueueName;
        private QueueingConsumer consumer;
    
        public RPCClient() throws IOException, TimeoutException {
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost(RabbitConfig.ip);
            factory.setPort(RabbitConfig.port);
            factory.setUsername(RabbitConfig.username);
            factory.setPassword(RabbitConfig.password);
    
            connection = factory.newConnection();
            channel = connection.createChannel();
    
            replyQueueName = channel.queueDeclare().getQueue();
            consumer = new QueueingConsumer(channel);
            channel.basicConsume(replyQueueName, true,consumer);
        }
    
        public String call(String message) throws IOException,
                ShutdownSignalException, ConsumerCancelledException,
                InterruptedException {
            String response = null;
            String corrId = UUID.randomUUID().toString();
    
            BasicProperties props = new BasicProperties.Builder()
                    .correlationId(corrId)
                    .replyTo(replyQueueName)
                    .build();
            channel.basicPublish("", requestQueueName, props, message.getBytes());
    
            while(true){
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                if(delivery.getProperties().getCorrelationId().equals(corrId)){
                    response = new String(delivery.getBody());
                    break;
                }
            }
    
            return response;
        }
    
        public void close() throws Exception{
            connection.close();
        }
    
        public static void main(String args[]) throws Exception{
            RPCClient fibRpc = new RPCClient();
            System.out.println(" [x] Requesting fib(30)");
            String response = fibRpc.call("30");
            System.out.println(" [.] Got '"+response+"'");
            fibRpc.close();
    
        }
    }

## 七、持久化 ##

>      持久化可以提高RabbitMQ的可靠性，以防在异常情况(重启、关闭、右机等)下的数据丢失。本节针对这个概念做一个总结。RabbitMQ的持久化分为三个部分:交换器的持久化、队列的持久化和消息的持久化。
> 
>  *  交换器的持久化是通过在声明队列是将durable参数置为true实现的，详细可以参考[Rabbit API使用][Rabbit API]。如果交换器不设置持久化，那么在RabbitMQ服务重启之后，相关的交换器元数据会丢失，不过消息不会丢失，只是不能将消息发送到这个交换器中了。对一个长期使用的交换器来说，建议将其置为持久化的。
>  *  队列的持久化是通过在声明队列时将durable参数置为true实现的，详细内容可以参考[Rabbit API使用][Rabbit API]。如果队列不设置持久化，那么在RabbitMQ服务重启之后，相关队列的元数据会丢失，此时数据也会丢失。正所谓"皮之不存，毛将焉附"，队列都没有了，消息又能存在哪里呢?
>  *  队列的持久化能保证其本身的元数据不会因异常情况而丢失，但是并不能保证内部所存储的消息不会丢失。要确保消息不会丢失，需要将其设置为持久化。通过将消息的投递模式(BasicPropertie s中的deliveryMode属性)设置为2即可实现消息的持久化。MessageProperties.PERSISTENTTEXTPLAIN实际上是封装了这个属性，这个“2”就是表示消息持久化。

public static final BasicProperties PERSISTENT_TEXT_PLAIN = new   
              BasicProperties("text/plain", (String)null, (Map)null, 2, 0, (String)null, (String)null,
                 (String)null, (String)null, (Date)null, (String)null, (String)null, (String)null, (String)null);

设置了队列和消息的持久化，当RabbitMQ服务重启之后，消息依旧存在。单单只设置队列持久化，重启之后消息会丢失;单单只设置消息的持久化，重启之后队列消失，继而消息也丢失。单单设置消息持久化而不设置队列的持久化显得毫无意义。

> 交换器、队列、消息都设置了持久化之后就能百分之百保证数据不丢失了吗?答案是否定的。
> 
>      1>   首先从消费者来说，如果在订阅消费队列时将autoAck参数设置为true，那么当消费者接收到相关消息之后，还没来得及处理就宕机了，这样也算数据丢失。这种情况很好解决，将autoAck参数设置为false，并进行手动确认，详细可以参考[Rabbit API使用][Rabbit API]。
> 
>      2>   其次，在持久化的消息正确存入RabbitMQ之后，还需要有一段时间(虽然很短，但是不可忽视〉才能存入磁盘之中。Rabbit并不会为每条消息都进行同步存盘(调用内核的fsync方法)的处理，可能仅仅保存到操作系统缓存之中而不是物理磁盘之中。如果在这段时间内RabbitMQ服务节点发生了岩机、重启等异常情况，消息保存还没来得及落盘，那么这些消息将会丢失。
> 
>     这个问题怎么解决?这里可以引入RabbitMQ的镜像队列机制(后面解析)，相当于配置了副本，如果主节点（master）在此特殊时间内挂掉，可以自动切换到从节点(slave) ,这样有效地保证了高可用性，除非整个集群都挂掉。虽然这样也不能完全保证RabbitMQ消息不丢失，但是配置了镜像队列要比没有配置镜像队列的可靠性要高很多，在实际生产环境中的关键业务队列一般都会设置镜像队列。还可以在发送端引入事务机制或者发送方确认机制来保证消息己经正确地发送并存储至RabbitMQ中，前提还要保证在调用channel.basicPublish方法的时候交换器能够将消息正确路由到相应的队列之中。

## 八、生产者确认 ##

**在使用RabbitMQ的时候，可以通过消息持久化操作来解决因为服务器的异常崩溃而导致的消息丢失**，除此之外，我们还会遇到一个问题，当消息的生产者将消息发送出去之后，消息到底有没有正确地到达服务器呢?如果不进行特殊配置，默认情况下发送消息的操作是不会返回任何信息给生产者的，也就是默认情况下生产者是不知道消息有没有正确地到达服务器。如果在消息到达服务器之前己经丢失，持久化操作也解决不了这个问题，因为消息根本没有到达服务器，何谈持久化?

> RabbitMQ针这个问题，提供了两种解决方式:
> 
>  *  1）通过事务机制实现
>  *  2）通过发送方确认（publisher confirm）机制实现

### 8.1 事务机制 ###

>       RabbitMQ客户端中与事务机制相关的方法有三个:
> 
>  *  channel.txSelect用于将当前信道设置成事务模式
>  *  channel.txCommit用于提交事务
>  *  channel.txRollback用于事务回滚
> 
>        在通过channel.txSelect方法开启事务之后，我们便可以发布消息给RabbitMQ（这里是指交换器）了，如果事务提交成功，则消息一定到达了RabbitMQ中，如果在事务提交执行之前由于RabbitMQ异常崩溃或者其他原因抛出异常，这个时候我们便可以将其捕获，进而通过执行channel.txRollback方法来实现事务回滚。注意这里的RabbitMQ中的事务机制与大多数数据库中的事务概念井不相同，需要注意区分。

例如代码：

try{
                channel.txSelect();
                channel.basicPublish("exchangeName","routeKey",props,"message".getBytes());
                channel.txCommit();
            }catch (Exception e){
                e.printStackTrace();
                channel.txRollback();
            }

如果需要发送多条消息，可以通过如下方式提交：

channel.txSelect();
            for (int i=0;i<100;i++){
                try{
                    channel.basicPublish("exchangeName","routeKey",props,"message".getBytes());
                    channel.txCommit();
                }catch (Exception e){
                    e.printStackTrace();
                    channel.txRollback();
                }
            }

事务确实能够解决消息发送方和RabbitMQ之间消息确认的问题，只有消息成功被RabbitMQ接收，事务才能提交成功，否则便可在捕获异常之后进行事务回滚，与此同时可以进行消息重发。但是使用事务机制会"吸干"RabbitMQ的性能，那么有没有更好的方法既能保证消息发送方确认消息已经正确送达，又能基本上不带来性能上的损失呢?从AMQP协议层面来看并没有更好的办法，但是RabbitMQ提供了一个改进方案，即发送方确认机制。

### 8.2 发送发消息确认 ###

上面介绍了RabbitMQ可能会遇到的一个问题，即消息发送方(生产者〉并不知道消息是否真正地到达了RabbitMQ。随后了解到在AMQP协议层面提供了事务机制来解决这个问题，但是采用事务机制实现会严重降低RabbitMQ的消息吞吐量，这里就引入了一种轻量级的方式一发送方确认（publisher  confirm )机制。

生产者将信道设置成confirm（确认)模式，一旦信道进入confirm模式，所有在该信道上面发布的消息都会被指派一个唯一的I D (从1开始)，一旦消息被投递到所有匹配的队列之后，RabbitMQ就会发送一个确认(Basic.Ack)给生产者(包含消息的唯一ID )，这就使得生产者知晓消息已经正确到达了目的地了。如果消息和队列是可持久化的，那么确认消息会在消息写入磁盘之后发出。RabbitMQ回传给生产者的确认消息中的deliveryTag包含了确认消息的序号，此外RabbitMQ也可以设置channel.basicAck方法中的multiple参数，表示到这个序号之前的所有消息都己经得到了处理。

事务机制在一条消息发送之后会使发送端阻塞，以等待RabbitMQ的回应，之后才能继续发送下一条消息。相比之下，发送方确认机制最大的好处在于它是异步的，一旦发布一条消息，生产者应用程序就可以在等信道返回确认的同时继续发送下一条消息，当消息最终得到确认之后，生产者应用程序便可以通过回调方法来处理该确认消息，如果RabbitMQ因为自身内部错误导致消息丢失，就会发送一条nack （Basic.Nack)命令，生产者应用程序同样可以在回调方法中处理该nack命令。

简要代码：

try{
                channel.confirmSelect();
                channel.basicPublish("exchangeName","routeKey",MessageProperties.PERSISTENT_TEXT_PLAIN,"message".getBytes());
                if (!channel.waitForConfirms()){
                    System.out.println("send message failed");
                    // do something else
                }
            }catch(Exception e){
                e.printStackTrace();
            }

如果发送多条消息，只需要将channel.basicPublish和channel.waitForConfirms方法包裹在循环里面即可，可以参考事务机制，不过不需要把channel.confirmSelect方法包裹在循环内部。

对于channel.waitForConfirms，他有四个同类的方法，如下：

boolean waitForConfirms() throws InterruptedException;
    
        boolean waitForConfirms(long timeout) throws InterruptedException, TimeoutException;
    
        void waitForConfirmsOrDie() throws IOException, InterruptedException;
    
        void waitForConfirmsOrDie(long timeout) throws IOException, InterruptedException, TimeoutException;

如果信道没有开启publisher confirm模式，则调用任何waitForConfirms方法都会报出java. lang.IllegalStateException。对于没有参数的waitForConfirms方法来说，其返回的条件是客户端收到了相应的Basic.Ack/.Nack或者被中断。参数timeout表示超时时间，一旦等待RabbitMQ回应超时就会抛出java.util.concurrent.TimeoutException的异常。两个waitForConfirmsOrDie方法在接收到RabbitMQ返回的Basic.Nack之后会抛出java.i o.IOException。业务代码可以根据自身的特性灵活地运用这四种方法来保障消息的可靠发送。

> **总结：**上面的事务机制和普通的发送方的消息确认机制其实QPS差不多，发方消息确认机制稍微高一点，但其实提升不高。另外两者不能混用，只能选择一种使用，否则会抛出相应的异常情况。这里的消息发送到RabbitMQ指的是发送到对应的RabbitMQ的交换器，如果交换器没有对应的队列，消息也会很容易的丢失，所以发送方也会使用到Mandatory参数和备用队列来提高消息传输的可靠性。

> 发送方消息确认机制改进，提高相应的QPS（吞吐量），方式有如下两种：
> 
>  *  批量confirm方法:每发送一批消息后，调用channel.waitForConfirms方法，等待服务器的确认返回。
>  *  异步confirm方法:提供一个回调方法，服务端确认了一条或者多条消息后客户端会回调这个方法进行处理

### 8.3 批量confirm方法 ###

在批量confirm方法中，客户端程序需要定期或者定量(达到多少条)，亦或者两者结合起来调用channel. waitForConfirms来等待RabbitMQ的确认返回。相比于前面示例中的普通confirm方法，批量极大地提升了confirm的效率，但是问题在于出现返回Basic.Nack或者超时情况时，客户端需要将这一批次的消息全部重发，这会带来明显的重复消息数量，并且当消息经常丢失时，批量confirm的性能应该是不升反降的。

示例代码：

int  BATCH_COUNT = 10;
            try{
                channel.confirmSelect();
                int msgCount = 0;
                while (true){
                    channel.basicPublish("exchangeName","routeKey",MessageProperties.PERSISTENT_TEXT_PLAIN,"message".getBytes());
                    //将发送出去的消息传入缓存中，缓存可以是一个ArrayList或者BlockingQueue之类
                    if (++msgCount >= BATCH_COUNT){
                        msgCount = 0;
                        try {
                            if (channel.waitForConfirms()){
                                //将缓存中的消息清空
                            }
                        }catch (Exception e){
                            e.printStackTrace();
                            //重新将缓存中的消息重新发送
                        }
                    }
                }

### 8.4 异步Confirm方法 ###

在客户端Channel接口中提供的addConfirmListener方法可以添加ConfirmListener这个回调接口，这个ConfirmListener接口包含两个方法:handleAck和handleNack，分别用来处理RabbitMQ回传的Basic.Ack和Basic.Nack。在这两个方法中都包含有一个参数deliveryTag(在publisher comfirm模式下用来标记消息的唯一有序序号)。我们需要为每一个信道维护一个"unconfirm"的消息序号集合，每发送一条消息，集合中的元素加1。每当调用ConfirmListener中的handleAck方法时，"unconfirm"集合中删掉相应的一条(multiple设置为false)或者多条(multiple设置为true)记录。从程序运行效率上来看，这个"unconfirm"集合最好采用有序集合SortedSet的存储结构。事实上，Java客户端SDK中的waitForConfirms方法也是通过SortedSet维护消息序号的。

Set<Long> confirmSet = new TreeSet<>(); //TreeSet是实现了SortedSet接口的实现类
            channel.confirmSelect();
            channel.addConfirmListener(new ConfirmListener() {
                @Override
                public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                    System.out.println("Nack SeqNo:"+deliveryTag+",multiple:"+multiple);
                    if (multiple){
                       ((TreeSet<Long>) confirmSet).headSet(deliveryTag-1).clear();
                    }else {
                        confirmSet.remove(deliveryTag);
                    }
                }
    
                @Override
                public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                    if (multiple){
                        //这里headSet表示后面参数之前的所有元素
                        ((TreeSet<Long>) confirmSet).headSet(deliveryTag-1).clear();
                    }else {
                        confirmSet.remove(deliveryTag);
                    }
                    //这里需要添加处理消息重发的场景
                }
            });
            //这里是一直发送消息的场景
            while (true){
                long nextPublishSeqNo = channel.getNextPublishSeqNo();
                channel.basicPublish("exchangeName","routeKey",MessageProperties.PERSISTENT_TEXT_PLAIN,"message".getBytes());
                confirmSet.add(nextPublishSeqNo);
            }

> 总结：最后我们事务、普通confirm、批量confirm和异步confirm这4种方式放到一起来比较一下彼此的QPS。测试后发现批量confirm和异步confirm这两种方式所呈现的性能要比其余两种好得多。
> 
>  *  事务机制和普通confirm的方式吐吞量很低，但是编程方式简单，不需要在客户端维护状态(这里指的是维护deliveryTag及缓存未确认的消息)。
>  *  批量confirm方式的问题在于遇到RabbitMQ服务端返回Basic.Nack需要重发批量消息而导致的性能降低。
>  *  异步confirm方式编程模型最为复杂，而且和批量confirm方式一样需要在客户端维护状态。在实际生产环境中采用何种方式，这里就仁者见仁智者见智了，不过强烈建议读者使用**异步confirm的方式**

## 九、消费端要点介绍 ##

消费者客户端可以通过推模式或者拉模式的方式来获取并消费消息，当消费者处理完业务逻辑需要手动确认消息己被接收，这样RabbitMQ才能把当前消息从队列中标记清除。当然如果消费者由于某些原因无法处理当前接收到的消息，可以通过channel.basicNack或者channel.basicReject来拒绝掉。

>  这里对于RabbitMQ消费端来说，还有几点需要注意:
> 
>  *  消息分发
>  *  消息顺序性
>  *  弃用QueueingConsumer

### 9.1 消息分发basicQos ###

当RabbitMQ队列拥有多个消费者时，队列收到的消息将以轮询( round-robin)的分发方式发送给消费者。每条消息只会发送给订阅列表里的一个消费者。这种方式非常适合扩展，而且它是专门为并发程序设计的。如果现在负载加重，那么只需要创建更多的消费者来消费处理消息即可。

很多时候轮询的分发机制也不是那么优雅。默认情况下，如果有n个消费者，那么RabbitMQ会将第m条消息分发给第m%n(取余的方式)个消费者，RabbitMQ不管消费者是否消费并已经确认(Basic.Ack)了消息。试想一下，如果某些消费者任务繁重，来不及消费那么多的消息，而某些其他消费者由于某些原因(比如业务逻辑简单、机器性能卓越等)很快地处理完了所分配到的消息，进而进程空闲，这样就会造成整体应用吞吐量的下降。那么该如何处理这种情况呢?

这里就要用到channel.basicQos(int  prefetchCount)这个方法，channel.basicQos方法允许限制信道上的消费者所能保持的最大未确认消息的数量。

举例说明，在订阅消费队列之前，消费端程序调用了channel.basicQos(5)，之后订阅了某个队列进行消费。RabbitMQ会保存一个消费者的列表，每发送一条消息都会为对应的消费者计数，如果达到了所设定的上限，那么RabbitMQ就不会向这个消费者再发送任何消息。直到消费者确认了某条消息之后，RabbitMQ将相应的计数减1，之后消费者可以继续接收消息，直到再次到达计数上限。这种机制可以类比于TCP/IP中的"滑动窗口"。 这种方式对于消费端的拉模式是无效的。

方法有三种重载方式：

void basicQos(int prefetchSize, int prefetchCount, boolean global) throws IOException;
    
        void basicQos(int prefetchCount, boolean global) throws IOException;
    
        void basicQos(int prefetchCount) throws IOException;

prefetchCount这个参数，当prefetchCount设置为0则表示没有上限。

prefetchSize这个参数表示消费者所能接收未确认消息的总体大小的上限，单位为B，设置为0则表示没有上限。

对于一个信道来说，它可以同时消费多个队列，当设置了prefetchCount大于0时，这个信道需要和各个队列协调以确保发送的消息都没有超过所限定的prefetchCount的值，这样会使RabbitMQ的性能降低，尤其是这些队列分散在集群中的多个Broker节点之中。RabbitMQ为了提升相关的性能，在AMQP 0-9-1协议之上重新定义了global这个参数

<table> 
 <tbody> 
  <tr> 
   <td style="width:89px;">global参数</td> 
   <td style="width:369px;">AMQP 0-9-1</td> 
   <td style="width:391px;">RabbitMQ</td> 
  </tr> 
  <tr> 
   <td style="width:89px;">false</td> 
   <td style="width:369px;">信道上<span style="color:#f33b45;">所有的消费者</span>都需要遵从prefetchCount的限定值</td> 
   <td style="width:391px;">信道上<span style="color:#f33b45;">新的消费者</span>需要遵从prefetchCount的限定值</td> 
  </tr> 
  <tr> 
   <td style="width:89px;">true</td> 
   <td style="width:369px;"><span style="color:#f33b45;">当前通信链路</span>（Connection）上所有的消费者都需要遵循prefetchCount的限定值</td> 
   <td style="width:391px;">信道上<span style="color:#f33b45;">所有的消费者</span>都需要遵从prefetchCount的限定值</td> 
  </tr> 
 </tbody> 
</table>

如果一个channel设置了两个basicQos，一个global是true,一个是false，同样两个都会起作用。

channel.basicQos(3,false); //这里表示每个消费者的限制为3
    channel.basicQos(5,true);//这里表示当前信道最多为5，所以当消费者1有3个未确认时，消费者2最多只能有2个获取到没有确认消费
    channel.basicConsume("queue1",false,consumer1)
    channel.basicConsume("queue2",false,consumer2)

### 9.2 消息顺序性 ###

>          这个并不一定能保证消息是顺序性的，如下都不一定会保证消息的顺序性
> 
> 1.  在存在事务机制下，有可能由于异常，需要重新发送
> 2.  在多个生产者时，也不能保证顺序性
> 3.  设置了超时消息，变成死信队列
> 4.  消费者拒绝消息，后重新消费.......

### 9.3 弃用QueueingConsumer ###

这个QueueingConsumer在RabbitMQ在4.X版本开始废除，理由是很有可能会造成内存不足，会给RabbitMQ性能造成很多不足。

## 十、消息传输保障 ##

消息可靠传输一般是业务系统接入消息中间件时首要考虑的问题.

> 一般消息中间件的消息传输保障分为三个层级:
> 
>  *  At most once:最多一次。消息可能会丢失，但绝不会重复传输。
>  *  At least  once :最少一次。消息绝不会丢失，但可能会重复传输。。
>  *  Exactly  once :恰好一次。每条消息肯定会被传输一次且仅传输一次。

RabbitMQ**只能做到最多一次和至少一次，不能做到恰好一次**，

**最多一次**就是发送消息就不管了

> 而**至少一次**则需要考虑一下问题：
> 
>  *  (1 ) 消息生产者需要开启事务机制或者publisher confirm机制，以确保消息可以可靠地传输到RabbitMQ中。
>  *  ( 2 )消息生产者需要配合使用mandatory参数或者备份交换器来确保消息能够从交换器路由到队列中，进而能够保存下来而不会被丢弃。
>  *  ( 3 )消息和队列都需要进行持久化处理，以确保RabbitMQ服务器在遇到异常情况时不会丢失消息
>  *  ( 4 )消费者在消费消息的同时需要将autoAck设置为false，然后通过手动确认的方式去确认已经正确消费的消息，以避免在消费端引起不必要的消息丢失。

恰好一次会在某些网络原因情况下，造成消息重复发送，或者消费端消费消息后确认信息由于网络故障丢失，重复消费。而且RabbitMQ也没有实现去重的功能，目前主流的xiao消息中间件其实都没有消息去重机制，所以去重一般会有客户端自己去考虑这个问题。

> ***总结：***
> 
> ** 提高消息的可靠性：**
> 
> 设置mandatory参数或者备份交换器(immediate参数己被陶汰);
> 
> 设置publisher confirml机制或者事务机制;
> 
> 设置交换器、队列和消息都为持久化;
> 
> 设置消费端对应的autoAck参数为false并在消费完消息之后再进行消息确认。

[70]: /images/20220512/4ec3b27b5e6c478d9db053d7f4ffe424.png
[70 1]: /images/20220512/c96a886d0bf345fba98a018002f09c93.png
[70 2]: /images/20220512/237f630894fe462fa8aeba0664f73165.png
[Rabbit API]: https://blog.csdn.net/weixin_40792878/article/details/82597177#t7