具有多个消费者的 Spring Boot RabbitMQ 单队列

问题描述

我是 Rabbitmq 的新手，我想对多个消费者使用单个队列，请帮我处理多个消费者，我们如何处理来自单个队列的多个消费者的请求？

下面是我的单队列和单消费者代码

我的配置类

@Configuration
public class ConfigureRabbitMq {

    public static final String EXCHANGE_NAME = "mikeexchange2";
    public static final String QUEUE_NAME = "mikequeue2";


    @Bean
    Queue createQueue() {
        return new Queue(QUEUE_NAME,true,false,false);
    }

    @Bean
    TopicExchange exchange(){
        return new TopicExchange(EXCHANGE_NAME);
    }

    @Bean
    Binding binding(Queue q,TopicExchange exchange){
        return BindingBuilder.bind(q).to(exchange).with("mike.#");
    }

    @Bean
    SimpleMessageListenerContainer container(ConnectionFactory connectionFactory,MessageListenerAdapter messageListenerAdapter){
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.setQueueNames(QUEUE_NAME);
        container.setMessageListener(messageListenerAdapter);
        return container;
    }


    @Bean
    MessageListenerAdapter listenerAdapter(Receive handler){
        return new MessageListenerAdapter(handler,"handleMessage");
    }
}

我的接收器类

@Service
public class Receive {

    public void handleMessage(String messageBody){
        System.out.println("HandleMessage!!!");
        System.out.println(messageBody);
        
    }

}

我的发件人类

@RestController
public class Send {

    private final  RabbitTemplate rabbitTemplate;
    
    public  Send(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }
    
    
    
    @RequestMapping(method = RequestMethod.GET,value = "/api/send/{msg}")
    public String sendMessage(@PathVariable("msg") String themessage){
        
        for(int i=0;i<5000000;i++) {
            rabbitTemplate.convertAndSend(ConfigureRabbitMq.EXCHANGE_NAME,"mike.springmessages",themessage+""+Integer.toString(i));
        }
        return "We have sent a message! :" + themessage;
    }
}

解决方法

根据您的评论，您实际上还不错..但是就像 Nik 所说的那样，使用默认配置，您正在实现负载平衡循环。

如果您使用 spring-boot，请确保为 spring-boot-starter-amqp 添加依赖项。

在您的接收方，您通常会遇到以下情况：

@Service
public class Receive {

 @RabbitListener(queues = "<queueName>",concurrency = <numConsumersPerInstance>)
 public void handleMessage(String messageBody){
   ...
 }

提供并发参数可以在一个服务实例中使用多线程消息。如果您确实希望一个实例同时仅使用一条消息，则可以将此设置为 1。检查 @RabbitListener 的源代码以获取有关并发的进一步指导。