使用Spring將AMQP消息解碼爲地圖

Question

我必須使用Spring解碼AMQP消息。爲了處理它，我現在使用：

// Configure queue. 
    RabbitAdmin admin = new RabbitAdmin(cf); 
    Queue queue = new Queue(queueName); 
    admin.declareQueue(queue); 
    FanoutExchange exchange = new FanoutExchange(exchangeName); 
    admin.declareExchange(exchange); 
    admin.declareBinding(BindingBuilder.bind(queue).to(exchange)); 

    // set up the listener and container 
    SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(cf); 

    MessageListenerAdapter adapter = new MessageListenerAdapter(listener); 
    container.setMessageListener(adapter); 
    container.setQueueNames(queueName); 
    container.start(); 

我的聽衆是

public class DataListener { 

    public void handleMessage(Object incomingMessage) { 
     LOGGER.error("AMQP: got message.{}", incomingMessage); 
    } 

} 

使用AmqpTemplate的convertAndSend方法發送的消息。沒有配置給AmqpTemplate，一切都是默認的。

我怎麼可能收到我的incomingMessage作爲字段的HashMap？我不想強烈地將它耦合到特定的對象類型。

Answer 1

假設你的意思是你的消息是一個POJO豆...

使用JSON - 在出站端使用Jackson2JsonMessageConverter，而不是默認SimpleMessageConverter，它使用Java序列化。

在接收端，相同的JSON轉換器將嘗試將傳入流轉換爲原始POJO。

爲了避免這種情況，請配置JSON消息轉換器以將類名映射到HashMap而不是原始POJO。

您可以通過爲轉換器提供自定義DefaultJackson2JavaTypeMapper來完成此操作，該配置爲將類名稱從__TypeId__標頭映射到java.util.HashMap。

編輯

或者你可以簡單地注入ClassMapper總是返回HashMap - 這裏有一個快速啓動應用程序，我寫來說明該技術：

@SpringBootApplication 
public class So36837736Application { 

    public static void main(String[] args) throws Exception { 
     ConfigurableApplicationContext context = SpringApplication.run(So36837736Application.class, args); 
     context.getBean(RabbitTemplate.class).convertAndSend(new Foo("bar")); 
     Thread.sleep(10000); 
     context.close(); 
    } 

    @Bean 
    public RabbitTemplate template(ConnectionFactory connectionFactory) { 
     RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory); 
     rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter()); 
     rabbitTemplate.setRoutingKey(queue().getName()); 
     return rabbitTemplate; 
    } 

    @Bean 
    public SimpleMessageListenerContainer container(ConnectionFactory connectionFactory) { 
     SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory); 
     container.setQueues(queue()); 
     MessageListenerAdapter adapter = new MessageListenerAdapter(new Object() { 

      @SuppressWarnings("unused") 
      public void handleMessage(Map<String, Object> map) { 
       System.out.println("\n\n\n" + map + "\n\n\n"); 
      } 

     }); 
     Jackson2JsonMessageConverter messageConverter = new Jackson2JsonMessageConverter(); 
     ClassMapper classMapper = new ClassMapper() { 

      @Override 
      public void fromClass(Class<?> clazz, MessageProperties properties) { 
      } 

      @Override 
      public Class<?> toClass(MessageProperties properties) { 
       return HashMap.class; 
      } 

     }; 
     messageConverter.setClassMapper(classMapper); 
     adapter.setMessageConverter(messageConverter); 
     container.setMessageListener(adapter); 
     return container; 
    } 

    @Bean 
    public Queue queue() { 
     return new AnonymousQueue(); 
    } 

    public static class Foo { 

     private final String bar; 

     private Foo(String bar) { 
      this.bar = bar; 
     } 

     public String getBar() { 
      return this.bar; 
     } 

    } 

}