# PropertyOperationWorker.java

    package cloud.wconnect.rateshop.consumer.operation;

    import cloud.wconnect.commonlib.component.BeanMapper;
    import cloud.wconnect.commonlib.service.MailInternalService;
    import cloud.wconnect.commonlib.util.JsonUtils;
    import cloud.wconnect.commonlib.util.RateLimitUtils;
    import cloud.wconnect.rateshop.pojo.dto.rateshop.consumer.operation.PropertyOperationMessage;
    import cloud.wconnect.rateshop.pojo.dto.rateshop.operation.PropertyOperationDTO;
    import cloud.wconnect.rateshop.service.AmazonSqsService;
    import cloud.wconnect.rateshop.springlistener.PropertyOperationEvent;
    import com.fasterxml.jackson.databind.ObjectMapper;
    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;

    @Slf4j
    @Component
    public class PropertyOperationWorker {

        @Autowired
        private AmazonSqsService amazonSqsService;
        @Autowired
        private MailInternalService mailInternalService;
        @Autowired
        private ApplicationEventPublisher eventPublisher;
        @Autowired
        private ObjectMapper objectMapper;
        @Autowired
        private BeanMapper beanMapper;

        /**
         * 资产运营数据消息的队列url
         */
        @Value("${sqs.queue-url.property_operation_queue}")
        private String queueUrl;

        /**
         * 消费 rate shop 功能 资产运营数据 队列任务
         *
         * @author zc
         * @date 2024/3/26
         */
        @Scheduled(cron = "0 0/1 * * * ?", zone = "${default.timezone}")
        public void run() {
            List<Message> messages = amazonSqsService.getMessagesBatch(queueUrl);
            if (CollectionUtils.isNotEmpty(messages)) {
                for (Message message : messages) {
                    // message 为 json 字符串：
                    // 警告: 执行任务的具体方法应该使用新的线程，而不是在本线程执行（要确保本线程不会被阻塞或异常退出）

                    try {
                        PropertyOperationMessage propertyOperationMessage = JsonUtils.convertStringToObject(objectMapper,
                                message.body(), PropertyOperationMessage.class);

                        // 发布事件
                        eventPublisher.publishEvent(new PropertyOperationEvent(this,
                                beanMapper.map(propertyOperationMessage, PropertyOperationDTO.class)));
                    } catch (Exception e) {
                        log.error("SQS consume error[PropertyOperationWorker] - Message: {}; Exception: {}", message.body(), e.getMessage());
                        if (RateLimitUtils.tryAcquire(this)) {
                            sendAlertEmail(queueUrl, message.body(), e.getMessage());
                        }
                    } finally {
                        // 删除消息
                        amazonSqsService.deleteMessage(queueUrl, message.receiptHandle());
                    }
                }
                run();
            }
        }

        /**
         * 发送 alertEmail
         *
         * @param queueUrl  消息队列 url
         * @param message   消息
         * @param exception 异常信息
         * @author zc
         * @date 2024/3/25
         */
        public void sendAlertEmail(String queueUrl, String message, String exception) {
            String subject = "[Warning] - process PropertyOperationWorker";
            String mailContent = MessageFormat.format("SQS url:{0}<br/>Message:{1}<br/>Exception:{2}",
                    queueUrl, message, exception);
            mailInternalService.sendAlertEmail(subject, mailContent);
        }
    }