黄埔班第44节：消息队列应用实战-5

import com.rabbitmq.client.Channel;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;
import java.util.Map;

@Component
public class OrderReceiver {
    @RabbitListener(queues = "order-queue-1")
    @RabbitHandler
    public void onOrderMessage(@Payload String orderInfo, @Headers Map<String,Object> headers, Channel channel) throws Exception{
        System.out.println("************消息接收开始***********");
        System.out.println("Order Info: "+orderInfo);
        Long deliverTag = (Long)headers.get(AmqpHeaders.DELIVERY_TAG);
        channel.basicAck(deliverTag,false); // 确认签收
    }
  
    @RabbitListener(queues = "order-queue-2")
    @RabbitHandler
    public void onOrderMessage(@Payload String orderInfo, @Headers Map<String,Object> headers, Channel channel) throws Exception{
        System.out.println("************消息接收开始***********");
        System.out.println("Order Info: "+orderInfo);
        Long deliverTag = (Long)headers.get(AmqpHeaders.DELIVERY_TAG);
        channel.basicAck(deliverTag,false); // 确认签收 
    }
}

@Autowired
RabbitTemplate rabbitTemplate;

public void sendOrder(String orderInfo,String messageId){
  CorrelationData correlationData = new CorrelationData();
  correlationData.setId(messageId);
  rabbitTemplate.convertAndSend("order-exchange-1","order.1.key",orderInfo,correlationData);
}

@Test
void testJson() {
  String orderInfo = "消息队列RabbitMQ从入门到精通,order-exchange-1");
  String messageId = "MS9999904-1";
  try {
    System.out.println("***********开始发送************");
    orderSender.sendOrder(orderInfo,messageId);
    System.out.println("-----------发送完成------------");
  }catch (Exception ex){
    ex.printStackTrace();
  }
}

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class OrderSender {
  @Autowired
  RabbitTemplate rabbitTemplate;

  public void sendOrder(String orderInfo,String message_id) throws Exception{
    /**
         * exchange: 交换机名字
         * routingkey: 队列关联的key
         * object: 要传输的消息对象
         * correlationData: 消息的唯一id
         */
    CorrelationData correlationData = new CorrelationData();
    correlationData.setId(message_id);
    // 设置callback内容
    rabbitTemplate.setConfirmCallback(confirmCallback);
    rabbitTemplate.setReturnCallback(returnCallback);
    rabbitTemplate.convertAndSend("order-exchange-2","order.3.key",orderInfo,correlationData);
  } 
  // exchange返回调用
  // 只有exchange交换机是正确存在的，消息都能发送到rabbitmq mq，routing key 错误了，消息就寻找不到正确的queue,消息就丢失了，但是对发送端来说消息是成功发送到MQ的，这不是异常，消息是先到达exchange层，再路由到queue，延时投递也是基于这个机制
  // 总结：confirmCallback 只判断到exchange层
  final ConfirmCallback confirmCallback = new ConfirmCallback() {
    @Override
    public void confirm(CorrelationData correlationData, boolean b, String s) {
      System.out.println("==========消息publish成功的ID: " + correlationData.getId());
      System.out.println("==========消息是否发布成功: " + b);
      System.out.println("==========失败的异常信息: " + s);
    }
  };

  // 判断层到达routingkey,只有在exchange正确，routingkey不正确的情况下才会触发改方法
  final ReturnCallback returnCallback = new ReturnCallback() {
    @Override
    public void returnedMessage(Message message, int replayCode, String replayText, String exchange, String routingkey) {
      System.out.println("******replayCode: "+replayCode);
      System.out.println("******replayText: "+replayText);
      System.out.println("******exchange: "+exchange);
      System.out.println("******routingkey: "+routingkey);
    }
  };
}

// 注意，ConfirmCallback和ReturnCallback的包都是org.springframework.amqp.rabbit.core.RabbitTemplate

@Data
@EqualsAndHashCode(callSuper = false)
@Accessors(chain = true)
public class MessageRetryDTO implements Serializable {
    private static final long serialVersionUID = 1L;

    /**
     * 原始消息body
     */
    private String bodyMsg;

    /**
     * 消息来源ID
     */
    private String sourceId;

    /**
     * 消息来源描述
     */
    private String sourceDesc;

    /**
     * 交换器
     */
    private String exchangeName;

    /**
     * 路由键
     */
    private String routingKey;

    /**
     * 队列
     */
    private String queueName;

    /**
     * 状态,1:初始化，2：成功，3：失败
     */
    private Integer status = 1;

    /**
     * 最大重试次数
     */
    private Integer maxTryCount = 3;

    /**
     * 当前重试次数
     */
    private Integer currentRetryCount = 0;

    /**
     * 重试时间间隔（毫秒）
     */
    private Long retryIntervalTime = 0L;

    /**
     * 任务失败信息
     */
    private String errorMsg;

    /**
     * 创建时间
     */
    private Date createTime;

    /**
     * 检查重试次数是否超过最大值
     *
     * @return
     */
    public boolean checkRetryCount() {
        retryCountCalculate();
        //检查重试次数是否超过最大值
        if (this.currentRetryCount < this.maxTryCount) {
            return true;
        }
        return false;
    }

    /**
     * 重新计算重试次数
     */
    private void retryCountCalculate() {
        this.currentRetryCount = this.currentRetryCount + 1;
    }
}

public abstract class CommonMessageRetryService {

    private static final Logger log = LoggerFactory.getLogger(CommonMessageRetryService.class);

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private MongoTemplate mongoTemplate;


