rabbitmq重试机制

客官°小女子只卖身不卖艺 2022-04-15 06:27 332阅读 0赞

# 1、应答模式 #

## NONE ##

可以称之为自动回调，即使无响应或者发生异常均会通知队列消费成功，会丢失数据。

## AUTO ##

自动检测异常或者超时事件，如果发生则返回noack，消息自动回到队尾，但是这种方式可能出现消息体本身有问题，返回队尾其他队列也不能消费，造成队列阻塞。

## MANUAL ##

手动回调，在程序中我们可以对消息异常记性捕获，如果出现消息体格式错误问题，手动回复ack，接着再次调用发送接口把消息推到队尾。

# 2、java示例 #

## 2.1 pom.xml ##

<dependencies>
    	
    	<dependency>
    		<groupId>org.springframework.boot</groupId>
    		<artifactId>spring-boot-starter-web</artifactId>
    	</dependency>
    	<dependency>
    		<groupId>org.springframework.boot</groupId>
    		<artifactId>spring-boot-starter-amqp</artifactId>
    	</dependency>
    </dependencies>

## 2.2 配置文件application.yml ##

spring:
      application:
        name: rabbitmq
      rabbitmq:
        host: 127.0.0.1
        port: 5672
        username: admin
        password: pwd
        listener:
          simple:
            retry:
              enabled: true #是否开启消费者重试（为false时关闭消费者重试，这时消费端代码异常会一直重复收到消息）
              max-attempts: 5 #最大重试次数
              initial-interval: 5000 #重试间隔时间（单位毫秒） 
              max-interval: 1200000 #重试最大时间间隔（单位毫秒） 
              multiplier: 5 #应用于前一重试间隔的乘法器。

## 2.3 配置类 ##

import org.springframework.amqp.core.Binding;
    import org.springframework.amqp.core.BindingBuilder;
    import org.springframework.amqp.core.Queue;
    import org.springframework.amqp.core.TopicExchange;
    import org.springframework.beans.factory.annotation.Qualifier;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    /** 
     * @ClassName: TopicRabbitConfig 
     * @Description: 
     * @author weiyb 
     * @date 2018年2月26日 下午4:38:16  
     */
    @Configuration
    public class TopicRabbitConfig {
    
    	public final static String QUEUE_NAME = "spring-boot-queue";
    	public final static String EXCHANGE_NAME = "spring-boot-exchange";
    	public final static String ROUTING_KEY = "spring-boot-key";
    
    	/**
    	 * 创建队列
    	 * @return
    	 * @author weiyb 
    	 */
    	@Bean("queueMessage")
    	public Queue queueMessage() {
    		return new Queue(QUEUE_NAME);
    	}
    
    
    	/**
    	 * 创建一个 topic 类型的交换器
    	 * @return
    	 * @author weiyb 
    	 */
    	@Bean("exchange")
    	TopicExchange exchange() {
    		return new TopicExchange(EXCHANGE_NAME);
    	}
    
    	/**
    	 * 使用路由键（routingKey）把队列（Queue）绑定到交换器（Exchange）
    	 * @param queueMessage
    	 * @param exchange
    	 * @return
    	 * @author weiyb 
    	 */
    	@Bean
    	Binding bindingExchangeMessage(@Qualifier("queueMessage") Queue queueMessage,@Qualifier("exchange") TopicExchange exchange) {
    		return BindingBuilder.bind(queueMessage).to(exchange).with(ROUTING_KEY);
    	}
    }

## 2.4 消费者（监听） ##

import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    import com.alibaba.fastjson.JSON;
    import com.spring.pro.config.exchange.TopicRabbitConfig;
    import com.spring.pro.config.exchange.transaction.TopicRabbitConfigTransaction;
    import com.spring.pro.model.User;
    
    @Component
    public class HelloReceiver {
    	private Logger logger = LoggerFactory.getLogger(getClass());
    
    	@RabbitListener(queues = TopicRabbitConfig.QUEUE_NAME)
    	public void process11(User user) {
    		logger.info("Receiver  process11: " + JSON.toJSONString(user));
    		// 这里抛出异常,测试消息重发机制
    //		throw new RuntimeException("***********");
    	}
    }

## 2.5 model ##

import java.io.Serializable;
    
    /**
     * @Title: User.java
     * @Package com.spring.pro.model
     * @Description:
     * @author ybwei
     * @date 2018年11月22日 下午2:46:10
     * @version V1.0
     */
    public class User implements Serializable {
    
    	private static final long serialVersionUID = 337531670671807745L;
    	private String id;
    	private String name;
    	private int age;
    
    	public User() {
    		super();
    	}
    
    	public User(String id, String name, int age) {
    		super();
    		this.id = id;
    		this.name = name;
    		this.age = age;
    	}
    
    	public String getId() {
    		return id;
    	}
    
    	public void setId(String id) {
    		this.id = id;
    	}
    
    	public String getName() {
    		return name;
    	}
    
    	public void setName(String name) {
    		this.name = name;
    	}
    
    	public int getAge() {
    		return age;
    	}
    
    	public void setAge(int age) {
    		this.age = age;
    	}
    }

## 2.6 生产者 ##

import java.util.Date;
    
    import org.springframework.amqp.core.AmqpTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    import com.spring.pro.config.exchange.TopicRabbitConfig;
    import com.spring.pro.model.User;
    
    @Component
    public class HelloSender {
    
    	@Autowired
    	private AmqpTemplate rabbitTemplate;
    
    	public void send1() {
    		User user=new User("1", "张三", 12);
    		this.rabbitTemplate.convertAndSend(TopicRabbitConfig.EXCHANGE_NAME, TopicRabbitConfig.ROUTING_KEY, user);
    	}
    
    }

## 2.7 测试 ##

import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    import org.springframework.test.context.junit4.SpringRunner;
    
    import com.spring.pro.send.HelloSender;
    
    @RunWith(SpringRunner.class)
    @SpringBootTest
    public class RabbitMqHelloTest {
    
    	@Autowired
    	private HelloSender helloSender;
    
    	@Test
    	public void hello1() throws Exception {
    		helloSender.send1();
    	}
    
    }

# 3、应用场景 #

1.  常用的mq通信。
2.  上游向下游回调通知策略。配置文件中的max-attempts（最大重试次数），multiplier（应用于前一重试间隔的乘法器）可以制定策略。

测试重试，第二种方式的日志如下：

[INFO ] 2018-11-22 18:03:51.853 [SimpleAsyncTaskExecutor-1] c.spring.pro.receiver.HelloReceiver - Receiver  process11: {"age":12,"id":"1","name":"张三"}
    [INFO ] 2018-11-22 18:03:56.881 [SimpleAsyncTaskExecutor-1] c.spring.pro.receiver.HelloReceiver - Receiver  process11: {"age":12,"id":"1","name":"张三"}
    [INFO ] 2018-11-22 18:04:21.893 [SimpleAsyncTaskExecutor-1] c.spring.pro.receiver.HelloReceiver - Receiver  process11: {"age":12,"id":"1","name":"张三"}
    [INFO ] 2018-11-22 18:06:26.907 [SimpleAsyncTaskExecutor-1] c.spring.pro.receiver.HelloReceiver - Receiver  process11: {"age":12,"id":"1","name":"张三"}

当前时间间隔=上次重试间隔\*multiplier。

当前时间间隔<max-interval（重试最大时间间隔）。