Spring AMQP RPC消费者并抛出异常

Question

我有一个 RPC 模式的消费者（RabbitListner），我想知道是否可以抛出可以被发布者处理的异常。

为了更清楚我的解释，案例如下：

• 发布者以 RPC 方式发送消息
• 消费者收到消息，检查消息的有效性，如果消息由于缺少参数而无法计数，那么我想抛出异常。异常可以是特定的业务异常或特定的 AmqpException，但我希望发布者可以处理此异常，如果它没有超时。

我尝试使用 AmqpRejectAndDontRequeueException，但我的发布者没有收到异常，而只是一个空的响应。

是否可以这样做，或者这样实施可能不是一个好习惯？

编辑 1：

在@GaryRussel 回复之后是我的问题的解决方案：

1. 对于 RabbitListner，我创建了一个错误处理程序：

    @Configuration
    public class RabbitErrorHandler implements         RabbitListenerErrorHandler {
    @Override public Object handleError(Message message, org.springframework.messaging.Message<?> message1, ListenerExecutionFailedException e) {
   throw e;
   }
   

   }

2. 将 bean 定义到配置文件中：

   @配置 公共类 RabbitConfig 扩展 RabbitConfiguration {

   @Bean
   public RabbitTemplate getRabbitTemplate() {
       Message.addWhiteListPatterns(RabbitConstants.CLASSES_TO_SEND_OVER_RABBITMQ);
   return new RabbitTemplate(this.connectionFactory());
   

   }

   /**
   * Define the RabbitErrorHandle
   * @return Initialize RabbitErrorHandle bean
   */
   @Bean
   public RabbitErrorHandler rabbitErrorHandler() {
      return new RabbitErrorHandler();
   }
   }
   

3. 使用参数创建 @RabbitListner，其中 rabbitErrorHandler 是我之前定义的 bean：

   @Override
   @RabbitListener(queues = "${rabbit.queue}"
       , errorHandler = "rabbitErrorHandler"
       , returnExceptions = "true")
    public ReturnObject receiveMessage(Message message) {
   

4. 对于 RabbitTemplate 我设置了这个属性：

      rabbitTemplate.setMessageConverter(new RemoteInvocationAwareMessageConverterAdapter());
   

当消息受到消费者威胁，但它发送了错误时，我获得了一个 RemoteInvocationResult，其中包含 e.getCause().getCause() 中的原始异常。

Answer 1

查看@RabbitListener 上的returnExceptions 属性（自2.0 起）。 Docs here.

returnExceptions属性，当true会导致异常返回给发送者。异常包装在 RemoteInvocationResult 对象中。

在发送方，有一个可用的RemoteInvocationAwareMessageConverterAdapter，如果配置到RabbitTemplate 中，将重新抛出服务器端异常，包裹在AmqpRemoteException 中。服务器异常的堆栈跟踪将通过合并服务器和客户端堆栈跟踪来合成。

重要

这种机制一般只适用于使用Java序列化的默认SimpleMessageConverter；异常通常不是“杰克逊友好的”，因此不能序列化为 JSON。如果您使用的是 JSON，请考虑在抛出异常时使用 errorHandler 来返回其他一些对 Jackson 友好的 Error 对象。

Answer 2

您必须将消息作为错误返回，消费应用程序可以选择将其视为异常。但是，我认为正常的异常处理流程不适用于消息传递。您的发布应用程序（RPC 服务的使用者）需要知道可能出现的问题并进行编程以处理这些可能性。

Answer 3

对我有用的是：

在“服务”方面：

• 服务

  @RabbitListener(id = "test1", containerFactory ="BEAN CONTAINER FACTORY", 
  queues = "TEST QUEUE", returnExceptions = "true")
  DataList getData() {
    // this exception will be transformed by rabbit error handler to a RemoteInvocationResult
   throw new IllegalStateException("mon expecion");
   //return dataHelper.loadAllData();
  }
  

在“请求”方面：

• 服务

  public void fetchData() throws AmqpRemoteException {
    var response = (DataList) amqpTemplate.convertSendAndReceive("TEST EXCHANGE", "ROUTING NAME", new Object());
  
    Optional.ofNullable(response)
           .ifPresentOrElse(this::setDataContent, this::handleNoData);
  }
  

• 配置

  @Bean
  AmqpTemplate amqpTemplate(ConnectionFactory connectionFactory, MessageConverter messageConverter) {
     var rabbitTemplate = new RabbitTemplate(connectionFactory);
     rabbitTemplate.setMessageConverter(messageConverter);
     return rabbitTemplate;
  }
  @Bean
  MessageConverter jsonMessageConverter() {
     ObjectMapper objectMapper = new ObjectMapper();
     objectMapper.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS, false);
     objectMapper.disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES);
     objectMapper.registerModule(new JavaTimeModule());
  
     var jsonConverter = new Jackson2JsonMessageConverter(objectMapper);
  
     DefaultClassMapper classMapper = new DefaultClassMapper();
     Map<String, Class<?>> idClassMapping = Map.of(
        DataList.class.getName(), DataList.class,
        RemoteInvocationResult.class.getName(), RemoteInvocationResult.class
     );
     classMapper.setIdClassMapping(idClassMapping);
  
     jsonConverter.setClassMapper(classMapper);
  
     // json converter with returned exception awareness
     // this will transform RemoteInvocationResult into a AmqpRemoteException 
     return new RemoteInvocationAwareMessageConverterAdapter(jsonConverter);
  }