# RevenueHourHotelPriceListener.java

    package cloud.wconnect.rateshop.springlistener.revenue;

    import cloud.wconnect.commonlib.cons.BaseConstant;
    import cloud.wconnect.commonlib.exception.ListenerException;
    import cloud.wconnect.commonlib.pojo.dto.PropertyDTO;
    import cloud.wconnect.commonlib.service.MailInternalService;
    import cloud.wconnect.commonlib.service.PropertyInternalService;
    import cloud.wconnect.commonlib.util.DateTimeUtils;
    import cloud.wconnect.commonlib.util.JsonUtils;
    import cloud.wconnect.commonlib.util.RateLimitUtils;
    import cloud.wconnect.rateshop.cons.rateshop.HotelTypeEnum;
    import cloud.wconnect.rateshop.cons.rateshop.RateTypeEnum;
    import cloud.wconnect.rateshop.entity.rateshop.PropertyHotel;
    import cloud.wconnect.rateshop.pojo.dto.rateshop.consumer.hotelprice.PlatformDTO;
    import cloud.wconnect.rateshop.pojo.dto.rateshop.consumer.hotelprice.PriceDTO;
    import cloud.wconnect.rateshop.pojo.dto.rateshop.consumer.hotelprice.PriceMessage;
    import cloud.wconnect.rateshop.service.rateshop.IPropertyHotelService;
    import cloud.wconnect.rateshop.springlistener.HotelPriceMessagesEvent;
    import com.fasterxml.jackson.databind.ObjectMapper;
    import lombok.extern.slf4j.Slf4j;
    import org.apache.commons.collections.CollectionUtils;
    import org.redisson.api.RLock;
    import org.redisson.api.RedissonClient;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.context.event.EventListener;
    import org.springframework.stereotype.Component;
    import software.amazon.awssdk.services.sqs.model.Message;

    import java.text.MessageFormat;
    import java.util.ArrayList;
    import java.util.Iterator;
    import java.util.List;
    import java.util.Optional;
    import java.util.stream.Collectors;

    /**
     * Revenue 功能处理每小时房价监听器
     *
     * @author lichen
     * @date 2024/3/25
     */
    @Component
    @Slf4j
    public class RevenueHourHotelPriceListener {

        @Autowired
        private MailInternalService mailInternalService;
        @Autowired
        private ObjectMapper objectMapper;
        @Autowired
        private IPropertyHotelService propertyHotelService;
        @Autowired
        private RevenueHourHotelPriceHandle revenueHourHotelPriceHandle;
        @Autowired
        private RedissonClient redissonClient;
        @Autowired
        private PropertyInternalService propertyInternalService;
        /**
         * 锁前缀
         */
        private static final String KEY_PREFIX = "hotel-price:revenue-hour:";
        /**
         * 最大自旋次数
         */
        private static final int MAXIMUM = 10;

        /**
         * 处理房价消息的监听器
         *
         * @author lichen
         * @date 2024/3/25
         */
        @EventListener(classes = {HotelPriceMessagesEvent.class})
        public void processHotelPriceMessage(HotelPriceMessagesEvent hotelPriceMessagesEvent) throws InterruptedException {
            ArrayList<Message> messages = new ArrayList<>(hotelPriceMessagesEvent.getPayLoad());
            int retryCount = 2;
            Exception exception = null;
            // 用于DB死锁后重试
            while (retryCount > 0) {
                try {
                    Iterator<Message> it = messages.iterator();
                    while (it.hasNext()) {
                        Message message = it.next();
                        consumeMessage(message);
                        it.remove();
                    }
                    return;
                } catch (Exception e) {
                    log.error(e.getMessage(), e);

                    if (e instanceof InterruptedException) {
                        throw e;
                    }

                    exception = e;
                    retryCount--;
                    Thread.sleep(300);
                }
            }

            if (retryCount == 0) {
                if (RateLimitUtils.tryAcquire(this)) {
                    sendAlertEmail(messages.stream().map(Message::body).collect(Collectors.joining(" | ")), exception);
                }
            }
        }

        /**
         * 消费消息
         *
         * @param message 单个消息对象
         * @author lichen
         * @date 2024/4/19
         */
        private void consumeMessage(Message message) throws InterruptedException {
            //把消息内容转化成PriceMessage
            PriceMessage priceMessage = JsonUtils.convertStringToObject(objectMapper, message.body(), PriceMessage.class);

            //判断是否BAR房价
            if (priceMessage.getRateType() != RateTypeEnum.BEST_AVAILABLE_RATE.getRateType()) return;

            //判断是否NoData
            List<PriceDTO> data = priceMessage.getData();
            List<PlatformDTO> platformDTOList = data.stream().flatMap(priceDTO -> priceDTO.getPlatform().stream()).collect(Collectors.toList());
            if (CollectionUtils.isEmpty(platformDTOList)) {
                //消息中的平台数据都是空，表示No Data，不用持久化入库
                return;
            }
            //查询酒店关联的资产信息
            List<PropertyHotel> propertyHotels = propertyHotelService.getByHotelIdAndCurrencyIdAndPos(priceMessage.getHotelId(),
                    priceMessage.getCurrencyId(), priceMessage.getPos());
            //是否有资产把酒店设置为自身酒店
            Optional<PropertyHotel> optional = propertyHotels.stream().filter(propertyHotel -> propertyHotel.getType() == HotelTypeEnum.OWN.getType()).findFirst();
            if (optional.isEmpty()) return;

            //查询资产信息
            PropertyDTO propertyDTO = propertyInternalService.getOne(optional.get().getPropertyId());

            //并发key
            String key = MessageFormat.format("{0}{1}.{2}.{3}.{4}.{5}.{6}", KEY_PREFIX, priceMessage.getRateType(), priceMessage.getHotelId(), priceMessage.getCurrencyId(),
                    DateTimeUtils.timestampToStr(priceMessage.getQueryDate(), BaseConstant.DATE_FORMAT_YYYYMMDD, propertyDTO.getTimezone()),
                    priceMessage.getDatasourceId(), priceMessage.getPos());

            //执行失败的次数
            int executionErrorTimes = 0;
            //加锁-自旋-处理，最多处理10次
            lockSpinProces(priceMessage, propertyDTO, key, executionErrorTimes);
        }

        /**
         * 加锁-自旋-处理，最多处理10次
         *
         * @param priceMessage        房价消息
         * @param propertyDTO         资产DTO
         * @param key                 redis的key
         * @param executionErrorTimes 执行失败的次数
         * @author lichen
         * @date 2024/4/2
         */
        private void lockSpinProces(PriceMessage priceMessage, PropertyDTO propertyDTO, String key, int executionErrorTimes) throws InterruptedException {
            //判断执行失败的次数是否达到最大次数
            if (executionErrorTimes > MAXIMUM) {
                throw new ListenerException("revenue handle hour hotel price error, The number of execution failures reached the upper limit");
            }

            RLock lock = redissonClient.getLock(key);
            if (lock.tryLock()) {
                try {
                    revenueHourHotelPriceHandle.handle(priceMessage, propertyDTO);
                } finally {
                    lock.unlock();
                }
            } else {
                executionErrorTimes++;
                //休眠300ms
                Thread.sleep(300);
                //自旋
                lockSpinProces(priceMessage, propertyDTO, key, executionErrorTimes);
            }
        }

        /**
         * 发送alertEmail
         *
         * @param messageBody 消息
         * @param e           异常对象
         * @author lichen
         * @date 2024/3/25
         */
        public void sendAlertEmail(String messageBody, Exception e) {
            //主题
            String subject = "[Warning] - Revenue hour hotel price listener process error";
            //邮件内容
            String mailContent = MessageFormat.format("Message:{0}<br/>Exception:{1}<br/>",
                    messageBody, e.getMessage());
            //发送alert
            mailInternalService.sendAlertEmail(subject, mailContent);
        }
    }