春季启动rabbitMq setExceptionHandler

Question

如何使用 Spring 框架转换以下代码？

ConnectionFactory factory = new ConnectionFactory();
factory.setExceptionHandler(new BrokerExceptionHandler(logger, instance));

public final class BrokerExceptionHandler extends StrictExceptionHandler {
   @Override
   public void handleReturnListenerException(Channel channel, Throwable exception) {
        logger.log(Level.SEVERE, "ReturnListenerException detected: ReturnListener.handleReturn", exception);
        this.publishAlert(exception, "ReturnListener.handleReturn");
        logger.log(Level.SEVERE, "Close application", exception);
        System.exit(-1);
   }
  ....
}

如果发生rabbitMQ异常，基本上我需要指定一个自定义异常处理程序，然后停止应用程序

每次出现异常时如何发布rabbitMq消息？

编辑

我以这种方式修改了我的配置类：

@Bean
SimpleMessageListenerContainer containerPredict(ConnectionFactory connectionFactory,
  MessageListenerAdapter listenerPredictAdapter) {
     SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
     container.setConnectionFactory(connectionFactory);
     container.setDefaultRequeueRejected(false);
     container.setErrorHandler(new BrokerExceptionHandler());
     container.setQueueNames(getQueueName());
     container.setMessageListener(listenerAdapter);
     return container;
}

这是我的 BrokerExceptionHandler 类

public class BrokerExceptionHandler implements ErrorHandler {
   private final Logger logger = Logger.getLogger(getClass().getSimpleName());

   @Autowired
   private Helper helper;

   @Override
   public void handleError(Throwable t) {
     logger.log(Level.SEVERE, "Exception Detected. Publishing error alert");
     String message = "Exception detected. Message: " + t.getMessage());
     
     // Notify the error to the System sending a new RabbitMq message
     System.out.println("---> Before convertAndSend");
     rabbitTemplate.convertAndSend(exchange, routing, message);
     System.out.println("---> After convertAndSend");
   }
}

我可以在控制台看到日志Exception Detected. Publishing error alert和---> Before convertAdnSend，但是新的警报没有发布，日志---> After convertAndSend也没有出现在控制台中。

这是日志：

2018-10-17 09:32:02.849 错误 1506 --- [tainer****-1] BrokerExceptionHandler：检测到异常。发布错误提示

---> 在 convertAndSend 之前

2018-10-17 09:32:02.853 INFO 1506 --- [tainer****-1] osarlSimpleMessageListenerContainer：重新启动 Consumer@4f5b08d：tags=[{amq.ctag-yUcUmg5BCo20ucG1wJZoWA=myechange}]，频道=Cached Rabbit Channel：AMQChannel(amqp://admin@XXX.XXX.XXX.XXX:5672/testbed_simulators,1), conn: Proxy@3964d79 Shared Rabbit Connection: SimpleConnection@61f39bb [delegate=amqp://admin@XXX .XXX.XXX.XXX:5672/testbed_simulators, localPort= 51528], acknowledgeMode=AUTO 本地队列大小=0

2018-10-17 09:32:02.905 INFO 1506 --- [tainer****-2] osamqp.rabbit.core.RabbitAdmin ：自动声明非持久、自动删除或独占队列 (myexchange) 持久：假，自动删除：真，排他：真。如果在连接工厂处于活动状态时代理停止并重新启动，它将重新声明，但所有消息都会丢失。

2018-10-17 09:32:02.905 INFO 1506 --- [tainer****-2] osamqp.rabbit.core.RabbitAdmin ：自动声明非持久、自动删除或独占队列 (myexchange) 持久：假，自动删除：真，排他：真。如果在连接工厂处于活动状态时代理停止并重新启动，它将重新声明，但所有消息都会丢失。

编辑

调试我看到在发送新消息之前，调用了以下代码：

File: SimpleMessageListenerContainer.class line 1212

if (!isActive(this.consumer) || aborted) {
  .....
}
else {
  ---> logger.info("Restarting " + this.consumer);
       restart(this.consumer);
}

编辑 2

示例代码：http://github.com/fabry00/spring-boot-rabbitmq

Answer 1

这取决于您如何进行配置；如果您使用的是 Spring Boot 的自动配置连接工厂...

@Bean
public InitializingBean connectionFactoryConfigurer(CachingConnectionFactory ccf) {
    return () -> ccf.getRabbitConnectionFactory().setExceptionHandler(...);
}

如果您要连接自己的 bean（例如通过 RabbitConnectionFactoryBean），则直接设置它。

编辑

您正在错误处理程序中抛出NullPointerException...

2018-10-17 11:51:58.733 DEBUG 38975 --- [containerKpis-1] o.s.a.r.l.SimpleMessageListenerContainer : Consumer raised exception, processing can restart if the connection factory supports it

java.lang.NullPointerException: null
    at com.test.BrokerExceptionHandler.handleError(BrokerExceptionHandler.java:27) ~[main/:na]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeErrorHandler(AbstractMessageListenerContainer.java:1243) ~[spring-rabbit-2.0.6.RELEASE.jar:2.0.6.RELEASE]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.handleListenerException(AbstractMessageListenerContainer.java:1488) ~[spring-rabbit-2.0.6.RELEASE.jar:2.0.6.RELEASE]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1318) ~[spring-rabbit-2.0.6.RELEASE.jar:2.0.6.RELEASE]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:817) ~[spring-rabbit-2.0.6.RELEASE.jar:2.0.6.RELEASE]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:801) ~[spring-rabbit-2.0.6.RELEASE.jar:2.0.6.RELEASE]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$700(SimpleMessageListenerContainer.java:77) ~[spring-rabbit-2.0.6.RELEASE.jar:2.0.6.RELEASE]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1042) ~[spring-rabbit-2.0.6.RELEASE.jar:2.0.6.RELEASE]
    at java.lang.Thread.run(Thread.java:748) [na:1.8.0_131]

