SpringBoot_消息队列_RabbitMQ

叁歲伎倆 2021-10-06 03:40 550阅读 0赞

### SpringBoot\_消息队列\_RabbitMQ ###

*  依赖pom.xml
 *  配置文件bootstrap.yml
 *  配置源码
 *   *  一对一
     *  匹配模式
     *  广播模式

# 依赖pom.xml #

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

# 配置文件bootstrap.yml #

spring:
      rabbitmq:
        host: localhost
        port: 5672
        username: guest
        password: guest

# 配置源码 #

## 一对一 ##

（1）配置

@Configuration
    public class RabbitMQConfig {
        static final String QUEUE = "direct_queue";
        /**
         * Direct模式
         * @return
         */
        @Bean
        public Queue directQueue() {
            // 第一个参数是队列名字， 第二个参数是指是否持久化
            return new Queue(QUEUE, true);
        }
    }

（2）发送方

@Autowired
    private AmqpTemplate amqpTemplate;
     
    public void sendDirectQueue() {
        User user = new User();
        user.setUserId("1111");
        user.setName("1111");
        log.info("【sendDirectQueue已发送消息】");
         // 第一个参数是指要发送到哪个队列里面， 第二个参数是指要发送的内容
        this.amqpTemplate.convertAndSend(RabbitMQConfig.QUEUE, user);
    }

（3）消费方

@Component
    @Slf4j
    public class Receiver {
     
        // queues是指要监听的队列的名字
        @RabbitListener(queues = RabbitMQConfig.QUEUE)
        public void receiverDirectQueue(User user) {
            log.info("【receiverDirectQueue监听到消息】" + user.toString());
        }
    }

## 匹配模式 ##

（1）配置

@Configuration
    public class RabbitMQConfig {
        
        public static final String TOPIC_QUEUE1 = "topic.queue1";
        public static final String TOPIC_QUEUE2 = "topic.queue2";
        public static final String TOPIC_EXCHANGE = "topic.exchange";
     
        /**
         * Topic模式
         * @return
         */
        @Bean
        public Queue topicQueue1() {
            return new Queue(TOPIC_QUEUE1);
        }
        @Bean
        public Queue topicQueue2() {
            return new Queue(TOPIC_QUEUE2);
        }
        @Bean
        public TopicExchange topicExchange() {
            return new TopicExchange(TOPIC_EXCHANGE);
        }
        @Bean
        public Binding topicBinding1() {
            return BindingBuilder.bind(topicQueue1()).to(topicExchange()).with("pr.message");
        }
        @Bean //接受配置lzc.#规则的消息
        public Binding topicBinding2() {
            return BindingBuilder.bind(topicQueue2()).to(topicExchange()).with("pr.#");
        }
    }

（2）发送方

@Component
    @Slf4j
    public class Sender {
     
        @Autowired
        private AmqpTemplate amqpTemplate;
     
        public void sendTopic() {
            User user1 = new User();
            user1.setUserId("123456");
            user1.setName("111");
     
            User user2 = new User();
            user2.setUserId("456789");
            user2.setName("张三");
     
            log.info("【sendTopic已发送消息】");
            // 第一个参数：TopicExchange名字
            // 第二个参数：Route-Key  lzc.#队列的消费方会收到两条消息
            // 第三个参数：要发送的内容
            this.amqpTemplate.convertAndSend(RabbitMQConfig.TOPIC_EXCHANGE, "pr.message", user1 );
            this.amqpTemplate.convertAndSend(RabbitMQConfig.TOPIC_EXCHANGE, "pr.pr", user2);
        }
    }

（3）消费方

@Component
    @Slf4j
    public class Receiver {
        // queues是指要监听的队列的名字
        @RabbitListener(queues = RabbitMQConfig.TOPIC_QUEUE1)
        public void receiveTopic1(User user) {
            log.info("【receiveTopic1监听到消息】" + user.toString());
        }
        @RabbitListener(queues = RabbitMQConfig.TOPIC_QUEUE2)
        public void receiveTopic2(User user) {
            log.info("【receiveTopic2监听到消息】" + user.toString());
        }
     }

## 广播模式 ##

（1）配置

@Configuration
    public class RabbitMQConfig {   
        public static final String TOPIC_QUEUE1 = "topic.queue1";
        public static final String TOPIC_QUEUE2 = "topic.queue2";
        public static final String FANOUT_EXCHANGE = "fanout.exchange";
        /**
         * 这里为了方便就没有创建新的队列了，直接使用topic时所创建的队列。
         * @return
         */
        @Bean
        public Queue topicQueue1() {
            return new Queue(TOPIC_QUEUE1);
        }
        @Bean
        public Queue topicQueue2() {
            return new Queue(TOPIC_QUEUE2);
        }
        /**
         * Fanout模式
         * Fanout 就是我们熟悉的广播模式或者订阅模式，给Fanout交换机发送消息，绑定了这个交换机的所有队列都收到这个消息。
         * @return
         */
        @Bean
        public FanoutExchange fanoutExchange() {
            return new FanoutExchange(FANOUT_EXCHANGE);
        }
        @Bean
        public Binding fanoutBinding1() {
            return BindingBuilder.bind(topicQueue1()).to(fanoutExchange());
        }
        @Bean
        public Binding fanoutBinding2() {
            return BindingBuilder.bind(topicQueue2()).to(fanoutExchange());
        }
    }

（2）发送方

@Component
    @Slf4j
    public class Sender {
     
        @Autowired
        private AmqpTemplate amqpTemplate;
     
        public void sendFanout() {
            User user = new User();
            user.setUserId("123456");
            user.setName("111");
            log.info("【sendFanout已发送消息】");
            // 注意， 这里的第2个参数为空。
            // 因为fanout 交换器不处理路由键，只是简单的将队列绑定到交换器上，
            // 每个发送到交换器的消息都会被转发到与该交换器绑定的所有队列上
            this.amqpTemplate.convertAndSend(RabbitMQConfig.FANOUT_EXCHANGE, "", user );
        }
     }

（3）消费方

@Component
    @Slf4j
    public class Receiver {
     
        // queues是指要监听的队列的名字
        @RabbitListener(queues = RabbitMQConfig.TOPIC_QUEUE1)
        public void receiveTopic1(User user) {
            log.info("【receiveTopic1监听到消息】" + user.toString());
        }
        @RabbitListener(queues = RabbitMQConfig.TOPIC_QUEUE2)
        public void receiveTopic2(User user) {
            log.info("【receiveTopic2监听到消息】" + user.toString());
        }
     }