# HotelPriceWorker.java

    package cloud.wconnect.rateshop.consumer.hotelprice;

    import cloud.wconnect.commonlib.service.MailInternalService;
    import cloud.wconnect.commonlib.util.RateLimitUtils;
    import cloud.wconnect.rateshop.service.AmazonSqsService;
    import cloud.wconnect.rateshop.springlistener.HotelPriceMessagesEvent;
    import lombok.extern.slf4j.Slf4j;
    import org.apache.commons.collections.CollectionUtils;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.context.ApplicationEventPublisher;
    import org.springframework.scheduling.annotation.Scheduled;
    import org.springframework.stereotype.Component;
    import software.amazon.awssdk.services.sqs.model.Message;

    import java.text.MessageFormat;
    import java.util.List;

    /**
     * 抓取房价SQS消息
     *
     * @author lichen
     * @date 2024/3/25
     */
    @Slf4j
    @Component
    public class HotelPriceWorker {

        @Autowired
        private AmazonSqsService amazonSqsService;
        @Autowired
        private MailInternalService mailInternalService;
        @Autowired
        private ApplicationEventPublisher applicationEventPublisher;
        /**
         * 房价消息的队列url
         */
        @Value("${sqs.queue-url.google_hotel_queue}")
        private String hotelPriceUrl;

        /**
         * 执行抓取房价的定时任务
         * 执行时间：每10分钟执行一次
         *
         * @author lichen
         * @date 2024/3/25
         */
        @Scheduled(cron = "${schedule.cron.hotel-price-worker}")
        public void run() {
            List<Message> messages = amazonSqsService.getMessagesBatch(hotelPriceUrl);
            // 单线程消费多个消息，避免消耗过多线程资源，也可缓解死锁问题
            if (CollectionUtils.isNotEmpty(messages)) {
                try {
                    //发布消息
                    applicationEventPublisher.publishEvent(new HotelPriceMessagesEvent(this, messages));
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                    if (RateLimitUtils.tryAcquire(this)) {
                        sendAlertEmail(hotelPriceUrl, e);
                    }
                }finally{
                    //删除消息
                    for (Message message : messages) {
                        amazonSqsService.deleteMessage(hotelPriceUrl, message.receiptHandle());
                    }
                }
                //消息不为空的情况下，重复调用
                run();
            }
        }

        /**
         * 发送alertEmail
         *
         * @param hotelPriceUrl 定时任务调度URL
         * @param e             异常对象
         * @author lichen
         * @date 2024/3/25
         */
        public void sendAlertEmail(String hotelPriceUrl, Exception e) {
            //主题
            String subject = "[Warning] - process HotelPriceWorker";
            //邮件内容
            String mailContent = MessageFormat.format("SQS url:{0}<br/>Exception:{1}<br/>",
                    hotelPriceUrl, e.getMessage());
            //发送alert
            mailInternalService.sendAlertEmail(subject, mailContent);
        }

    }