Rabbit MQ 发消息例子以及发失败后定时重发

超、凢脫俗 2022-04-01 06:14 179阅读 0赞

第一步，创建自己交换器名称，交换器模式等信息

import org.slf4j.Logger;  
import org.springframework.amqp.core.AcknowledgeMode;  
import org.springframework.amqp.core.FanoutExchange;  
import org.springframework.amqp.core.TopicExchange;  
import org.springframework.amqp.rabbit.annotation.EnableRabbit;  
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;  
import org.springframework.amqp.rabbit.connection.ConnectionFactory;  
import org.springframework.amqp.rabbit.core.RabbitTemplate;  
import org.springframework.amqp.rabbit.listener.ConditionalRejectingErrorHandler;  
import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException;  
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;  
import org.springframework.amqp.support.converter.MessageConverter;  
import org.springframework.context.annotation.Bean;  
import org.springframework.context.annotation.Configuration;  
import org.springframework.context.annotation.Primary;  
import org.springframework.util.ErrorHandler;  
@Configuration  
@EnableRabbit  
public class EventConfig \{

/\*\*  
\* 交换器名称  
\*/  
public static final String exchangeName = “aaa.bbb.ccc.ddd”;

@Bean(name = exchangeName)
    FanoutExchange exchangeName() {
        return new FanoutExchange(exchangeName);
    }
    
    @Primary
    @Bean("exchangeName")
    RabbitTemplate exchangeNameTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
        return rabbitTemplate;
    }

