博客

一：消费者出现业务逻辑异常，自动补偿机制

问题还原 ：消费者程序出现异常，服务器一直报错

问题原因

rabbitmq 默认情况下 如果消费者程序出现异常的情况下，会自动实现补偿机制

补偿（重试机制） 队列服务器 发送补偿请求

@RabbitListener 底层 使用Aop进行拦截，如果程序没有抛出异常，自动提交事务

如果Aop使用异常通知拦截 获取异常信息的话，自动实现补偿机制 ，该消息会缓存到rabbitmq服务器端进行存放，一直重试到不抛异常为准。

问题解决

配置RabbitMQ

配置最大重试次数为5

重试间隔时间3000

spring:
  rabbitmq:
  ####连接地址
    host: 127.0.0.1
   ####端口号   
    port: 5672
   ####账号 
    username: guest
   ####密码  
    password: guest
   ### 地址
    virtual-host: /admin_host
    listener:
      simple:
        retry:
        ####开启消费者（程序出现异常的情况下会）进行重试
          enabled: true
         ####最大重试次数
          max-attempts: 5
        ####重试间隔次数
          initial-interval: 3000

配置RabbitMQ之后的效果

思考：如何选择重试机制

情况1: 消费者获取到消息后，调用第三方接口，但接口暂时无法访问，是否需要重试?

情况2: 消费者获取到消息后，抛出数据转换异常，是否需要重试?

总结：

对于情况2，如果消费者代码抛出异常是需要发布新版本才能解决的问题，那么不需要重试，重试也无济于事。应该采用日志记录+定时任务+人工进行补偿

思考：消费出现异常或者是消费延迟消费，会造成MQ进行重试补偿，在重试过程中，可能会造成重复消费。

消费者如何保证消息幂等性，不被重复消费

解决办法:

①使用全局MessageID判断消费方使用同一个，解决幂等性。

②或者使用业务逻辑保证唯一（比如订单号码）

①使用全局MessageID判断消费方使用同一个，解决幂等性。

发送消息

   public void send(String queueName) {

        JSONObject jsonObject = new JSONObject();
        jsonObject.put("email", "644064779");
        jsonObject.put("timestamp", System.currentTimeMillis());
        String jsonString = jsonObject.toJSONString();
        System.out.println("jsonString:" + jsonString);

        // 生产者发送消息的时候需要设置消息id
        Message message = MessageBuilder.withBody(jsonString.getBytes())

                .setContentType(MessageProperties.CONTENT_TYPE_JSON).setContentEncoding("utf-8")

                .setMessageId(UUID.randomUUID() + "").build();

        amqpTemplate.convertAndSend(queueName, message);
    }

接收消息

@RabbitListener(queues = "fanout_email_queue")
	public void process(Message message) throws Exception {

		String messageId = message.getMessageProperties().getMessageId();

		String msg = new String(message.getBody(), "UTF-8");

		System.out.println("消费者获取生产者消息msg:" + msg + ",消息id:" + messageId);

		JSONObject jsonObject = JSONObject.parseObject(msg);

		String email = jsonObject.getString("email");

		String emailUrl = "http://127.0.0.1:8083/sendEmail?email=" + email;

		System.out.println("消费者开始调用第三方服务器,emailUrl:" + emailUrl);

		JSONObject result = HttpClientUtils.httpGet(emailUrl);
		// 如果调用第三方接口无法访问，如何实现自动重试.
		if (result == null) {
			throw new Exception("调用第三方邮件服务器接口失败!");
		}
		System.out.println("消费者结束调用第三方服务器成功,result:" + result + "程序执行结束");
	}