SpringBoot--整合RabbitMQ

ゝ一纸荒年。 2021-08-24 18:10 587阅读 0赞

### 整合SpringBoot ###

SpringBoot未AMQP提供了自动化配置依赖spring-boot-starter-amqp，因此首先创建项目，并添加依赖：

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

在application.properties中配置RabbitMQ的基本连接信息：

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

在RabbitMQ中，所有的消息生产者提交的消息都会交由Exchange进行再分配，Exchange会根据不同的策略将消息分发到不同的Queue中。RabbitMQ中一共提供了4中不同的Exchange：

*  Direct
 *  Fanout
 *  Topic
 *  Header

Direct：

DirectExchange的路由策略是讲消息队列绑定到一个DirectExchange上，当一条消息到达DirectExchange时会被转发到与该条消息routing key 相同的Queue上，例如消息队列的列名为"hello-queue"，则routingkey为"hello-queue"的消息会被该消息队列接收。

DirectExchange配置如下：

@Configuration
    public class RabbitDirectConfig { 
    
        public final static String DIRECTNAME = "my-direct";
    
        //提供一个消息队列Queue
        @Bean
        Queue queue(){ 
            return new Queue("hello-queue");
        }
    
        //创建一个DirectExchange对象，三个参数分别为：名字、重启后是否依然有效以及长期未使用是否删除
        @Bean
        DirectExchange directExchange(){ 
            return new DirectExchange(DIRECTNAME, true, false);
        }
    
        //创建一个Binding对象，将Exchange和Queue绑定在一起
        @Bean
        Binding binding(){ 
            return BindingBuilder.bind(queue()).to(directExchange()).with("direct");
        }
    }

配置一个消费者：

@Component
    public class DirectReceiver { 
    
        @RabbitListener(queues = "hello-queue")
        public void handler1(String msg){ 
            System.out.println("DirectReceiver:" + msg);
        }
    }

单元测试：

@RunWith(SpringRunner.class)
    @SpringBootTest
    class DemoApplicationTests { 
    
        @Autowired
        RabbitTemplate rabbitTemplate;
    
        @Test
        public void directTest(){ 
            rabbitTemplate.convertAndSend("hello-queue", "hellow direct!");
        }
    
    }

确定RabbitMQ已经启动，然后启动项目，控制台打印：

DirectReceiver:hellow direct!

如果发送的消息为对象类型，可以将其转换为JSON格式，需要配置一个MessageConverter：

@Configuration
    public class MyAMQPConfig { 
    
        @Bean
        public MessageConverter messageConverter(){ 
            return new Jackson2JsonMessageConverter();
        }
    }

测试：

@Test
        public void directTest(){ 
            //rabbitTemplate.convertAndSend("hello-queue", "hellow direct!");
            rabbitTemplate.convertAndSend("hello-queue", new Book(1, "红楼梦", "曹雪芹"));
        }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70]

Fanout：

FanoutExchange的数据交换策略是把所有到达FanoutExchange的消息转发给所有与它绑定的Queue，在这种策略中，routingkey将不起任何作用，FanoutExchange的配置方式如下：

@Configuration
    public class RabbitFanoutConfig { 
        
        public final static String FANOUTNAME = "my-fanout";
        
        @Bean
        FanoutExchange fanoutExchange(){ 
            return new FanoutExchange(FANOUTNAME, true, false);
        }
        
        @Bean
        Queue queue1(){ 
            return new Queue("queue-one");
        }
        
        @Bean
        Queue queue2(){ 
            return new Queue("queue-two");
        }
        
        @Bean
        Binding binding1(){ 
            return BindingBuilder.bind(queue1()).to(fanoutExchange());
        }
        
        @Bean
        Binding binding2(){ 
            return BindingBuilder.bind(queue2()).to(fanoutExchange());
        }
    }

在这里收下创建FanoutExchange，参数的含义与DirectExchange一致，然后创建两个Queue，再将这两个Queue都绑定到FanoutExchange上。接下来创建两个消费者，代码如下：

@Component
    public class FanoutReceiver { 
    
        @RabbitListener(queues = "queue-one")
        public void handler1(String msg){ 
            System.out.println("FanoutReceiver:handler1:" + msg);
        }
    
        @RabbitListener(queues = "queue-two")
        public void handler2(String msg){ 
            System.out.println("FanoutReceiver:handler2:" + msg);
        }
    }

单元测试：

@Test
        public void fanoutTest(){ 
            //这里发送消息不需要routingkey，指定exchange即可
            rabbitTemplate.convertAndSend(RabbitFanoutConfig.FANOUTNAME, null, "hello fanout!");
        }

运行测试，控制台打印：

FanoutReceiver:handler2:hello fanout!
    FanoutReceiver:handler1:hello fanout!

Topic：

TopicExchange是比较复杂也比较灵活的一种路由策略，在TopicExchange中，Queue通过routingkey绑定到TopicExchange上，当消息到达TopicExchange后，TopicExchange根据消息的routingkey将消息路由到一个或者多个Queue上。TopicExchange配置如下：

