CODEWITHSUNDEEP
javaspringrabbitmqqueue
Андрей Сурков
Андрей Сурков

Reputation: 83

RabbitMQ queue and routing key

In documentation https://docs.spring.io/spring-amqp/reference/htmlsingle/ i see 

@RabbitListener(bindings = @QueueBinding(
        value = @Queue(value = "myQueue", durable = "true"),
        exchange = @Exchange(value = "auto.exch", ignoreDeclarationExceptions = "true"),
        key = "orderRoutingKey")
  )
  public void processOrder(Order order) {

  }

  @RabbitListener(bindings = @QueueBinding(
        value = @Queue,
        exchange = @Exchange(value = "auto.exch"),
        key = "invoiceRoutingKey")
  )
  public void processInvoice(Invoice invoice) {

  }

Here 1 queue and 2 another routing keys, everyone for his method But my code doesn't get message from key!

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = DRIVER_QUEUE, durable = "true"),
            exchange = @Exchange(value = "exchange", ignoreDeclarationExceptions = "true", autoDelete = "true"),
            key = "order")
    )
    public String getOrders(byte[] message) throws InterruptedException {
         System.out.println("Rout order");
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = DRIVER_QUEUE, durable = "true"),
            exchange = @Exchange(value = "exchange", ignoreDeclarationExceptions = "true", autoDelete = "true"),
            key = "invoice")
    )
    public String getOrders(byte[] message) throws InterruptedException {
         System.out.println("Rout invoice");
    }

they all get message from queue and not see key... site send in queue message with key "invoice" and i see in console "Route order" Whats problem?? Thank a lot!

rabbitmq 3.7.3 spring 4.2.9 org.springframework.amqp 1.7.5

Upvotes: 4

Views: 10013

Answers (1)

Sampisa
Sampisa

Reputation: 1583

The error is that you send all messages to the same queue.

You should use a different queue for each listener. Your bindings just tell that messages with RK="invoice" and RK="order" must go on the same queue, not that listener processes queue elements with that RK.

You should bind e.g. exchange to DRIVER_QUEUE1 (e.g. "queue-orders") via key "invoice" and exchange to DRIVER_QUEUE2 (e.g. "queue-invoices") via key "order". In this way you separate messages, and you can put in place two listeners, one for invoices and one for orders. E.g. something like this:

@RabbitListener(queues = "queue-orders")
public void handleOrders(@Payload Order in,
      @Header(AmqpHeaders.RECEIVED_ROUTING_KEY) String key) {
   logger.info("Key: {}, msg: {}",key,in.toString());
}


@RabbitListener(queues = "queue-invoices")
public void handleInvoices(@Payload Invoice in, 
      @Header(AmqpHeaders.RECEIVED_ROUTING_KEY) String key) {
   logger.info("Key: {}, msg: {}",key,in.toString());
}

I don't like whole complete annotation, since when broker configuration is made IMHO full annotation becomes useless (or better, adds extra-check useless for me). But if you prefer, whole annotation should be like

@RabbitListener(bindings = @QueueBinding(
        value = @Queue(value = "queue-orders", durable = "true"),
        exchange = @Exchange(value = "exchange", ignoreDeclarationExceptions = "true", autoDelete = "true"),
        key = "invoice")
)

then you can send messages via convertAndSend(exchangename, routingkey, object) like in

Order order = new Order(...);
rabbitTemplate.convertAndSend("exchange", "order", order);
Invoice invoice = new Invoice(...);
rabbitTemplate.convertAndSent("exchange", "invoice", invoice);

If your boot application implements RabbitListenerConfigurer, then you can configure everything as e.g.

@SpringBootApplication
public class MyApplication implements RabbitListenerConfigurer {
   // other config stuff here....

    @Bean("queue1")
    public Queue queue1() {
        return new Queue("queue-orders", true);
    }

    @Bean("queue2")
    public Queue queue2() {
        return new Queue("queue-invoices", true);
    }

    @Bean
    public Binding binding1(@Qualifier("queue1") Queue queue, TopicExchange exchange) {        
        return BindingBuilder.bind(queue).to(exchange).with("invoice");
    }

    @Bean
    public Binding binding2(@Qualifier("queue2") Queue queue, TopicExchange exchange) {        
        return BindingBuilder.bind(queue).to(exchange).with("order");
    }

    @Bean
    public RabbitTemplate rabbitTemplate(final ConnectionFactory connectionFactory) {
        final RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
    rabbitTemplate.setMessageConverter(producerJackson2MessageConverter());
        return rabbitTemplate;
    }

    @Bean
    public Jackson2JsonMessageConverter producerJackson2MessageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    @Bean
    public DefaultMessageHandlerMethodFactory messageHandlerMethodFactory() {
        DefaultMessageHandlerMethodFactory factory = new DefaultMessageHandlerMethodFactory();
        factory.setMessageConverter(consumerJackson2MessageConverter());
        return factory;
    }

    @Override
    public void configureRabbitListeners(final RabbitListenerEndpointRegistrar registrar) {
        registrar.setMessageHandlerMethodFactory(messageHandlerMethodFactory());
    }

    // Exchange.
    @Bean
    public TopicExchange exchange() {
        return new TopicExchange("exchange");
    }
}

Hoping to have anwered to your request.

Upvotes: 9

Related Questions