springboot 整合 rabbitmq

灰太狼 2022-06-03 08:55 92阅读 0赞

**pom.xml**

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
    
        <groupId>com.example</groupId>
        <artifactId>springboot</artifactId>
        <version>0.0.1-SNAPSHOT</version>
        <packaging>jar</packaging>
    
        <name>springboot</name>
        <description>Demo project for Spring Boot</description>
    
        <parent>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-parent</artifactId>
            <version>1.5.6.RELEASE</version>
            <relativePath /> 
        </parent>
    
        <properties>
            <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
            <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
            <java.version>1.8</java.version>
            <spring.version>4.3.13.RELEASE</spring.version>
        </properties>
    
        <dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
            </dependency>
    
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-test</artifactId>
                <scope>test</scope>
            </dependency>
            
            <dependency>
                <groupId>org.mybatis.spring.boot</groupId>
                <artifactId>mybatis-spring-boot-starter</artifactId>
                <version>1.3.0</version>
            </dependency>
            
            <dependency>
                <groupId>com.google.code.gson</groupId>
                <artifactId>gson</artifactId>
                <version>2.8.0</version>
            </dependency>
            
            <dependency>
                <groupId>mysql</groupId>
                <artifactId>mysql-connector-java</artifactId>
                <version>5.1.35</version>
            </dependency>
            
            <dependency>
                <groupId>com.alibaba</groupId>
                <artifactId>druid</artifactId>
                <version>1.0.26</version>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-amqp</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-configuration-processor</artifactId>
                <optional>true</optional>
            </dependency>
            <dependency>
                <groupId>org.springframework.retry</groupId>
                <artifactId>spring-retry</artifactId>
                <version>1.2.2.RELEASE</version>
            </dependency>
            <dependency>
                <groupId>org.quartz-scheduler</groupId>
                <artifactId>quartz</artifactId>
                <version>2.2.1</version>
            </dependency>
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-context-support</artifactId>
                <version>${spring.version}</version>
            </dependency>
        </dependencies>
    
        <build>
            <plugins>
                <plugin>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-maven-plugin</artifactId>
                </plugin>
            </plugins>
        </build>
    
    
    </project>

我刚开始启动项目时报错，最后发现是由于缺少了jar包以及jar包版本不一致,我使用的是springboot 1.5.6,如果和我一样可以参考上面的，我本机测试是没问题的。不一样的话，可以自己去百度查看你使用的springboot版本对应的jar包版本号。

**application.yml**

server:
      context-path: /spring-boot
      port: 8082
    spring:
      rabbitmq:
        host: 部署了rabbitmq的主机ip
        port: 5672
        username: qinwei
        password: 123456
        virtual-host: /
        publisherConfirms: true
        listener:
          simple:
            acknowledge-mode: manual
            concurrency: 5
            max-concurrency: 10
        cache:
          channel:
            size: 50
    mq:
      topicExchange: spring-exchange
      queue: spring_queue
      routeKey: spring_key
      myQueue: rabbit_queue
      myRouteKey: rabbit_key

**MQProperties.java**

@Component
    @ConfigurationProperties(prefix = "mq")
    public class MQProperties { 
    
        private String topicExchange;
        private String routeKey;
        private String queue;
        private String myQueue;
        private String myRouteKey;
    
        public String getMyQueue() {
            return myQueue;
        }
    
        public void setMyQueue(String myQueue) {
            this.myQueue = myQueue;
        }
    
        public String getMyRouteKey() {
            return myRouteKey;
        }
    
        public void setMyRouteKey(String myRouteKey) {
            this.myRouteKey = myRouteKey;
        }
    
        public String getTopicExchange() {
            return topicExchange;
        }
    
        public void setTopicExchange(String topicExchange) {
            this.topicExchange = topicExchange;
        }
    
        public String getRouteKey() {
            return routeKey;
        }
    
        public void setRouteKey(String routeKey) {
            this.routeKey = routeKey;
        }
    
        public String getQueue() {
            return queue;
        }
    
        public void setQueue(String queue) {
            this.queue = queue;
        }
    }

