springboot 集成 rabbitmq 实战应用

雨点打透心脏的1/2处 2023-10-02 19:59 8阅读 0赞

springboot 集成 rabbitmq 实战应用今天分享：分为生产者和消费者两个角色

一、生产者

1、pom文件引入

<dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-amqp</artifactId>
            </dependency>

2、yml配置文件添加

spring:
      rabbitmq:
        host: 194.11.33.120
        port: 5672
        username: admin
        password: admin
        publisher-confirms: true
        publisher-returns: true
        virtual-host: /
        listener:
          simple:
            acknowledge-mode: manual

3、对象的初始化

import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.core.*;
    import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
    import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
    import org.springframework.amqp.rabbit.connection.ConnectionFactory;
    import org.springframework.amqp.rabbit.core.RabbitAdmin;
    import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
    import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    import java.util.List;
    
    /**
     *类说明：消息队列配置
     */
    @Slf4j
    @Configuration
    public class RabbitConfig {
    
        public final static String EXCHANGE_LOG = "order.log.producer.reply";
        public final static String KEY_LOG = "order.log.reply";
    
        @Value("${spring.rabbitmq.host}")
        private String addresses;
    
        @Value("${spring.rabbitmq.port}")
        private String port;
    
        @Value("${spring.rabbitmq.username}")
        private String username;
    
        @Value("${spring.rabbitmq.password}")
        private String password;
    
        @Value("${spring.rabbitmq.virtual-host}")
        private String virtualHost;
    
        @Value("${spring.rabbitmq.publisher-confirms}")
        private boolean publisherConfirms;
    
        @Value("${spring.rabbitmq.publisher-returns}")
        private boolean publisherReturns;
    
        @Bean
        public ConnectionFactory connectionFactory() {
            CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
            connectionFactory.setAddresses(addresses+":"+port);
            connectionFactory.setUsername(username);
            connectionFactory.setPassword(password);
            connectionFactory.setVirtualHost(virtualHost);
            /** 如果要进行消息回调，则这里必须要设置为true */
    //        connectionFactory.setPublisherConfirms(publisherConfirms);
    //        connectionFactory.setPublisherReturns(publisherReturns);
            return connectionFactory;
        }
    
        @Bean
        public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory){
            return new RabbitAdmin(connectionFactory);
        }
    
    
        @Bean
        public RabbitListenerContainerFactory<?> rabbitListenerContainerFactory(ConnectionFactory connectionFactory, List<SimpleMessageListenerContainer> list) {
            SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
            factory.setConnectionFactory(connectionFactory);
            factory.setConcurrentConsumers(2);
    //抓取参数非常关键，一次抓取的消息多了，消费速度一慢，就会造成响应延迟，抓取少了又会导致并发量低，消息堵塞
            factory.setPrefetchCount(10);
    
            /*
             * AcknowledgeMode.NONE：自动确认
             * AcknowledgeMode.AUTO：根据情况确认
             * AcknowledgeMode.MANUAL：手动确认
             */
            factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
    /*        factory.setDefaultRequeueRejected(false);
            factory.setAdviceChain(
                    RetryInterceptorBuilder
                            .stateless()
                            .recoverer(new RejectAndDontRequeueRecoverer())
                            .retryOperations(retryTemplate())
                            .build()
            );*/
            return factory;
        }
    
        @Bean
        public Queue queuelogMessage() {
            return new Queue("order.log.queue.reply");
        }
    
        @Bean
        public DirectExchange exchange() {
            return new DirectExchange(EXCHANGE_LOG);
        }
    
        @Bean
        public Binding bindingLogExchangeMessage() {
            return BindingBuilder
                    .bind(queuelogMessage())
                    .to(exchange())
                    .with(KEY_LOG);
        }
    }

4、发送信息的工具类

import cn.nan.config.RabbitConfig;
    import cn.nan.mall.model.MessageLog;
    import com.alibaba.fastjson.JSON;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.rabbit.connection.CorrelationData;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    /**
     * 类说明：
     */
    @Slf4j
    @Component
    public class OrderLogSender {
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        public void send(String actionId) {
            log.info("TopicSender send the 1st : " + actionId);
            this.rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_LOG, RabbitConfig.KEY_LOG, actionId);
        }
    
        public void send(MessageLog messageLog, CorrelationData correlationData) {
            String msg1 = JSON.toJSONString(messageLog);
            log.info("TopicSender send the 1st : " + msg1);
            this.rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_LOG, RabbitConfig.KEY_LOG, msg1,correlationData);
        }
    }

