# RevenueDayHotelPriceListener.java

    package cloud.wconnect.rateshop.springlistener.revenue;

    import cloud.wconnect.commonlib.cons.BaseConstant;
    import cloud.wconnect.commonlib.exception.ListenerException;
    import cloud.wconnect.commonlib.service.MailInternalService;
    import cloud.wconnect.commonlib.util.DateTimeUtils;
    import cloud.wconnect.commonlib.util.JsonUtils;
    import cloud.wconnect.commonlib.util.RateLimitUtils;
    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.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.stream.Collectors;

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

        @Autowired
        private MailInternalService mailInternalService;
        @Autowired
        private ObjectMapper objectMapper;
        @Autowired
        private RevenueDayHotelPriceHandle revenueDayHotelPriceHandle;
        @Autowired
        private RedissonClient redissonClient;
        /**
         * 锁前缀
         */
        private static final String KEY_PREFIX = "hotel-price:revenue-days:";
        /**
         * 最大自旋次数
         */
        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);

            //判断是否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;
            }

            //版本号(系统默认时区下的yyyyMMdd)
            String version = DateTimeUtils.timestampToStr(priceMessage.getQueryDate(), BaseConstant.DATE_FORMAT_YYYYMMDD, null);
            //并发key
            String key = MessageFormat.format("{0}{1}.{2}.{3}.{4}.{5}.{6}", KEY_PREFIX, priceMessage.getRateType(), priceMessage.getHotelId(),
                    priceMessage.getCurrencyId(), version, priceMessage.getDatasourceId(), priceMessage.getPos());
            //执行失败的次数
            int executionErrorTimes = 0;
            //加锁-自旋-处理，最多处理10次
            lockSpinProces(priceMessage, key, executionErrorTimes);
        }

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

            RLock lock = redissonClient.getLock(key);
            if (lock.tryLock()) {
                try {
                    //按天存储昨天到未来15天的房价
                    revenueDayHotelPriceHandle.handle(priceMessage);
                } finally {
                    lock.unlock();
                }
            } else {
                executionErrorTimes++;
                //休眠300ms
                Thread.sleep(300);
                //自旋
                lockSpinProces(priceMessage, key, executionErrorTimes);
            }
        }

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