springBoot集成rabbitmq 之路由模式模式(Routing)

阳光穿透心脏的1/2处 2024-03-27 16:27 51阅读 0赞

### springBoot集成rabbitmq 之路由模式模式(Routing) ###

**springBoot整合rabbitmq的例子: https://blog.csdn.net/weixin\_45730866/article/details/128971917，建议先看。**  
  
  
consumer服务

配置类

package com.lideru.consumer.routing; 
    import org.springframework.amqp.core.Binding;
    import org.springframework.amqp.core.BindingBuilder;
    import org.springframework.amqp.core.DirectExchange;
    import org.springframework.amqp.core.Queue;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration; 
    @Configuration
    public class RabbitRoutingConfig {
         
        /**
         * Queue 可以有4个参数
         *      1.name          队列名
         *      2.durable       持久化消息队列 ,rabbitmq重启的时候不需要创建新的队列 默认true
         *      3.auto-delete   表示消息队列没有在使用时将被自动删除 默认是false
         *      4.exclusive     表示该消息队列是否只在当前connection生效,默认是false
         */
        @Bean
        public Queue createRoutingQueueA() {
        
            return new Queue("Queue_Routing_A",true);
        }
        @Bean
        public Queue createRoutingQueueB() {
        
            return new Queue("Queue_Routing_B",true);
        } 
        @Bean
        public DirectExchange routingExchange() {
        
            //配置广播路由器
            return new DirectExchange("Exchange_Routing");
        } 
        @Bean
        public Binding bingQueueAToRoutingExchange() {
        
            return  BindingBuilder.bind(createRoutingQueueA()).to(routingExchange()).with("info");
        } 
        @Bean
        public Binding bingQueueBToRoutingExchange() {
        
            return  BindingBuilder.bind(createRoutingQueueB()).to(routingExchange()).with("error");
        } 
    }

接收者A B

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component; 
    import java.util.Date; 
    @Component
    @RabbitListener(queues = "Queue_Routing_B")
    public class RoutingReceiverA {
        
        @RabbitHandler
        public void process(String message) {
        
            System.out.println("ReceiverA : " + message +"  "+ new Date());
        }
    }

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component; 
    import java.util.Date; 
    @Component
    @RabbitListener(queues ="Queue_Routing_A")
    public class RoutingReceiverB {
        
        @RabbitHandler
        public void process(String message) {
        
            System.out.println("ReceiverB : " + message +"   "+ new Date());
        }
    }

pruducer服务

@Component
    public class RoutingSender {
        
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        public void send(int i) {
        
            String contentA = "路由 A=" + i + "," + new Date();
            String contentB = "路由 B=" + (i + 1) + "," + new Date();
            //消息发送，使用void convertAndSend(String exchange, String routingKey, Object message) throws AmqpException;
            //但不指定routingKey。因为FanoutExchange类型的交换机，routingKey不起作用，它向所有的队列发送广播，只要队列绑定到该交换机即接受消息。
            this.rabbitTemplate.convertAndSend("Exchange_Routing", "info", contentA);
            System.out.println("发送成功," + new Date() + "," + contentA);
            this.rabbitTemplate.convertAndSend("Exchange_Routing", "error", contentB);
            System.out.println("发送成功," + new Date() + "," + contentB);
        }
    }

测试

import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    @RestController
    public class RoutingTestsController  {
        
        @Autowired
        private RoutingSender routingSender; 
        @GetMapping("/routingTest")
        public void routingTest() {
        
            int i = 5201314;
            routingSender.send(i);
        }
    }

调用http://127.0.0.1:8080/routingTest，producer 和 consumer服务控制台打印

发送成功,Mon Feb 13 11:18:02 CST 2023,路由 A=5201314,Mon Feb 13 11:18:02 CST 2023
    发送成功,Mon Feb 13 11:18:02 CST 2023,路由 B=5201315,Mon Feb 13 11:18:02 CST 2023

ReceiverB : 路由 A=5201314,Mon Feb 13 11:18:02 CST 2023   Mon Feb 13 11:18:02 CST 2023
    ReceiverA : 路由 B=5201315,Mon Feb 13 11:18:02 CST 2023  Mon Feb 13 11:18:02 CST 2023

由此可见producer 共计生产了 2 个消息，分别被消费者 A 和 B 各消费了 1 条。