二、消费者

1、pom和yml配置同生产者一致

2、对象实例化:里面有消费者的配置，同时也有生产者的配置

import cn.nan.mall.model.MessageLog;
    import cn.nan.mall.service.MessageLogService;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.core.*;
    import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
    import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
    import org.springframework.amqp.rabbit.connection.ConnectionFactory;
    import org.springframework.amqp.rabbit.connection.CorrelationData;
    import org.springframework.amqp.rabbit.core.RabbitAdmin;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
    import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    import java.util.List;
    
    /**
     *类说明：消息队列配置
     */
    @Slf4j
    @Configuration
    public class RabbitConfig {
    
        public final static String EXCHANGE_LOG = "order.log.producer";
        public final static String KEY_LOG = "order.log";
    
        @Value("${spring.rabbitmq.host}")
        private String addresses;
    
        @Value("${spring.rabbitmq.port}")
        private String port;
    
        @Value("${spring.rabbitmq.username}")
        private String username;
    
        @Value("${spring.rabbitmq.password}")
        private String password;
    
        @Value("${spring.rabbitmq.virtual-host}")
        private String virtualHost;
    
        @Value("${spring.rabbitmq.publisher-confirms}")
        private boolean publisherConfirms;
    
        @Value("${spring.rabbitmq.publisher-returns}")
        private boolean publisherReturns;
    
        @Autowired
        private MessageLogService messageLogService;
    
        @Bean
        public ConnectionFactory connectionFactory() {
            CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
            connectionFactory.setAddresses(addresses+":"+port);
            connectionFactory.setUsername(username);
            connectionFactory.setPassword(password);
            connectionFactory.setVirtualHost(virtualHost);
            /** 如果要进行消息回调，则这里必须要设置为true */
            connectionFactory.setPublisherConfirms(publisherConfirms);
            connectionFactory.setPublisherReturns(publisherReturns);
            return connectionFactory;
        }
    
        @Bean
        public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory){
            return new RabbitAdmin(connectionFactory);
        }
    
        @Bean
        public RabbitTemplate newRabbitTemplate() {
            RabbitTemplate template = new RabbitTemplate(connectionFactory());
            template.setConfirmCallback((correlationData, ack, cause) -> {
                if (ack) {
                    log.info("消息成功发送到Exchange,messageId:" + correlationData.getId());
                    //修改日志表状态，状态改成 1，投递成功且未确认
                    if(updateMessageLog(1,Long.valueOf(correlationData.getId())) == 1) {
                        log.info("------modify status 1 ok--------");
                    }
                } else {
                    log.info("消息发送到Exchange失败, {}, cause: {}", correlationData, cause);
                }
            });
            template.setMandatory(true);
            template.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
                log.info("消息从Exchange路由到Queue失败: exchange: {}, route: {}, replyCode: {}, replyText: {}, message: {}", exchange, routingKey, replyCode, replyText, message);
            });
            //不使用临时队列
    //        template.setUseTemporaryReplyQueues(false);
    //        template.setReplyAddress("amq.rabbitmq.reply-to");
    //        template.setUserCorrelationId(true);
    //        template.setReplyTimeout(10000);
            return template;
        }
    
        private int updateMessageLog(int status,Long messageId) {
            MessageLog messageLog = new MessageLog();
            messageLog.setMessageId(messageId);
            messageLog.setStatus(status);
            messageLog.setTryCount(1);
            return messageLogService.updateMessageLog(messageLog);
        }
    
    
        @Bean
        public Queue queuelogMessage() {
            return new Queue("order.log.queue");
        }
    
        @Bean
        public DirectExchange exchange() {
            return new DirectExchange(EXCHANGE_LOG);
        }
    
        @Bean
        public Binding bindingLogExchangeMessage() {
            return BindingBuilder
                    .bind(queuelogMessage())
                    .to(exchange())
                    .with(KEY_LOG);
        }
    
        /**
         * 消费者配置
         * @param connectionFactory
         * @param list
         * @return
         */
        @Bean
        public RabbitListenerContainerFactory<?> rabbitListenerContainerFactory(ConnectionFactory connectionFactory, List<SimpleMessageListenerContainer> list) {
            SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
            factory.setConnectionFactory(connectionFactory);
            factory.setConcurrentConsumers(2);
    //抓取参数非常关键，一次抓取的消息多了，消费速度一慢，就会造成响应延迟，抓取少了又会导致并发量低，消息堵塞
            factory.setPrefetchCount(10);
    
            /*
             * AcknowledgeMode.NONE：自动确认
             * AcknowledgeMode.AUTO：根据情况确认
             * AcknowledgeMode.MANUAL：手动确认
             */
            factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
    /*        factory.setDefaultRequeueRejected(false);
            factory.setAdviceChain(
                    RetryInterceptorBuilder
                            .stateless()
                            .recoverer(new RejectAndDontRequeueRecoverer())
                            .retryOperations(retryTemplate())
                            .build()
            );*/
            return factory;
        }
    
        //===============生产者发送确认==========
        @Bean
        public RabbitTemplate.ConfirmCallback confirmCallback(){
            return new RabbitTemplate.ConfirmCallback(){
    
                @Override
                public void confirm(CorrelationData correlationData,
                                    boolean ack, String cause) {
                    if (ack) {
                        log.info("发送者确认发送给mq成功");
                    } else {
                        //处理失败的消息
                        log.info("发送者发送给mq失败,考虑重发:"+cause);
                    }
                }
            };
        }
    
        @Bean
        public RabbitTemplate.ReturnCallback returnCallback(){
            return new RabbitTemplate.ReturnCallback(){
    
                @Override
                public void returnedMessage(Message message,
                                            int replyCode,
                                            String replyText,
                                            String exchange,
                                            String routingKey) {
                    log.info("无法路由的消息，需要考虑另外处理。");
                    log.info("Returned replyText："+replyText);
                    log.info("Returned exchange："+exchange);
                    log.info("Returned routingKey："+routingKey);
                    String msgJson  = new String(message.getBody());
                    log.info("Returned Message："+msgJson);
                }
            };
        }
    }

