SpringBoot学习－（二十七）SpringBoot整合RabbitMQ

Dear 丶 2022-05-25 08:00 266阅读 0赞

**基本步骤**

*  添加pom文件依赖
 *  修改application.properties配置文件
 *  新增rabbitmq配置
 *  创建消息生产者
 *  创建消息消费者
 *  测试

## 添加pom文件依赖 ##

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

## 修改application.properties配置文件 ##

server.port=8900
    
    spring.rabbitmq.host=localhost
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=guest
    spring.rabbitmq.password=guest

## 新增rabbitmq配置 ##

package com.ahut.config;
    
    import org.springframework.amqp.core.*;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Qualifier;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    /** * @author cheng * @className: RabbitMQConfig * @description: RabbitMQ配置 * @dateTime 2018/5/4 16:27 */
    @Configuration
    public class RabbitMQConfig { 
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        private String DIRECT_EXCHANGE = "exchange-direct-springboot";
    
        private String QUEUE_NAME_A = "queue-a-springboot";
    
        private String QUEUE_NAME_B = "queue-b-springboot";
    
        private String ROUTING_KEY_A = "routing-key-a-springboot";
    
        private String ROUTING_KEY_B = "routing-key-b-springboot";
    
        @Bean
        public AmqpTemplate amqpTemplate() {
            // 使用jackson 消息转换器
            rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
            rabbitTemplate.setEncoding("UTF-8");
            // 消息发送失败返回到队列中，yml需要配置 publisher-returns: true
            //rabbitTemplate.setMandatory(true);
            return rabbitTemplate;
        }
    
        @Bean("directExchange")
        public Exchange createExchange() {
            return ExchangeBuilder.directExchange(DIRECT_EXCHANGE).durable(true).build();
        }
    
        @Bean("queueA")
        public Queue createQueueA() {
            return QueueBuilder.durable(QUEUE_NAME_A).build();
        }
    
        @Bean("queueB")
        public Queue createQueueB() {
            return QueueBuilder.durable(QUEUE_NAME_B).build();
        }
    
        @Bean
        public Binding bindingQueueA(@Qualifier("queueA") Queue queue, @Qualifier("directExchange") Exchange exchange) {
            return BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY_A).noargs();
        }
    
        @Bean
        public Binding bindingQueueB(@Qualifier("queueB") Queue queue, @Qualifier("directExchange") Exchange exchange) {
            return BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY_B).noargs();
        }
    
    }

## 创建消息生产者 ##

package com.ahut.rabbitmq;
    
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.amqp.rabbit.support.CorrelationData;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Service;
    
    import java.util.UUID;
    
    /** * @author cheng * @className: SenderService * @description: 生产者 * @dateTime 2018/5/4 16:45 */
    @Service
    public class SenderService { 
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        private String DIRECT_EXCHANGE = "exchange-direct-springboot";
    
        private String ROUTING_KEY_A = "routing-key-a-springboot";
    
        private String ROUTING_KEY_B = "routing-key-b-springboot";
    
        public void sendMessage(String message) {
            CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
            rabbitTemplate.convertAndSend(DIRECT_EXCHANGE, ROUTING_KEY_A, message, correlationData);
            rabbitTemplate.convertAndSend(DIRECT_EXCHANGE, ROUTING_KEY_B, message, correlationData);
        }
    
    }

## 创建消息消费者 ##

package com.ahut.rabbitmq;
    
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    /** * @author cheng * @className: Receiver * @description: 消费者 * @dateTime 2018/5/4 16:52 */
    @Component
    public class Receiver { 
    
        @RabbitListener(queues = {
       "queue-a-springboot"})
        public void listenerQueueA(Message message) {
            System.out.println("================>队列queue-a-springboot接收消息:" + new String(message.getBody()));
        }
    
        @RabbitListener(queues = {
       "queue-b-springboot"})
        public void listenerQueueB(Message message) {
            System.out.println("================>队列queue-b-springboot接收消息:" + new String(message.getBody()));
        }
    
    }

## 测试 ##

package com.ahut.rabbitmq;
    
    import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    import org.springframework.test.context.junit4.SpringRunner;
    
    import javax.swing.*;
    
    /** * @author cheng * @className: RabbitMQTest * @description: * @dateTime 2018/5/4 17:00 */
    @SpringBootTest
    @RunWith(SpringRunner.class)
    public class RabbitMQTest { 
    
        @Autowired
        private SenderService senderService;
    
        @Test
        public void testSend() {
            String message = "springboot整合rabbitmq";
            senderService.sendMessage(message);
        }
    
    }