# RateShopHotelPriceListener.java

    package cloud.wconnect.rateshop.springlistener.rateshop;

    import cloud.wconnect.commonlib.exception.ListenerException;
    import cloud.wconnect.commonlib.service.MailInternalService;
    import cloud.wconnect.commonlib.util.JsonUtils;
    import cloud.wconnect.commonlib.util.RateLimitUtils;
    import cloud.wconnect.rateshop.entity.rateshop.RoomInfo;
    import cloud.wconnect.rateshop.mapper.rateshop.RoomInfoMapper;
    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.pojo.dto.rateshop.consumer.hotelprice.RoomDTO;
    import cloud.wconnect.rateshop.springlistener.HotelPriceMessagesEvent;
    import com.fasterxml.jackson.databind.ObjectMapper;
    import com.google.common.collect.Lists;
    import jakarta.annotation.Resource;
    import lombok.extern.slf4j.Slf4j;
    import org.apache.commons.codec.digest.DigestUtils;
    import org.apache.commons.collections.CollectionUtils;
    import org.apache.commons.lang3.StringUtils;
    import org.redisson.api.RLock;
    import org.redisson.api.RedissonClient;
    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.*;
    import java.util.stream.Collectors;

    import static java.util.stream.Collectors.toSet;

    /**
     * rate shop模块处理房价消息的监听器
     *
     * @author lichen
     * @date 2024/3/25
     */
    @Component
    @Slf4j
    public class RateShopHotelPriceListener {

        @Resource
        private RateShopHotelPriceHandle rateShopHotelPriceHandle;
        @Resource
        private MailInternalService mailInternalService;
        @Resource
        private ObjectMapper objectMapper;
        @Resource
        private RedissonClient redissonClient;
        @Resource
        private RoomInfoMapper roomInfoMapper;

        /**
         * 锁前缀
         */
        private static final String KEY_PREFIX = "hotel-price:rate-shop:";
        private static final String ROOM_KEY_PREFIX = "hotel-price:rate-shop:room:";
        /**
         * 最大自旋次数
         */
        private static final int MAXIMUM = 5;

        /**
         * 处理房价消息的监听器
         *
         * @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/5/14
         */
        private void consumeMessage(Message message) throws InterruptedException {
            //把消息内容转化成PriceMessage
            PriceMessage priceMessage = JsonUtils.convertStringToObject(objectMapper, message.body(), PriceMessage.class);
            //判断是否NoData
            List<PlatformDTO> platformDTOList = priceMessage.getData().stream()
                    .flatMap(priceDTO -> priceDTO.getPlatform().stream()).toList();
            if (CollectionUtils.isEmpty(platformDTOList)) {
                return;
            }

            //处理房价时需要对room_info，latest_rate_YYYY进行并发操作,把这两块的资源上锁
            //过滤出需要批量新增的房型数据
            List<RoomInfo> saveRoomInfos = this.filterSaveRoomInfos(priceMessage);
            //加锁的redis key集合
            List<String> roomInfoKeys = saveRoomInfos.stream().map(roomInfo -> {
                //酒店ID + 平台ID + 销售地区 + 房型 签名
                String sign = DigestUtils.sha256Hex(roomInfo.generateKey());
                return ROOM_KEY_PREFIX + sign;
            }).sorted().toList();

            //入住日期集合
            List<Long> checkInDateList = priceMessage.getData().stream().map(PriceDTO::getCheckInDate)
                    .distinct().sorted().toList();
            //加锁的redis key集合
            List<String> rateKeys = checkInDateList.stream().map(checkInDate ->
                    //并发key
                    MessageFormat.format("{0}{1}.{2}.{3}.{4}.{5}.{6}", KEY_PREFIX, priceMessage.getRateType(), priceMessage.getHotelId(),
                            priceMessage.getCurrencyId(), checkInDate, priceMessage.getDatasourceId(), priceMessage.getPos())
            ).toList();

            //全部的key集合
            List<String> keys = Lists.newArrayListWithCapacity(roomInfoKeys.size() + rateKeys.size());
            keys.addAll(roomInfoKeys);
            keys.addAll(rateKeys);

            //锁集合
            RLock[] locks = keys.stream().map(redissonClient::getLock).toList().toArray(new RLock[keys.size()]);

            //加锁-自旋-处理
            lockSpinProcess(priceMessage, saveRoomInfos, locks, 0);
        }

        /**
         * 过滤出新增的房型集合
         *
         * @param priceMessage 房价消息
         * @return java.util.List<cloud.wconnect.rateshop.entity.rateshop.RoomInfo>
         * @author lichen
         * @date 2024/4/3
         */
        private List<RoomInfo> filterSaveRoomInfos(PriceMessage priceMessage) {
            //酒店ID
            Integer hotelId = priceMessage.getHotelId();

            List<PlatformDTO> platformDTOList = priceMessage.getData().stream().flatMap(priceDTO -> priceDTO.getPlatform().stream()).toList();
            //去除房型的空格
            platformDTOList.forEach(platformDTO -> {
                List<RoomDTO> room = platformDTO.getRoom();
                if (CollectionUtils.isNotEmpty(room)) {
                    room.forEach(roomDTO -> {
                        String title = roomDTO.getTitle();
                        if (StringUtils.isNotBlank(title)) {
                            roomDTO.setTitle(title.trim());
                        }
                    });
                }
            });
            //平台ID关联的房型集合
            Map<Integer, Set<String>> platformId2RoomTitle = platformDTOList.stream()
                    .collect(Collectors.toMap(
                            PlatformDTO::getId,
                            p -> p.getRoom().stream().map(RoomDTO::getTitle).collect(toSet()),
                            (v1, v2) -> {
                                v1.addAll(v2);
                                return v1;
                            })
                    );
            //消息中的roomInfo信息
            List<RoomInfo> roomInfosByMsg = new ArrayList<>();

            //platformId2RoomTitle -> roomInfos
            platformId2RoomTitle.forEach((platformId, titles) -> titles.forEach(title -> {
                if (StringUtils.isNotBlank(title)) {
                    roomInfosByMsg.add(new RoomInfo()
                            .setPlatformId(platformId)
                            .setRoom(title)
                            .setHotelId(hotelId)
                            .setPos(priceMessage.getPos()));
                }
            }));

            if (CollectionUtils.isEmpty(roomInfosByMsg)) {
                return Collections.emptyList();
            }

            //根据platformId和room先查询出数据库已经存在的房型
            List<RoomInfo> alreadyExistRoomInfo = roomInfoMapper.selectByPlatformIdAndRoomList(hotelId, roomInfosByMsg);
            List<String> alreadyExistRoomInfoKey = Collections.emptyList();
            if (CollectionUtils.isNotEmpty(alreadyExistRoomInfo)) {
                alreadyExistRoomInfoKey = alreadyExistRoomInfo.stream().map(RoomInfo::generateKey).toList();
            }
            List<String> finalAlreadyExistRoomInfoKey = alreadyExistRoomInfoKey;

            //筛选出需要新增的房型
            return roomInfosByMsg.stream().filter(roomInfo -> {
                String key = roomInfo.generateKey();
                return !finalAlreadyExistRoomInfoKey.contains(key);
            }).collect(
                    Collectors.collectingAndThen(
                            Collectors.toCollection(() -> new TreeSet<>(
                                    Comparator.comparing(RoomInfo::generateKey))), ArrayList::new
                    )
            );
        }

        /**
         * 加锁-自旋-处理，最多处理10次
         *
         * @param priceMessage        房价消息
         * @param saveRoomInfos       新增的房型集合
         * @param locks               锁集合
         * @param executionErrorTimes 执行失败的次数
         * @author lichen
         * @date 2024/4/1
         */
        private void lockSpinProcess(PriceMessage priceMessage, List<RoomInfo> saveRoomInfos, RLock[] locks, int executionErrorTimes) throws InterruptedException {
            //判断执行失败的次数是否达到最大次数
            if (executionErrorTimes > MAXIMUM) {
                throw new ListenerException("rate shop handle hotel price error, The number of execution failures reached the upper limit");
            }

            RLock multiLock = redissonClient.getMultiLock(locks);
            //加锁，默认过期时间是30秒。默认每隔5秒左右自动续期至30秒
            if (multiLock.tryLock()) {
                try {
                    rateShopHotelPriceHandle.handle(priceMessage, saveRoomInfos);
                } finally {
                    //解锁。非当前线程拥有的者进行解锁，会抛出一个异常IllegalMonitorStateException
                    multiLock.unlock();
                }
            } else {
                //休眠300ms
                Thread.sleep(300);
                //自旋
                lockSpinProcess(priceMessage, saveRoomInfos, locks, ++executionErrorTimes);
            }
        }

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