消息队列篇-可靠消息投递_publisher-confirm-type: correlated

一、生产者确认

在yml文件中配置：

spring:
	rabbitmq:
		publisher-confirm-type: CORRELATED
    	publisher-returns: true
1
2
3
4

publisher-confirm-type：消息确认的类型

1、publisher-confirm-type

	/**
	 * The type of publisher confirms to use.
	 */
public enum ConfirmType {

	/**
	 * Use {@code RabbitTemplate#waitForConfirms()} (or {@code waitForConfirmsOrDie()}
	 * within scoped operations.
	 */
	SIMPLE,

	/**
	 * Use with {@code CorrelationData} to correlate confirmations with sent
	 * messsages.
	 */
	CORRELATED,

	/**
	 * Publisher confirms are disabled (default).
	 */
	NONE

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

1. NONE：禁用发布确认模式，默认值
2. CORRELATED：发布消息成功到交互器你后会触发回调方法
3. SIMPLE：发布消息成功后使用rabbitTemplate调用waitForConfirms或waitForConfirmsOrDie方法等待broker节点返回发送结果，根据返回结果来判定下一步的逻辑，要注意的点是waitForConfirmsOrDie方法如果返回false则会关闭channel，则接下来无法发送消息到broker;

2、publisher-returns

未投递到Queue退回模式

3、测试发送消息

1.正常发送

@Test
public void sendMessageTest(){

	/**
	 * 发送消息
	 */
	rabbitTemplate.convertAndSend("java-direct-exchange","hello-java","hello world!");

	log.info("发送成功");
}
1
2
3
4
5
6
7
8
9
10

2.测试到达交换机，没有到达队列

结果：

returnMessage执行。

二、消费者确认

配置yml文件：

spring:
	rabbitmq:
    	listener:
      		simple:
        		acknowledge-mode: manual/auto/none
1
2
3
4
5

消费者三种消息确认模式：

• MANUAL：手动确认模式，需要手动调用chennel.basicAck确认。可以捕获异常控制重试次数，甚至可以控制失败消息的处理方式
• AUTO：自动确认模式，默认，消费者没有异常会自动确认，有异常则不确认，无限重试，导致程序死循环。
• NONE：不确认模式，不管程序是否异常，只要执行了监听方法，消息就被消费了。相当于rabbitmq中的自动确认，不推荐。

1.AUTO 自动确认

/**
   * 监听java-queue队列中OrderReturnApplyEntity的消息
   * @param msg
   * @param orderReturnApplyEntity
   */
  @RabbitHandler
  public void  recivedMessage(Message msg, OrderReturnApplyEntity orderReturnApplyEntity){
      System.out.println("orderReturnApplyEntity:"+orderReturnApplyEntity);
      //添加异常
      int i = 1 / 0;
  }

1
2
3
4
5
6
7
8
9
10
11
12

发送消息：

消费者不停重试：

消息一致是未确认：

停掉消费者重新回到队列，消息没有丢失：

2.NONE 不确认模式

在yml中添加配置

acknowledge-mode: none
再次运行上述异常代码 并发送消息：
消费者抛出异常：

消息被消费：

3.MANUAL 手动确认

在yml中添加配置

acknowledge-mode: manual

改造消费者代码：

@RabbitHandler
 public void  recivedMessage(Message msg, OrderReturnApplyEntity orderReturnApplyEntity, Channel channel) throws IOException {

     try {
         System.out.println("接收到消息：" + msg);

         int i = 1 / 0;
         // 确认收到消息，false只确认当前consumer一个消息收到，true确认所有consumer获得的消息
         channel.basicAck(msg.getMessageProperties().getDeliveryTag(), false);
     } catch (Exception e) {
         if (msg.getMessageProperties().getRedelivered()) {
             System.out.println("消息重试后依然失败，拒绝再次接收");
             // 拒绝消息，不再重新入队（如果绑定了死信队列消息会进入死信队列，没有绑定死信队列则消息被丢弃，也可以把失败消息记录到redis或者mysql中），也可以设置为true再重试。
             channel.basicReject(msg.getMessageProperties().getDeliveryTag(), false);
         } else {
             System.out.println("消息消费时出现异常，即将再次返回队列处理");
             // Nack消息，重新入队（重试一次）
             channel.basicNack(msg.getMessageProperties().getDeliveryTag(), false, true);
         }
         e.printStackTrace();
     }
 }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

1. 确认消息：

basicAck表示确认消息，false只确认当前一个消息