package xxx.xxx.xxx.xx.x.xxxxx.x.xxx.xx.x.x.mq;
    
    import .....
    import lombok.extern.slf4j.Slf4j;
    import org.apache.curator.framework.recipes.leader.LeaderLatch;
    import org.aspectj.lang.annotation.AfterReturning;
    import org.aspectj.lang.annotation.Aspect;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.amqp.rabbit.support.CorrelationData;
    import org.springframework.beans.factory.annotation.Qualifier;
    import org.springframework.core.annotation.Order;
    import org.springframework.stereotype.Component;
    
    import java.util.ArrayList;
    import java.util.Collection;
    import java.util.List;
    import java.util.concurrent.ConcurrentLinkedQueue;
    import java.util.concurrent.ExecutorService;
    
    import static java.util.concurrent.Executors.newSingleThreadExecutor;
    
    
    @Aspect
    @Component
    @Slf4j
    @Order(100)
    public class TestRecordResultEventInterceptor {
        private final ExecutorService executorService = newSingleThreadExecutor();
        private final LeaderLatch leaderLatch;
        //消息列队模板
        private final RabbitTemplate rabbitTemplate;
    
        private final TestReturnRepository testReturnRepository;
    
    
    
        /**
         * 一个内存队列
         */
        private final ConcurrentLinkedQueue<TestRecordSendMqInfo> eventsToSend;
    
        public TestRecordResultEventInterceptor(LeaderLatch leaderLatch, TestReturnRepository testReturnRepository, @Qualifier("exchangeMaterialTestRecordResult") RabbitTemplate rabbitTemplate) {
            this.leaderLatch = leaderLatch;
            this.testReturnRepository = testReturnRepository;
            this.eventsToSend = new ConcurrentLinkedQueue<>();
            this.rabbitTemplate = rabbitTemplate;
    
            rabbitTemplate.setConfirmCallback((CorrelationData correlationData, boolean ack, String cause) -> {
                if (!ack) {
                    log.error("failed to asn sendMqInfo: " + correlationData + ". " + cause);
                    return;
                }
    //            markSuccess(correlationData.getId());
            });
        }
    
        public void addEvents(Collection<TestRecordSendMqInfo> newEvents) {
            this.eventsToSend.addAll(newEvents);
        }
    
        public void addEvent(TestRecordSendMqInfo newEvent) {
            this.eventsToSend.add(newEvent);
        }
    
    
        @AfterReturning(value = "execution(* com.lenovo.chinaservice.pscm..*.*(..)) && @annotation(com.lenovo.chinaservice.pscm.quality.test.applyresponse.model.annotation.SendMQNeed) ")
        private void sendEvent() {
            while (!eventsToSend.isEmpty()) {
                final TestRecordSendMqInfo event = eventsToSend.poll();
                executorService.submit(new SendEventTask(rabbitTemplate, event));
            }
        }
    
        private  class SendEventTask<T> implements Runnable {
            private RabbitTemplate rabbitTemplate;
            private TestRecordSendMqInfo sendMqInfo;
    
            public SendEventTask(RabbitTemplate rabbitTemplate, TestRecordSendMqInfo event) {
                this.rabbitTemplate = rabbitTemplate;
                this.sendMqInfo = event;
            }
    
            @Override
            public void run() {
                if(sendMqInfo == null){
                    log.error("when execute send mq info is null");
                    return;
                }
                try {
                    log.info("send mq info [{}]", sendMqInfo);
    
                    rabbitTemplate.convertAndSend(EventConfig.exchangeMaterialTestRecordResult, null, sendMqInfo, new CorrelationData(sendMqInfo.getUniqueKey().toString()));
                } catch (Exception e) {
                    log.error("send mq fail",e);
                }
            }
        }
    
    
        private void markSuccess(String correlationDataId) {
    //        List<Long> list = new ArrayList<>();
    //        list.add(Long.valueOf(correlationDataId));
    //
    //
    //        List<TestReturnSnBaseVO> snBaseVOList = testReturnRepository.querySnByIdList(list);
    //        if (snBaseVOList == null) {
    //            log.error("no record for[{}]", correlationDataId);
    //            return;
    //        }
    //        TestReturnSnBaseVO testReturnSnBaseVO = snBaseVOList.get(0);
    //        if ((ApplyResponseMQStatusEnum.YES.getStatus()).equals(testReturnSnBaseVO.getResponseStatus())) {
    //            log.info("test return sn  ="+testReturnSnBaseVO.getAssetSN() + " already send ");
    //            return;
    //        }
    //        testReturnSnBaseVO.setResponseStatus(ApplyResponseMQStatusEnum.YES.getStatus());
    //        testReturnRepository.updateResponseStatus(testReturnSnBaseVO);
        }
    
    //
        /**
         * 把没有发送成功的重新加入队列
         */
        @Scheduled(cron = "0 0/10 * * * ?")
        public void sendApplyReponseMQTask() {
            if (!leaderLatch.hasLeadership()) {
                log.info("MQScheduled not leader");
                return;
            }
    
            MaterialApply materialApply = new MaterialApply();
            materialApply.setStatus(ApplyResponseStatusEnum.REPLIED.getStatus());
            materialApply.setResponseStatus(ApplyResponseMQStatusEnum.NO.getStatus());
            List<MaterialApply> materialApplyList = applyResponseRepository.queryMaterialApplyListByParameter(materialApply);
            if (CollectionUtils.isEmpty(materialApplyList)) {
                log.info(" not exist no send MQ info");
                return;
            }
            for (final MaterialApply record : materialApplyList) {
    
                TestRecordSendMqInfo TestRecordSendMqInfo = new TestRecordSendMqInfo();
                TestRecordSendMqInfo.setId(record.getId());
                TestRecordSendMqInfo.setExchangeName(EventConfig.exchangeMaterialResponse);
                TestRecordSendMqInfo.setMaterialNo(record.getReplyMaterialNo());
                TestRecordSendMqInfo.setMaterialName(record.getReplyMaterialName());
                TestRecordSendMqInfo.setPrCode(record.getPrCode());
                TestRecordSendMqInfo.setMemo(record.getRemark());
                executorService.submit(new SendEventTask(rabbitTemplate, TestRecordSendMqInfo));
            }
        }
    }