Spring AMQP - 发送者和接收消息

2 回答

• 3

  它导致例外

  有什么例外？

  TypeId：com.diff.approach.JobListenerDTO

  这意味着您正在发送DTO，而不是您在问题中描述的哈希映射 .

  如果要删除typeId标头，可以使用消息后处理器...

  rabbitTemplate.convertAndSend(Constants.INPUT_QUEUE, dto, m -> {
      m.getMessageProperties.getHeaders().remove("__TypeId__");
      return m;
  });
  

  （或 , new MessagePostProcessor() {...} 如果您不使用Java 8） .

  EDIT

  您使用的是什么版本的Spring AMQP？使用1.6，您甚至不必删除 __TypeId__ 标头 - 框架查看侦听器参数类型并告诉Jackson转换器类型，以便它自动转换为（如果可以） . 正如你在这里看到的那样;它没有删除类型ID工作正常...

  package com.example;
  
  import java.util.HashMap;
  import java.util.concurrent.CountDownLatch;
  import java.util.concurrent.TimeUnit;
  
  import org.springframework.amqp.core.Queue;
  import org.springframework.amqp.rabbit.annotation.RabbitListener;
  import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
  import org.springframework.amqp.rabbit.connection.ConnectionFactory;
  import org.springframework.amqp.rabbit.core.RabbitAdmin;
  import org.springframework.amqp.rabbit.core.RabbitTemplate;
  import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
  import org.springframework.boot.SpringApplication;
  import org.springframework.boot.autoconfigure.SpringBootApplication;
  import org.springframework.context.ConfigurableApplicationContext;
  import org.springframework.context.annotation.Bean;
  
  @SpringBootApplication
  public class So39443850Application {
  
      private static final String QUEUE = "so39443850";
  
      public static void main(String[] args) throws Exception {
          ConfigurableApplicationContext context = SpringApplication.run(So39443850Application.class, args);
          context.getBean(RabbitTemplate.class).convertAndSend(QUEUE, new DTO("baz", "qux"));
          context.getBean(So39443850Application.class).latch.await(10, TimeUnit.SECONDS);
          context.getBean(RabbitAdmin.class).deleteQueue(QUEUE);
          context.close();
      }
  
      private final CountDownLatch latch = new CountDownLatch(1);
  
      @RabbitListener(queues = QUEUE, containerFactory = "adapterOPListenerContainerFactory")
      public void listen(HashMap<String, Object> message) {
          System.out.println(message.getClass() + ":" + message);
          latch.countDown();
      }
  
      @Bean
      public Queue queue() {
          return new Queue(QUEUE);
      }
  
      @Bean
      public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
          RabbitTemplate template = new RabbitTemplate(connectionFactory);
          template.setMessageConverter(new Jackson2JsonMessageConverter());
          return template;
      }
  
      @Bean
      public SimpleRabbitListenerContainerFactory adapterOPListenerContainerFactory(ConnectionFactory connectionFactory) {
          SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
          factory.setConnectionFactory(connectionFactory);
          factory.setMessageConverter(new Jackson2JsonMessageConverter());
          return factory;
      }
  
      public static class DTO {
  
          private String foo;
  
          private String baz;
  
          public DTO(String foo, String baz) {
              this.foo = foo;
              this.baz = baz;
          }
  
          public String getFoo() {
              return this.foo;
          }
  
          public void setFoo(String foo) {
              this.foo = foo;
          }
  
          public String getBaz() {
              return this.baz;
          }
  
          public void setBaz(String baz) {
              this.baz = baz;
          }
  
      }
  
  }
  

  结果：

  class java.util.HashMap:{foo=baz, baz=qux}
  

  这在the documentation中有描述......

  在1.6之前的版本中，转换JSON的类型信息必须在消息头中提供，或者需要自定义ClassMapper . 从版本1.6开始，如果没有类型信息头，则可以从目标方法参数推断出类型 .

  您还可以将自定义 ClassMapper 配置为始终返回 HashMap .

  回复于 2024-05-09T10:49:56+08:00
• 0

  • 想要在收到消息时使用“a”不同的Java calss？

  使用自定义ClassMapper配置@Bean Jackson2JsonMessageConverter

  • 收到消息时想要使用“很多”不同的Java calss吗？如 ：

  @MyAmqpMsgListener
  void handlerMsg(
          // Main message class, by MessageConverter
          @Payload MyMsg myMsg, 
  
          // Secondary message class - by MessageConverter->ConversionService
          @Payload Map<String, String> map,
  
          org.springframework.messaging.Message<MyMsg> msg,
          org.springframework.amqp.core.Message amqpMsg
  ) {
      // ...
  }
  

  提供自定义@Bean Converter ， ConversionService ， RabbitListenerAnnotationBeanPostProcessor ：

  @Bean
  FormattingConversionServiceFactoryBean rabbitMqCs(
          Set<Converter> converters
  ) {
      FormattingConversionServiceFactoryBean fac = new FormattingConversionServiceFactoryBean();
      fac.setConverters(converters);
      return fac;
  }
  @Bean
  DefaultMessageHandlerMethodFactory messageHandlerMethodFactory(
          @Qualifier("rabbitMqCs")
          FormattingConversionService rabbitMqCs
  ) {
      DefaultMessageHandlerMethodFactory defaultFactory = new DefaultMessageHandlerMethodFactory();
      defaultFactory.setConversionService(rabbitMqCs);
      return defaultFactory;
  }
  
  // copied from RabbitBootstrapConfiguration
  @Bean(name = RabbitListenerConfigUtils.RABBIT_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME)
  @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
  public RabbitListenerAnnotationBeanPostProcessor rabbitListenerAnnotationProcessor(
          MessageHandlerMethodFactory handlerFac
  ) {
      RabbitListenerAnnotationBeanPostProcessor bpp = new RabbitListenerAnnotationBeanPostProcessor();
      bpp.setMessageHandlerMethodFactory(handlerFac);
      return bpp;
  }
  
  @Bean(name = RabbitListenerConfigUtils.RABBIT_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME)
  public RabbitListenerEndpointRegistry defaultRabbitListenerEndpointRegistry() {
      return new RabbitListenerEndpointRegistry();
  }
  

  参考文献：

  • Jackson2JsonMessageConverter

  • AMQP-461

  • 调试源代码PayloadArgumentResolver

  回复于 2024-05-09T10:49:56+08:00