    /**
     * 初始化消息
     *
     * @param message
     */
    public void initMessage(Message message) {
        log.info("{} 收到消息: {}，业务数据：{}", this.getClass().getName(), message.toString(), new String(message.getBody()));
        try {
            //封装消息
            MessageRetryDTO messageRetryDto = buildMessageRetryInfo(message);
            if (log.isInfoEnabled()) {
                log.info("反序列化消息:{}", messageRetryDto.toString());
            }
            prepareAction(messageRetryDto);
        } catch (Exception e) {
            log.warn("处理消息异常，错误信息：", e);
        }
    }

    /**
     * 准备执行
     *
     * @param retryDto
     */
    protected void prepareAction(MessageRetryDTO retryDto) {
        try {
            execute(retryDto);
            doSuccessCallBack(retryDto);
        } catch (Exception e) {
            log.error("当前任务执行异常，业务数据：" + retryDto.toString(), e);
            //执行失败，计算是否还需要继续重试
            if (retryDto.checkRetryCount()) {
                if (log.isInfoEnabled()) {
                    log.info("重试消息:{}", retryDto.toString());
                }
                retrySend(retryDto);
            } else {
                if (log.isWarnEnabled()) {
                    log.warn("当前任务重试次数已经到达最大次数，业务数据：" + retryDto.toString(), e);
                }
                doFailCallBack(retryDto.setErrorMsg(e.getMessage()));
            }
        }
    }

    /**
     * 任务执行成功，回调服务(根据需要进行重写)
     *
     * @param messageRetryDto
     */
    private void doSuccessCallBack(MessageRetryDTO messageRetryDto) {
        try {
            successCallback(messageRetryDto);
        } catch (Exception e) {
            log.warn("执行成功回调异常，队列描述：{}，错误原因：{}", messageRetryDto.getSourceDesc(), e.getMessage());
        }
    }

    /**
     * 任务执行失败，回调服务(根据需要进行重写)
     *
     * @param messageRetryDto
     */
    private void doFailCallBack(MessageRetryDTO messageRetryDto) {
        try {
            saveMessageRetryInfo(messageRetryDto.setErrorMsg(messageRetryDto.getErrorMsg()));
            failCallback(messageRetryDto);
        } catch (Exception e) {
            log.warn("执行失败回调异常，队列描述：{}，错误原因：{}", messageRetryDto.getSourceDesc(), e.getMessage());
        }
    }

    /**
     * 执行任务
     *
     * @param messageRetryDto
     */
    protected abstract void execute(MessageRetryDTO messageRetryDto);

    /**
     * 成功回调
     *
     * @param messageRetryDto
     */
    protected abstract void successCallback(MessageRetryDTO messageRetryDto);

    /**
     * 失败回调
     *
     * @param messageRetryDto
     */
    protected abstract void failCallback(MessageRetryDTO messageRetryDto);

    /**
     * 构建消息补偿实体
     * @param message
     * @return
     */
    private MessageRetryDTO buildMessageRetryInfo(Message message){
        //如果头部包含补偿消息实体，直接返回
        Map<String, Object> messageHeaders = message.getMessageProperties().getHeaders();
        if(messageHeaders.containsKey("message_retry_info")){
            Object retryMsg = messageHeaders.get("message_retry_info");
            if(Objects.nonNull(retryMsg)){
                return JSONObject.parseObject(String.valueOf(retryMsg), MessageRetryDTO.class);
            }
        }
        //自动将业务消息加入补偿实体
        MessageRetryDTO messageRetryDto = new MessageRetryDTO();
        messageRetryDto.setBodyMsg(new String(message.getBody(), StandardCharsets.UTF_8));
        messageRetryDto.setExchangeName(message.getMessageProperties().getReceivedExchange());
        messageRetryDto.setRoutingKey(message.getMessageProperties().getReceivedRoutingKey());
        messageRetryDto.setQueueName(message.getMessageProperties().getConsumerQueue());
        messageRetryDto.setCreateTime(new Date());
        return messageRetryDto;
    }

    /**
     * 异常消息重新入库
     * @param retryDto
     */
    private void retrySend(MessageRetryDTO retryDto){
        //将补偿消息实体放入头部，原始消息内容保持不变
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setContentType(MessageProperties.CONTENT_TYPE_JSON);
        messageProperties.setHeader("message_retry_info", JSONObject.toJSON(retryDto));
        Message message = new Message(retryDto.getBodyMsg().getBytes(), messageProperties);
        rabbitTemplate.convertAndSend(retryDto.getExchangeName(), retryDto.getRoutingKey(), message);
    }

    /**
     * 将异常消息存储到mongodb中
     * @param retryDto
     */
    private void saveMessageRetryInfo(MessageRetryDTO retryDto){
        try {
            mongoTemplate.save(retryDto, "message_retry_info");
        } catch (Exception e){
            log.error("将异常消息存储到mongodb失败，消息数据：" + retryDto.toString(), e);
        }
    }
}

@Component
public class OrderServiceListener extends CommonMessageRetryService {

    private static final Logger log = LoggerFactory.getLogger(OrderServiceListener.class);

    /**
     * 监听订单系统下单成功消息
     * @param message
     */
    @RabbitListener(queues = "mq.order.add")
    public void consume(Message message) {
        log.info("收到订单下单成功消息: {}", message.toString());
        super.initMessage(message);
    }


    @Override
    protected void execute(MessageRetryDTO messageRetryDto) {
        //调用扣减库存服务，将业务异常抛出来
    }

    @Override
    protected void successCallback(MessageRetryDTO messageRetryDto) {
        //业务处理成功，回调
    }

    @Override
    protected void failCallback(MessageRetryDTO messageRetryDto) {
        //业务处理失败，回调
    }
}