**RabbitMQConfig.java**

@Component
    @EnableRabbit
    public class RabbitMQConfig { 
    
        @Autowired
        private MQProperties mqProperties;
    
        @Autowired
        private CachingConnectionFactory connectionFactory;
        /*这里的connectionFactory不需要手动创建，因为org.springframework.boot.autoconfigure.amqp.RabbitAutoConfiguration.RabbitConnectionFactoryCreator已经创建好了，只要你在application.yml中添加了相应的配置就可以了*/
    
        @Bean
        @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
        public RabbitTemplate rabbitTemplate() {
            return new RabbitTemplate(connectionFactory);
        }
    
        @Bean
        public Queue queue1() {
            return new Queue(mqProperties.getQueue());
        }
    
        @Bean
        public Queue queue2() {
            return new Queue(mqProperties.getMyQueue());
        }
    
        @Bean
        public TopicExchange topicExchange() {
            return new TopicExchange(mqProperties.getTopicExchange(), true, false);
        }
    
        @Bean
        public Binding binding1() {
            // *表示一个单词,#表示零个或多个单词
            return BindingBuilder.bind(queue1()).to(topicExchange()).with(mqProperties.getRouteKey() + ".*");
        }
    
        @Bean
        public Binding binding2() {
            return BindingBuilder.bind(queue2()).to(topicExchange()).with(mqProperties.getMyRouteKey() + ".*");
        }
    }

生产者 **Sender.java**

@Component
    public class Sender implements RabbitTemplate.ConfirmCallback { 
    
        private RabbitTemplate rabbitTemplate;
    
        @Autowired
        private MQProperties mqProperties;
    
        public Sender(RabbitTemplate rabbitTemplate) {
            this.rabbitTemplate = rabbitTemplate;
            rabbitTemplate.setConfirmCallback(this);
        }
    
        public void send1(String message) {
            System.out.println("Sender1 : " + message);
            rabbitTemplate.convertAndSend(mqProperties.getTopicExchange(), mqProperties.getRouteKey() + ".springboot",
                    message, new CorrelationData(UUID.randomUUID().toString()));
        }
    
        public void send2(String message) {
            System.out.println("Sender2 : " + message);
            rabbitTemplate.convertAndSend(mqProperties.getTopicExchange(), mqProperties.getMyRouteKey() + ".qinwei",
                    message, new CorrelationData(UUID.randomUUID().toString()));
        }
    
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            System.out.println(" 回调id:" + correlationData);
            if (ack) {
                System.out.println("消息成功消费");
            } else {
                System.out.println("消息消费失败:" + cause);
            }
        }
    }

消费者 **Receiver.java**

@Component
    public class Receiver {
    
        @RabbitListener(queues = "${mq.queue}")
        public void receive1(Message message, Channel channel, String msg) {
            System.out.println("消费内容为1: " + new String(message.getBody()));
            try {
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    
        @RabbitListener(queues = "${mq.myQueue}")
        public void receive2(Message message, Channel channel, String msg) {
            System.out.println("消费内容为2: " + msg);
            try {
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

测试类 **RabbitMqTest .java**

@RestController
    public class RabbitMqTest { 
    
        @Autowired
        private Sender sender;
    
        @RequestMapping(value = "/rabbitmq/{message}", method = RequestMethod.GET)
        public String hello(@PathVariable("message") String message) {
            SimpleDateFormat sdf = new SimpleDateFormat("yyy-MM-dd HH:mm:ss.SSS");
            String context = message + " " + sdf.format(new Date());
            /*如果message为qinwei1则只发送到queue1；qinwei2只发送到queue2；qinwei12则同时发送到2个queue*/
            if (message.contains("1")) {
                sender.send1(context);
            }
            if (message.contains("2")) {
                sender.send2(context);
            }
            return context;
        }
    }