rabbitmq系列（三）消息幂等性处理

分手后的思念是犯贱 2021-09-18 05:10 413阅读 0赞

## 一、springboot整合rabbitmq ##

1.  我们需要新建两个工程，一个作为生产者，另一个作为消费者。在pom.xml中添加amqp依赖：

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

1.  在application.yml文件中添加rabbitmq的相关信息：

spring:
      rabbitmq:
        # 连接地址
        host: 127.0.0.1
        # 端口
        port: 5672
        # 登录账号
        username: guest
        # 登录密码
        password: guest
        # 虚拟主机
        virtual-host: /

1.  在生产者工程中新建配置项rabbitmqConfig.java，申明名称为”byte-zb“直连交换机和队列，使用”byte-zb“的routing-key将队列和交换机绑定，代码如下：

@Configuration
    public class RabbitConfig { 
    
        public static final String QUEUE_NAME = "byte-zb";
    
        public static final String EXCHANGE_NAME = "byte-zb";
    
        public static final String ROUTING_KEY = "byte-zb";
    
        // 队列申明
        @Bean
        public Queue queue(){ 
            return new Queue(QUEUE_NAME);
        }
    
        // 申明交换机
        @Bean
        public DirectExchange directExchange(){ 
    
            return new DirectExchange(EXCHANGE_NAME);
        }
    
        // 数据绑定申明
        @Bean
        public Binding directBinding(){ 
    
            return BindingBuilder.bind(queue()).to(directExchange()).with(ROUTING_KEY);
        }
    }

1.  创建生产者发送一条消息，代码如下：

@RestController
    public class Producer { 
        public static final String QUEUE_NAME = "byte-zb";
    
        public static final String EXCHANGE_NAME = "byte-zb";
    
        @Autowired
        private AmqpTemplate amqpTemplate;
    
        @RequestMapping("/send")
        public void sendMessage(){ 
    
            JSONObject jsonObject = new JSONObject();
            jsonObject.put("email","11111111111");
            jsonObject.put("timestamp",System.currentTimeMillis());
            String json = jsonObject.toJSONString();
            System.out.println(json);
            amqpTemplate.convertAndSend(EXCHANGE_NAME,QUEUE_NAME,json);
        }
    
    }

1.  在消费者工程里创建消费者消费消息，代码如下：

@Component
    public class Consumer throws Exception{ 
    
        public static final String QUEUE_NAME = "byte-zb";
    
        @RabbitListener(queues = QUEUE_NAME)
        public void receiveMessage(String message){ 
    
            System.out.println("接收到的消息为" message);
        }
    }

我们启动生产者，然后请求send接口，然后打开rabbitmq控制台发现多了一个名为”byte-zb“的交换机和队列，并且队列中出现了一个未消费的消息，然后启动消费者，我们会在控制台上发现打印了一条消息，同时rabbitmq控制台中”byte-zb“的队列中消息没有了。

## 二、自动补偿机制 ##

如果消费者消息消费不成功的话，会出现什么情况呢？我们修改一下消费者代码，然后看看。

我们会看到消费者工程控制台一直在刷新报错，**当消费者配出异常，也就是说当消息消费不成功的话，该消息会存放在rabbitmq的服务端，一直进行重试，直到不抛出异常为止。**

如果一直抛异常，我们的服务很容易挂掉，那有没有办法控制重试几次不成功就不再重试了呢？答案是有的。我们在消费者application.yml中增加一段配置。

spring:
      rabbitmq:
        # 连接地址
        host: 127.0.0.1
        # 端口
        port: 5672
        # 登录账号
        username: guest
        # 登录密码
        password: guest
        # 虚拟主机
        virtual-host: /
        listener:
          simple:
            retry:
              enabled: true # 开启消费者进行重试
              max-attempts: 5 # 最大重试次数
              initial-interval: 3000 # 重试时间间隔