2018-10-17 11:51:58.734  INFO 38975 --- [containerKpis-1] o.s.a.r.l.SimpleMessageListenerContainer : Restarting Consumer@1aabf50d: tags=[{amq.ctag-VxxHKiMsWI_w8DIooAsySA=myapp.mydomain.KPIS}], channel=Cached Rabbit Channel: AMQChannel(amqp://guest@127.0.0.1:5672/,1), conn: Proxy@b88a7d6 Shared Rabbit Connection: SimpleConnection@25dc64a [delegate=amqp://guest@127.0.0.1:5672/, localPort= 55662], acknowledgeMode=AUTO local queue size=0

要开启调试日志，添加

logging.level.org.springframework.amqp=debug

致您的application.properties。

this.helper 是 null 因为错误处理程序不是 Spring Bean - @Autowired 仅在 Spring 管理对象时才有效；你正在使用new BrokerExceptionHandler()。

EDIT2

我加了这两个豆子

@Bean
public BrokerExceptionHandler errorHandler() {
    return new BrokerExceptionHandler();
}

@Bean
public MessageConverter json() { // Boot auto-configures in template
    return new Jackson2JsonMessageConverter();
}

现在……

---> Before publishing Alert event
--- ALERT
2018-10-17 12:14:45.304  INFO 43359 --- [containerKpis-1] Helper                                   : publishAlert
2018-10-17 12:14:45.321 DEBUG 43359 --- [containerKpis-1] o.s.a.r.c.CachingConnectionFactory       : Creating cached Rabbit Channel from AMQChannel(amqp://guest@127.0.0.1:5672/,3)
2018-10-17 12:14:45.321 DEBUG 43359 --- [containerKpis-1] o.s.amqp.rabbit.core.RabbitTemplate      : Executing callback RabbitTemplate$$Lambda$638/975724213 on RabbitMQ Channel: Cached Rabbit Channel: AMQChannel(amqp://guest@127.0.0.1:5672/,3), conn: Proxy@77f3f419 Shared Rabbit Connection: SimpleConnection@10c86af1 [delegate=amqp://guest@127.0.0.1:5672/, localPort= 56220]
2018-10-17 12:14:45.321 DEBUG 43359 --- [containerKpis-1] o.s.amqp.rabbit.core.RabbitTemplate      : Publishing message (Body:'{"timestamp":1539792885303,"code":"ERROR","severity":"ERROR","message":"Exception detected. Message: Listener method 'kpisEvent' threw exception"}' MessageProperties [headers={sender=myapp, protocolVersion=1.0.0, senderType=MY_COMPONENT_1, __TypeId__=com.test.domain.Alert, timestamp=1539792885304}, contentType=application/json, contentEncoding=UTF-8, contentLength=146, deliveryMode=PERSISTENT, priority=0, deliveryTag=0])on exchange [myevent.ALERT], routingKey = [/]
--- ALERT 2
---> After publishing Alert event
2018-10-17 12:14:45.323 DEBUG 43359 --- [pool-1-thread-6] o.s.a.r.listener.BlockingQueueConsumer   : Storing delivery for consumerTag: 'amq.ctag-eYbzZ09pCw3cjdtSprlZMQ' with deliveryTag: '1' in Consumer@4b790d86: tags=[{amq.ctag-eYbzZ09pCw3cjdtSprlZMQ=myapp.myevent.ALERT}], channel=Cached Rabbit Channel: AMQChannel(amqp://guest@127.0.0.1:5672/,2), conn: Proxy@77f3f419 Shared Rabbit Connection: SimpleConnection@10c86af1 [delegate=amqp://guest@127.0.0.1:5672/, localPort= 56220], acknowledgeMode=AUTO local queue size=0
2018-10-17 12:14:45.324 DEBUG 43359 --- [ontainerReset-1] o.s.a.r.listener.BlockingQueueConsumer   : Received message: (Body:'{"timestamp":1539792885303,"code":"ERROR","severity":"ERROR","message":"Exception detected. Message: Listener method 'kpisEvent' threw exception"}' MessageProperties [headers={sender=myapp, protocolVersion=1.0.0, senderType=MY_COMPONENT_1, __TypeId__=com.test.domain.Alert, timestamp=1539792885304}, contentType=application/json, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=myevent.ALERT, receivedRoutingKey=/, deliveryTag=1, consumerTag=amq.ctag-eYbzZ09pCw3cjdtSprlZMQ, consumerQueue=myapp.myevent.ALERT])
2018-10-17 12:14:45.324  INFO 43359 --- [ontainerReset-1] Application                              : ---> kpisAlert RECEIVED
2018-10-17 12:14:45.325 ERROR 43359 --- [ontainerReset-1] Application                              : ---> Message: Exception detected. Message: Listener method 'kpisEvent' threw exception
2018-10-17 12:14:45.326 DEBUG 43359 --- [containerKpis-1] o.s.a.r.listener.BlockingQueueConsumer   : Rejecting messages (requeue=false)

EDIT3

或者，如果您更喜欢 Gson...

  @Bean
  public MessageConverter json() {
    Gson gson = new GsonBuilder().create();
    return new MessageConverter() {

      @Override
      public Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException {
        return new Message(gson.toJson(object).getBytes(), messageProperties);
      }

      @Override
      public Object fromMessage(Message message) throws MessageConversionException {
        throw new UnsupportedOperationException();
      }

    };
  }

EDIT4

我将您应用的当前版本更改如下：

  @Bean
  public MessageConverter jsonConverter() {
    Gson gson = new GsonBuilder().create();
    EventKpisCollected collected = new EventKpisCollected();
    return new MessageConverter() {

      @Override
      public Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException {
        System.out.println("toMessage");
        return new Message(gson.toJson(object).getBytes(), messageProperties);
      }

      @Override
      public Object fromMessage(Message message) throws MessageConversionException {
        System.out.println("fromMessage");
        return collected.decode(new String(message.getBody()));
      }
    };
  }

...

  @Bean
  SimpleMessageListenerContainer containerKpis(ConnectionFactory connectionFactory,
      MessageListenerAdapter listenerKpisAdapter) {

    SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
    container.setConnectionFactory(connectionFactory);
    container.setDefaultRequeueRejected(false);
    container.setErrorHandler(errorHandler());
    container.setQueueNames(getQueueKpis());
    container.setMessageListener(listenerKpisAdapter);
    return container;
  }

  @Bean
  SimpleMessageListenerContainer containerReset(ConnectionFactory connectionFactory,
      MessageListenerAdapter listenerAlertAdapter) {

    SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
    container.setConnectionFactory(connectionFactory);
    container.setDefaultRequeueRejected(false);
    container.setErrorHandler(errorHandler());
    container.setQueueNames(getQueueAlert());
    container.setMessageListener(listenerAlertAdapter);
    return container;
  }

  @Bean
  MessageListenerAdapter listenerKpisAdapter(Application receiver) {
    MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter(receiver, "kpisEvent");
    messageListenerAdapter.setMessageConverter(jsonConverter());
    return messageListenerAdapter;
  }

  @Bean
  MessageListenerAdapter listenerAlertAdapter(Application receiver) {
    MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter(receiver, "alertEvent");
//    messageListenerAdapter.setMessageConverter(jsonConverter()); converter only handles events.
    return messageListenerAdapter;
  }

和

fromMessage
2018-10-19 13:46:53.734  INFO 10725 --- [containerKpis-1] Application                              : ---> kpisEvent RECEIVED
2018-10-19 13:46:53.734  INFO 10725 --- [containerKpis-1] Application                              : ---> kpisEvent DECODED, windowId: 1522751098000-1522752198000

通过框架完成事件解码（仅针对当前的事件 - 您将需要第二个转换器用于警报）。