Spring AMQP RPC消费者和抛出异常

Question

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

为了更清楚地说明我的情况如下：

• 发布者以RPC模式发送消息
• 消费者收到消息，检查消息的有效性，如果由于缺少参数，消息无法计数，那么我想抛出异常。异常可以是特定的业务异常，也可以是特定的AmqpException，但我希望发布者可以在未进入超时状态时处理该异常

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

这样做是可能的，还是不是一个好的实践？

编辑1：

在@GaryRussel回复之后，我的问题就解决了：

>

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

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

}

将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();
}
}

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

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

对于RabbitTemplate，我设置了以下属性：

   rabbitTemplate.setMessageConverter(new RemoteInvocationAwareMessageConverterAdapter());

当消息受到消费者的威胁，但它发送了一个错误，我获得一个远程调用结果，其中包含原始异常到e.get原因（）。

Answer 1

对我起作用的是：

在“服务”方面：

>

服务

@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);
}

Answer 2

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

Answer 3

请参阅@RabbitListener（自2.0以来）上的returExceptions属性。文档在这里。

returnExceptions属性，当true时，将导致异常返回给发送方。异常被包装在RemoteInvocationResult对象中。

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

重要的

这种机制通常只适用于默认的SimpleMessageConverter，它使用Java序列化；异常通常不“Jackson友好”，因此无法序列化为JSON。如果您正在使用JSON，请考虑使用Error Healthor在抛出异常时返回其他杰克逊友好的错误对象。