上面配置的意思是消费异常后，重试五次，每次隔3s。继续启动消费者看看效果，我们发现重试五次以后，就不再重试了。

## 三、结合实际案例来使用消息补偿机制 ##

像上面那种情况出现的异常其实不管怎么重试都不会成功，实际上用到消息补偿的就是调用第三方接口的这种。

案例：生者往队列中扔一条消息，包含邮箱和发送内容。消费者拿到消息后将调用邮件接口发送邮件。有时候可能邮件接口由于网络等原因不通，这时候就需要去重试了。

**在调用接口的工具类中，如果出现异常我们直接返回null**，工具类具体代码就不贴了，如果返回null之后怎么处理呢？我们只需要抛出异常，rabbitListener捕获到异常后就会自动重试。

我们改造一下消费者代码：

@Component
    public class Consumer { 
    
        public static final String QUEUE_NAME = "byte-zb";
    
        @RabbitListener(queues = QUEUE_NAME)
        public void receiveMessage(String message) throws Exception { 
    
            System.out.println("接收到的消息为" message);
            JSONObject jsonObject = JSONObject.parseObject(message);
            String email = jsonObject.getString("email");
            String content = jsonObject.getString("timestamp");
    
            String httpUrl = "http://127.0.0.1:8080/email?email" email "&content=" content;
            // 如果发生异常则返回null
            String body = HttpUtils.httpGet(httpUrl, "utf-8");
            //
            if(body == null){ 
                throw new Exception();
            }
        }
    }

当然我们可以自定义异常抛出。具体怎么试验呢，**第一步启动生产者和消费者，这时候我们发现消费者在重试，第二步我们启动邮件服务，这时候我们会发现邮件发送成功了，消费者不再重试了**。

## 四、解决消息幂等性问题 ##

一些刚接触java的同学可能对幂等性不太清楚。幂等性就是重复消费造成结果不一致。为了保证幂等性，因此消费者消费消息只能消费一次消息。我么可以是用全局的消息id来控制幂等性。当消息被消费了之后我们可以选择缓存保存这个消息id，然后当再次消费的时候，我们可以查询缓存，如果存在这个消息id，我们就不错处理直接return即可。先改造生产者代码，在消息中添加消息id：

@RequestMapping("/send")
        public void sendMessage(){ 
    
            JSONObject jsonObject = new JSONObject();
            jsonObject.put("email","11111111111");
            jsonObject.put("timestamp",System.currentTimeMillis());
            String json = jsonObject.toJSONString();
            System.out.println(json);
    
            	Message message = MessageBuilder.withBody(json.getBytes()).setContentType(MessageProperties.CONTENT_TYPE_JSON)
                    .setContentEncoding("UTF-8").setMessageId(UUID.randomUUID() "").build();
            amqpTemplate.convertAndSend(EXCHANGE_NAME,QUEUE_NAME,message);
        }

消费者代码改造：

@Component
    public class Consumer { 
    
        public static final String QUEUE_NAME = "byte-zb";
    
        @RabbitListener(queues = QUEUE_NAME)
        public void receiveMessage(Message message) throws Exception { 
    
            Jedis jedis = new Jedis("localhost", 6379);
    
            String messageId = message.getMessageProperties().getMessageId();
            String msg = new String(message.getBody(),"UTF-8");
            System.out.println("接收导的消息为：" msg "==消息id为：" messageId);
    
            String messageIdRedis = jedis.get("messageId");
    
            if(messageId == messageIdRedis){ 
                return;
            }
            JSONObject jsonObject = JSONObject.parseObject(msg);
            String email = jsonObject.getString("email");
            String content = jsonObject.getString("timestamp");
    
            String httpUrl = "http://127.0.0.1:8080/email?email" email "&content=" content;
            // 如果发生异常则返回null
            String body = HttpUtils.httpGet(httpUrl, "utf-8");
            //
            if(body == null){ 
                throw new Exception();
            }
            jedis.set("messageId",messageId);
        }
    }