3、监听消费

import cn.nan.mall.model.MessageLog;
    import cn.nan.mall.service.MessageLogService;
    import com.alibaba.fastjson.JSON;
    import com.rabbitmq.client.Channel;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.messaging.handler.annotation.Headers;
    import org.springframework.stereotype.Component;
    
    import java.util.Map;
    
    /**
     * 类说明：
     */
    @Slf4j
    @Component
    public class OrderLogReceiver {
    
        @Autowired
        private MessageLogService messageLogService;
    
        @RabbitListener(queues = "order.log.queue.reply")
        @RabbitHandler // 此注解加上之后可以接受对象型消息
        public void onMessage(Message message, Channel channel,@Headers Map<String, Object> headers) throws Exception {
            try {
                String msg = new String(message.getBody());
                log.info("OrderLogReceiver>>>>>>>message received:" + msg);
                MessageLog messageLog = JSON.parseObject(msg, MessageLog.class);
                try {
                    messageLog.setStatus(9);
                    messageLog.setTryCount(null);
                    messageLogService.updateMessageLog(messageLog);
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);//手工确认，可接下一条
                } catch (Exception e) {
                    e.printStackTrace();
                    log.info(e.getMessage());
                    channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);//失败，则直接忽略此订单
                    log.info("OrderLogReceiver>>>>>>message nack");
                    throw e;
                }
            } catch (Exception e) {
                log.info(e.getMessage());
            }
        }
    }

deliveryTag（唯一标识 ID）：当一个消费者向 RabbitMQ 注册后，会建立起一个 Channel ，Rab bitMQ 会用 basic.deliver 方法向消费者推送消息，这个方法携带了一个 delivery tag，  
它代表了 RabbitMQ 向该 Channel 投递的这条消息的唯一标识 ID，是一个单调递增的正整数，delivery tag 的范围仅限于 Channel  
multiple：为了减少网络流量，手动确认可以被批处理，当该参数为 true 时，则可以一次性确认 delivery\_tag 小于等于传入值的所有消息

到此、springboot 集成 rabbitmq 实战应用分享完毕，下篇会分享一些细节，顺序、重复消费、消息丢失等等，敬请期待！