21.Spring-Boot中RabbitMQ使用

女爷i 2022-05-29 05:51 138阅读 0赞

注意：默认账号guest具有所有操作的权限，并且出于安全的考虑，guest用户只能通过localhost登录使用，如果想通过远程ip连接，需要重庆创建新的用户。

模式使用的就是direct交换器模式

**一：Direct交换器使用**

1.配置pom包，主要是添加spring-boot-starter-amqp的支持

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

2.在application.properties中配置rabbit连接参数

spring.rabbitmq.host=11.12.112.159
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=admin
    spring.rabbitmq.password=123456

3.队列交换器配置

package com.niugang;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.amqp.core.Binding;
    import org.springframework.amqp.core.BindingBuilder;
    import org.springframework.amqp.core.DirectExchange;
    import org.springframework.amqp.core.Queue;
    @Configuration
    public class RabbitDirectConfig {
    //创建队列
    @Bean
    public Queue queue() {
    // 创建一个队列路由键为hello
    return new Queue("hello");
    }
        //创建交换器
    @Bean
    public DirectExchange directExchange() {
    return new DirectExchange("direct-exchange");
    }
          //绑定交换器和路由键
    @Bean
    public Binding bindingExchange() {
    return BindingBuilder.bind(queue()).to(directExchange()).with("hello");
    }
    }

4.创建消息生成者（发送者）

package com.niugang.mq.send;
    import org.springframework.amqp.core.AmqpTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    @Component
    public class HelloSender {
    @Autowired
    private AmqpTemplate rabbitTemplate;
    public void send(String context) {
    System.out.println("Sender : " + context);
    //通过源码默认的交换器模式为direct
    //绑定路由键hello
    this.rabbitTemplate.convertAndSend("direct-exchange","hello", context);
    }
    }

5.创建消费者（接收者）

**接收者1**

package com.niugang.mq.receiver;
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    @Component
    //@RabbitListener监听具体的队列或绑定
    @RabbitListener(queues = "hello")
    public class HelloReceiver1 {
        @RabbitHandler
        public void process(String hello) {
            System.out.println("Receiver1  : " + hello);
        }
    }

**接收者2**

package com.niugang.mq.receiver;
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    @Component
    //@RabbitListener监听具体的队列或绑定
    @RabbitListener(queues = "hello")
    public class HelloReceiver2 {
        @RabbitHandler
        public void process(String hello) { 
        System.out.println("Receiver2  : " + hello);
        }
    }

6.测试

package com.niugang.test;
    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 com.niugang.mq.send.HelloSender;
    
    @RunWith(SpringRunner.class)
    @SpringBootTest
    public class RabbitTest {
    @Autowired
    private HelloSender helloSender;
    @Test
    public void hello() throws Exception {
    for (int i = 0; i <= 10; i++) {
    helloSender.send("hello"+i);
    Thread.sleep(500);
    }
    }
    }

7.运行结果

![format_png][]

对于Direct模式的交换器，当队列拥有多个消费者时，队列收到的消息将以循环的方式发送给消费者。每条消息只能发送给一个订阅的消费者。

二：Fanout交换器的使用

1.2步同上

3.配置交换器和队列

package com.niugang;
    
    
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.amqp.core.Binding;
    import org.springframework.amqp.core.BindingBuilder;
    import org.springframework.amqp.core.FanoutExchange;
    import org.springframework.amqp.core.Queue;
    @Configuration
    public class RabbitFanoutConfig {
    //创建队列
    @Bean
    public Queue queueA() {
    return new Queue("queue.A");
    }
    @Bean
    public Queue queueB() {
    return new Queue("queue.B");
    }
    @Bean
    public Queue queueC() {
    return new Queue("queue.C");
    }
    
           //创建交换器
    @Bean
    public FanoutExchange fanoutExchange() {
    return new FanoutExchange("fanout-exchange");
    }
          //绑定交换器和路由键
    @Bean
    public Binding bindingExchangeA() {
    return BindingBuilder.bind(queueA()).to(fanoutExchange());
    }
    @Bean
    public Binding bindingExchangeB() {
    return BindingBuilder.bind(queueB()).to(fanoutExchange());
    }
    @Bean
    public Binding bindingExchangeC() {
    return BindingBuilder.bind(queueC()).to(fanoutExchange());
    }
    }

4.创建消息生成者（发送者）

package com.niugang.mq.send;
    import org.springframework.amqp.core.AmqpTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    @Component
    public class FanoutSender {
    @Autowired
    private AmqpTemplate rabbitTemplate;
    public void send(String context) {
    System.out.println("Sender : " + context);
        //因为fanout交换器模式是广播消息,所系这块队列只需要给个空字符串
    this.rabbitTemplate.convertAndSend("fanout-exchange","", context);
    }
    }

5.创建消费者（接收者）

消费者A:

package com.niugang.mq.receiver;
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    @Component
    //@RabbitListener监听具体的队列或绑定
    @RabbitListener(queues = "queue.A")
    public class FanoutReceiverA {
        @RabbitHandler
        public void process(String hello) {
            System.out.println("queue.A : " + hello);
        }
    }

消费者B:

package com.niugang.mq.receiver;
    
    
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    
    @Component
    //@RabbitListener监听具体的队列或绑定
    @RabbitListener(queues = "queue.B")
    public class FanoutReceiverB {
        @RabbitHandler
        public void process(String hello) {
            System.out.println("queue.B : " + hello);
        }
    
    
    }

消费者C:

package com.niugang.mq.receiver;
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    @Component
    //@RabbitListener监听具体的队列或绑定
    @RabbitListener(queues = "queue.C")
    public class FanoutReceiverC {
        @RabbitHandler
        public void process(String hello) {
            System.out.println("queue.C : " + hello);
        }
    
    
    }

6..测试

package com.niugang.test;
    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 com.niugang.mq.send.FanoutSender;
    @RunWith(SpringRunner.class)
    @SpringBootTest
    public class RabbitTest {
    @Autowired
    private FanoutSender fanoutSender;
        @Test
    public void send() throws Exception {
    fanoutSender.send("hello fanout"); 
    }
    }

7.运行结果

结果1

![format_png 1][]

结果2

![format_png 2][]

从运行结果来看，fanout交换器广播消息的顺序是随机的。

**微信公众号**               ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25pdWdhbmcwOTIw_size_16_color_FFFFFF_t_70][]

[format_png]: /images/20220529/0ba982bb5b6b4c3fb746ddaccdbcaef8.png
[format_png 1]: /images/20220529/5a91cd2324a746e396583e5fc4b0cc39.png
[format_png 2]: /images/20220529/0bdbf5b032ff40d3b4f508da645670e4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25pdWdhbmcwOTIw_size_16_color_FFFFFF_t_70]: /images/20220529/d42ebd64a4734b3d9bf4654a8dedecbd.png