根据标头属性（例如路由键）处理入站 AMQP 消息

Question

我有一项服务，可以接收AMQP消息。该服务与队列绑定，该队列接收所有与一组路由密钥匹配的消息。

我的设置如下：

...

private SomeController controller;

@Autowired
private SimpleMessageListenerContainer receiverContainer;

@Bean
public IntegrationFlow inboundFlow(){
     var adpater = Amqp.inboundAdapter(receiverContainer);
     return IntegrationFlows.from(adapter)
               // some transformations
               .handle(controller, "processMessage")
               .get();
}

这已经很好。但是，现在我想根据标头属性处理不同控制器的消息。在这种情况下，我想为每个路由键提供一个控制器。使用一个带有多个路由键的单个队列只是为每个密钥以不同的方式处理它也是一个好主意吗？

Answer 1

在交换和单个队列之间拥有多个绑定确实是合法的。

在本教程中查看更多信息： https://wwwww.rabbitmq.com /tutorials/tutorial-four-spring-amqp.html 。

amqp.inboundadapter（）依赖于 defaultAmqpheaderMapper.inboundmapper（） 默认情况下，该 by默认情况下为我们填充了 amqpheaders.received_routing_routing_key meesse header。因此，您确实可以使用 route（message.class，m-＆gt; m.getheaders（）。路由键值。

Answer 2

我只是想添加一个代码示例，并入 artem bilan 's（正确）答案，因为此外，我必须合并一个网关（由Artem Bilan暗示了“适当的 channelmapping（））。

更多有关您为什么需要网关或某些情况下的桥梁，请参阅

我的初始代码snippit变成了以下内容：

...

@Autowired
private FirstController firstController;

@Autowired
private SecondController secondController;

@Autowired
private SimpleMessageListenerContainer receiverContainer;

@Bean
public IntegrationFlow inboundFlow(){
     var adpater = Amqp.inboundAdapter(receiverContainer);
     return IntegrationFlows.from(adapter)
               // some transformations
               .route(Message.class, getMessageRoutingKey(m),
                        m -> m.subFlowMapping("routingKey1", firstFlow())
                                // after the first subFlow, all further integrationflows are wrapped in a gateway
                                .subFlowMapping("routingKey2", sf -> sf.gateway(secondFlow())))
               .get();
}


@Bean
public IntegrationFlow firstFlow() {
      return f -> f
               // e.g. additional transformations
               .handle(firstController, "processMessageInFirstFashion");
}


@Bean
public IntegrationFlow secondFlow() {
      return f -> f
               // e.g. additional transformations
               .handle(secondController, "processMessageInSecondFashion");
}


private static String getMessageRoutingKey(final Message<?> message) {
       return message.getHeaders().get(AmqpHeaders.RECEIVED_ROUTING_KEY).toString();
}