RabbitMQ与Spring Boot的集成

ゝ一世哀愁。 2022-09-01 00:57 221阅读 0赞

# 引入pom #

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

# 配置文件application.yaml #

server:
      port: 8080
    spring:
      #给项目来个名字
      application:
        name: rabbitmq-provider
      #配置rabbitMq 服务器
      rabbitmq:
        # host: 127.0.0.1
        # port: 5672
        username: root
        password: root
        #虚拟host 可以不设置,使用server默认host
        # virtual-host:
        addresses: 192.168.80.205:5672
        # 消息确认
        publisher-confirm-type: correlated
    
        # 消息手动确认
        publisher-returns: true
        template:
          mandatory: true

# 配置文件RabbitMqConfig.java #

package com.morris.rabbit.mq;
    
    import org.springframework.amqp.core.*;
    import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
    import org.springframework.amqp.rabbit.connection.ConnectionFactory;
    import org.springframework.amqp.rabbit.connection.CorrelationData;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
    import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    @Configuration
    public class RabbitMqConfig { 
    
        @Bean
        public RabbitListenerContainerFactory<?> rabbitListenerContainerFactory(ConnectionFactory connectionFactory) { 
            SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
            factory.setConnectionFactory(connectionFactory);
            factory.setMessageConverter(new Jackson2JsonMessageConverter()); // 接收消息的序列化
            return factory;
        }
    
        @Bean
        public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) { 
            RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
            rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter()); // 发送消息的序列化
            rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> { 
                System.out.println("ack: " + ack);
                System.out.println("cause: " + cause);
            });
            // 失败通知
            rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> { 
                // 没路由
                System.out.println("replyCode: " + replyCode);
                System.out.println("replyText: " + replyText);
            });
            rabbitTemplate.setMandatory(true); // 失败确认
            return rabbitTemplate;
        }
    
        @Bean
        public FanoutExchange testFanoutExchange() { 
            return new FanoutExchange("sendFanoutMessage",false,false);
        }
    
        @Bean
        public Queue testFanoutQueue() { 
            return new Queue("testFanoutQueue",false);
        }
    
        @Bean
        public Binding bindingTestFanoutQueueToTestFanoutExchange() { 
            return BindingBuilder.bind(testFanoutQueue()).to(testFanoutExchange());
        }
    
        @Bean
        public FanoutExchange testNoRouteExchange() { 
            return new FanoutExchange("testNoRouteExchange",false,false);
        }
    }

# 发送消息 #

package com.morris.rabbit.mq;
    
    import org.springframework.amqp.core.AmqpTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    @RestController
    @RequestMapping("mq")
    public class SendMessageController { 
    
        @Autowired
        private AmqpTemplate rabbitTemplate;  //使用RabbitTemplate,这提供了接收/发送等等方法
    
        @GetMapping("sendFanoutMessage")
        public String sendFanoutMessage() { 
            User user = new User().setName("sendFanoutMessage").setAddress("HongKong").setAge(18);
            rabbitTemplate.convertSendAndReceive("testFanoutExchange", "", user);
            return "sendFanoutMessage ok";
        }
    
        @GetMapping("sendNoRouteMessage")
        public String sendNoRouteMessage() { 
            User user = new User().setName("sendNoRoute").setAddress("HongKong").setAge(18);
            rabbitTemplate.convertSendAndReceive("testNoRouteExchange", "", user);
            return "sendNoRoute ok";
        }
    }

# 接收消息 #

package com.morris.rabbit.mq;
    
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    @Component
    public class ReceiveMessageService { 
    
        @RabbitListener(queues = "testFanoutQueue") // 监听的队列名称
        public void process(User user) { 
            System.out.println("消费者收到消息 : " + user);
        }
    
    }

# 消息重试的配置 #

spring:
        listener:
          type: simple
          simple:
            acknowledge-mode: manual
            retry:
              enabled: true # 开启重试机制
              max-attempts: 3 #最大重试传递次数
              initial-interval: 5000ms #第一次和第二次尝试传递消息的间隔时间 单位毫秒
              max-interval: 300000ms #最大重试时间间隔，单位毫秒
              multiplier: 3 #应用前一次重试间隔的乘法器，multiplier默认为1
            #以上配置的间隔0s 5s 15s 45s
            #重试次数超过上面的设置之后是否丢弃(消费者listener抛出异常，是否重回队列，默认true：重回队列， false为不重回队列(结合死信交换机))
            default-requeue-rejected: true