# CancellationFlightsServiceImpl.java

    package cloud.wconnect.rateshop.service.rateshop.impl;

    import cloud.wconnect.commonlib.util.DateTimeUtils;
    import cloud.wconnect.rateshop.entity.rateshop.AirportSourceUrl;
    import cloud.wconnect.rateshop.entity.rateshop.CancellationFlights;
    import cloud.wconnect.rateshop.mapper.rateshop.CancellationFlightsMapper;
    import cloud.wconnect.rateshop.pojo.dto.rateshop.consumer.airport.CancellationFlightsMessage;
    import cloud.wconnect.rateshop.service.rateshop.IAirportSourceUrlService;
    import cloud.wconnect.rateshop.service.rateshop.ICancellationFlightsService;
    import com.baomidou.dynamic.datasource.annotation.DS;
    import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
    import com.google.common.collect.Lists;
    import org.apache.commons.collections.CollectionUtils;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Service;
    import org.springframework.transaction.annotation.Transactional;

    import java.util.List;
    import java.util.Map;
    import java.util.Objects;
    import java.util.Set;
    import java.util.function.Function;
    import java.util.stream.Collectors;

    /**
     * <p>
     * 机场取消航班数量 服务实现类
     * </p>
     *
     * @author zc
     * @since 2024-03-23
     */
    @Service
    @DS("rate_shop")
    public class CancellationFlightsServiceImpl extends ServiceImpl<CancellationFlightsMapper, CancellationFlights> implements ICancellationFlightsService {

        @Autowired
        private IAirportSourceUrlService sourceUrlService;

        /**
         * 处理机场取消航班数量数据
         *
         * @param messageList 消息数据
         * @author zc
         * @date 2024/3/26
         */
        @Override
        @Transactional(rollbackFor = Exception.class)
        public void handleMessages(List<CancellationFlightsMessage> messageList) {
            if (CollectionUtils.isEmpty(messageList)) {
                return;
            }

            // 获取机场源 url 数据
            Set<Integer> sourceUrlIds = messageList.stream()
                    .map(CancellationFlightsMessage::getSourceUrlId).collect(Collectors.toSet());
            Map<Integer, AirportSourceUrl> urlId2SourceUrl = sourceUrlService.listByIds(sourceUrlIds).stream()
                    .collect(Collectors.toMap(AirportSourceUrl::getId, Function.identity()));

            long nowTimestamp = DateTimeUtils.nowTimestamp();
            List<CancellationFlights> cancellationFlightsList = Lists.newArrayList();
            for (CancellationFlightsMessage message : messageList) {
                int urlId = message.getSourceUrlId();
                AirportSourceUrl sourceUrl = urlId2SourceUrl.get(urlId);
                if (Objects.isNull(sourceUrl)) {
                    continue;
                }

                CancellationFlights cancellationFlights = new CancellationFlights();
                cancellationFlights.setTheDate(message.getTheDate());
                cancellationFlights.setUrlId(urlId);
                cancellationFlights.setUrl(sourceUrl.getUrl());
                cancellationFlights.setSymbol(sourceUrl.getSymbol());
                cancellationFlights.setCancellationFlights(message.getCancellationFlights());
                cancellationFlights.setUpdatedTime(nowTimestamp);

                cancellationFlightsList.add(cancellationFlights);
            }
            if (CollectionUtils.isNotEmpty(cancellationFlightsList)) {
                // 保存消息数据
                baseMapper.insertOrUpdateBatchByUnique(cancellationFlightsList);
            }
        }
    }