rabbitmq多个消费者消费同一个队列中的同一条消息

小灰灰 2024-03-01 04:25 22阅读 0赞

使用springboot整合rabbitmq实现，一个生产者生产一条数据，多个消费者消费同一条数据案例  
1.引入库

<dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-amqp</artifactId>
            </dependency>
            <dependency>
                <groupId>com.alibaba</groupId>
                <artifactId>fastjson</artifactId>
                <version>1.2.62</version>
            </dependency>

2.rabbitmq相关配置

@Configuration
    public class RabbitmqConfig {
        
    	@Bean
        public Queue queue1() {
        
            return new Queue("queue1",true);
        }
    
        @Bean
        public FanoutExchange exchange1() {
        
            return new FanoutExchange("exchange1",true, false);
        }
    
        @Bean
        public Binding binding1() {
        
            return BindingBuilder.bind(queue1()).to(exchange1());
        }
    }

3.生产者代码  
代码如下（示例）：

@Resource
    private RabbitTemplate rabbitTemplate;
    
    @Test
    public void pushMessage() {
        
        HashMap<Object, Object> hashMap = new HashMap<>();
        hashMap.put("name","zhangsan");
        hashMap.put("age","18");
        rabbitTemplate.convertAndSend("exchange1","", hashMap);
    }

4.多个消费者，消费同一条数据。  
消费者1号：  
代码如下（示例）：

@Component
    public class RabbitMqConsumer {
        
    //    @RabbitListener(queues = "queue1")
        @RabbitListener(bindings = @QueueBinding(
                // value = @Queue(value = "queue1"),
    		value = @Queue(), //切记： 此处无需设置队列名称，否在得话，多个消费者只有一个消费者能消费数据。其它消费者无法消费数据。
            exchange = @Exchange(value = "exchange1",type = ExchangeTypes.FANOUT)
        ))
        public void getData(Message message) {
        
            try {
        
                String str = new String(message.getBody(),"utf-8");
                System.out.println(str);
            } catch (Exception e) {
        
                e.printStackTrace();
            }
        }
    }

消费者2号：  
代码如下（示例）：

@Component
    public class RabbitMqConsumer {
        
    	// @RabbitListener(queues = "queue1")
        @RabbitListener(bindings = @QueueBinding(
    	// value = @Queue(value = "queue1"),
    		value = @Queue(), //切记： 此处无需设置队列名称，否在得话，多个消费者只有一个消费者能消费数据。其它消费者无法消费数据。
            exchange = @Exchange(value = "exchange1",type = ExchangeTypes.FANOUT)
        ))
        public void getData(Message message) {
        
            try {
        
                String str = new String(message.getBody(),"utf-8");
                System.out.println(str);
            } catch (Exception e) {
        
                e.printStackTrace();
            }
        }
    }