 // 确认收到消息，false只确认当前consumer一个消息收到，true确认所有consumer获得的消息
channel.basicAck(msg.getMessageProperties().getDeliveryTag(), false);
1
2

2. 拒绝消息：

basicReject表示拒绝消息，false表示不再重新入队列

 // 拒绝消息，不再重新入队（如果绑定了死信队列消息会进入死信队列，没有绑定死信队列则消息被丢弃，也可以把失败消息记录到redis或者mysql中），也可以设置为true再重试。
channel.basicReject(msg.getMessageProperties().getDeliveryTag(), false);
1
2

3. 不确认消息：
   basicNack表示不确认消息，参数二表示是否批量，参数三表示是否重新入队列

 // Nack消息，重新入队（重试一次）
channel.basicNack(msg.getMessageProperties().getDeliveryTag(), false, true);
1
2

发送消息：

消费消息：

接收到消息：(Body:'{"id":1,"orderId":null,"skuId":null,"orderSn":null,"createTime":1624356355348,"memberUsername":null,"returnAmount":null,"returnName":null,"returnPhone":null,"status":null,"handleTime":null,"skuImg":null,"skuName":"iPhone","skuBrand":null,"skuAttrsVals":null,"skuCount":null,"skuPrice":null,"skuRealPrice":null,"reason":null,"description述":null,"descPics":null,"handleNote":null,"handleMan":null,"receiveMan":null,"receiveTime":null,"receiveNote":null,"receivePhone":null,"companyAddress":null}' MessageProperties [headers={spring_listener_return_correlation=79900859-553e-461b-9bc4-83241e4138e4, __TypeId__=com.simon.gulimall.order.entity.OrderReturnApplyEntity}, contentType=application/json, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=java-direct-exchange, receivedRoutingKey=hello-java, deliveryTag=1, consumerTag=amq.ctag-Q5c52wics9W5XNrHtwQTSg, consumerQueue=java-queue])
消息消费时出现异常，即将再次返回队列处理
接收到消息：(Body:'{"id":1,"orderId":null,"skuId":null,"orderSn":null,"createTime":1624356355348,"memberUsername":null,"returnAmount":null,"returnName":null,"returnPhone":null,"status":null,"handleTime":null,"skuImg":null,"skuName":"iPhone","skuBrand":null,"skuAttrsVals":null,"skuCount":null,"skuPrice":null,"skuRealPrice":null,"reason":null,"description述":null,"descPics":null,"handleNote":null,"handleMan":null,"receiveMan":null,"receiveTime":null,"receiveNote":null,"receivePhone":null,"companyAddress":null}' MessageProperties [headers={spring_listener_return_correlation=79900859-553e-461b-9bc4-83241e4138e4, __TypeId__=com.simon.gulimall.order.entity.OrderReturnApplyEntity}, contentType=application/json, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=true, receivedExchange=java-direct-exchange, receivedRoutingKey=hello-java, deliveryTag=2, consumerTag=amq.ctag-Q5c52wics9W5XNrHtwQTSg, consumerQueue=java-queue])
java.lang.ArithmeticException: / by zero
	at com.simon.gulimall.order.service.impl.OrderItemServiceImpl.recivedMessage(OrderItemServiceImpl.java:41)
	
消息重试后依然失败，拒绝再次接收
1
2
3
4
5
6
7

第一步报错，进行重试，如果重试依然报错，就进行拒绝。