@Configuration
    public class RabbitTopicConfig { 
        
        public final static String TOPICNAME = "my-topic";
        
        @Bean
        TopicExchange topicExchange(){ 
            return new TopicExchange(TOPICNAME, true, false);
        }
        
        @Bean
        Queue xiaomi(){ 
            return new Queue("xiaomi");
        }
        
        @Bean
        Queue huawei(){ 
            return new Queue("huawei");
        }
        
        @Bean
        Queue phone(){ 
            return new Queue("phone");
        }
        
        @Bean
        Binding xiaomiBinding(){ 
            return BindingBuilder.bind(xiaomi()).to(topicExchange()).with("xiaomi.#");
        }
        
        @Bean
        Binding huaweiBinding(){ 
            return BindingBuilder.bind(huawei()).to(topicExchange()).with("huawei.#");
        }
        
        @Bean
        Binding phoneBinding(){ 
            return BindingBuilder.bind(phone()).to(topicExchange()).with("#.phone.#");
        }
    }

将三个Queue分别绑定到TopicExchange上，第一个Binding中的"xiaomi.\#"表示消息的routingkey凡是以"xiaomi"开头的，都将被路由到名称为"xiaomi"的Queue上；第二个Binding中的"huawei.\#“表示消息的routingkey凡是以"huawei"开头的，都将被路由到名称为"huawei"的Queue上；第三个Binding中的”\#.phone.\#"则表示消息的routingkey中凡是包含"phone"的，都将被路由到名称为"phone"的Queue上。

针对单个Queue创建三个消费者，代码如下：

@Component
    public class TopicReceiver { 
        
        @RabbitListener(queues = "phone")
        public void handler1(String msg){ 
            System.out.println("PhoneReceeiver:" + msg);
        }
        
        @RabbitListener(queues = "xiaomi")
        public void handler2(String msg){ 
            System.out.println("XiaoMiReceiver:" + msg);
        }
        
        @RabbitListener(queues = "huawei")
        public void handler3(String msg){ 
            System.out.println("HuaWeiReceiver:" + msg);
        }
    }

单元测试：

@Test
        public void topicTest(){ 
            rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPICNAME, "xiaomi.news", "小米新闻..");
            rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPICNAME, "huawei.news", "华为新闻..");
            rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPICNAME, "xiaomi.phone", "小米手机..");
            rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPICNAME, "huawei.phone", "华为手机..");
            rabbitTemplate.convertAndSend(RabbitTopicConfig.TOPICNAME, "phone.news", "手机新闻..");
        }

运行单元测试，控制台打印：

XiaoMiReceiver:小米新闻..
    PhoneReceeiver:小米手机..
    HuaWeiReceiver:华为新闻..
    PhoneReceeiver:华为手机..
    XiaoMiReceiver:小米手机..
    HuaWeiReceiver:华为手机..
    PhoneReceeiver:手机新闻..

Header：  
HeaderExchange是一种使用较少的路由策略，HeaderExchange会根据消息的Header将消息路由到不同的Queue傻姑娘，这种策略也和routingkey无关，配置如下：

@Configuration
    public class RabbitHeaderConfig { 
        
        public final static String HEADERNAME = "my-header";
        
        @Bean
        HeadersExchange headersExchange(){ 
            return new HeadersExchange(HEADERNAME, true, false);
        }
        
        @Bean
        Queue queueName(){ 
            return new Queue("name-queue");
        }
        
        @Bean
        Queue queueAge(){ 
            return new Queue("age=queue");
        }
        
        @Bean
        Binding bindingName(){ 
            Map<String, Object> map = new HashMap<>();
            map.put("name", "my");
            //whereAny表示消息的Header中只要有一个Header匹配上map中的key/value，就把该消息路由到名为"name-queue"的Queue上
            //也可以使用whereAll方法，表示消息的所有Header都要匹配
            //whereAny和whereAll实际上对应了一个名为x-match的属性
            return BindingBuilder.bind(queueName()).to(headersExchange()).whereAny(map).match();
        }
        
        @Bean
        Binding bindingAge(){ 
            //bindingAge中的配置则表示只要消息的Header中包含age，无论age的值是多少，都将消息路由到名为"age-queue"的Queue上
            return BindingBuilder.bind(queueAge()).to(headersExchange()).where("age").exists();
        }
    }

创建两个消费者(参数用byte\[\]接收)：

@Component
    public class HeaderReceiver { 
        
        @RabbitListener(queues = "name-queue")
        public void handler1(byte[] msg){ 
            System.out.println("HeaderReceiver:name:" + new String(msg, 0, msg.length));
        }
        
        @RabbitListener(queues = "age-queue")
        public void handler2(byte[] msg){ 
            System.out.println("HeaderReceiver:age:" + new String(msg, 0, msg.length));
        }
    }

单元测试：

@Test
        public void headerTest(){ 
            Message nameMsg = MessageBuilder.withBody("hello header! name-queue".getBytes())
                    .setHeader("name", "sang").build();
            Message ageMsg = MessageBuilder.withBody("hello header! age-queue".getBytes())
                    .setHeader("age", "99").build();
            rabbitTemplate.send(RabbitHeaderConfig.HEADERNAME, null, ageMsg);
            rabbitTemplate.send(RabbitHeaderConfig.HEADERNAME, null, nameMsg);
        }

运行测试，控制台打印：

HeaderReceiver:age:hello header! age-queue
    HeaderReceiver:name:hello header! name-queue

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70]: /images/20210813/449c7d7ecc4248128fdf2206cc43410d.png