Spring Cloud功能和死信交换

问题描述

我正在尝试使用RabbitMQ编写一个响应式Spring Cloud Function服务，该服务将消耗一个队列并生成一个交换。

我有2个问题。

为什么我会在日志中看到下面的错误。
我该如何拒绝doOnError？ doOnError只能访问throwable，而不能访问拒绝消息。

这是应用程序代码。它是从此问题Spring Reactive Stream - Unexpected Shutdown

复制而来的

@SpringBootApplication
public class StreamApplication {

  public static void main(String[] args) {
    SpringApplication.runStreamApplication args);
  }

  @Bean
  public Function<Flux<Message<String>>,Flux<Message<String>>> transform() {

    return inbound -> inbound.map(msg -> {
      try {
        System.out.println("ACKING MESSAGE");
        Channel channel = msg.getHeaders().get(AmqpHeaders.CHANNEL,Channel.class);
        channel.basicAck(msg.getHeaders().get(AmqpHeaders.DELIVERY_TAG,Long.class),false);
      }
      catch (IOException e) {
        e.printStackTrace();
      }
      return msg;
    });
  }
}

这是application.yml。对于来自队列的传入和用于交换的传出，它有2种不同的绑定器。

spring:
  cloud:
    stream:
      override-cloud-connectors: true
      rabbit:
        bindings:
          events-processor:
            producer:
              bindQueue: false
              declareExchange: false
              routing-key-expression: headers.eventType
          events:
            consumer:
              acknowledge-mode: MANUAL   
              prefetch: 10
              #auto-bind-dlq: false
              dead-letter-exchange: dead-letter-exchange
              bindQueue: false
              declareExchange: false
      function:
        definition: transform
        bindings:
          transform-in-0: events
          transform-out-0: events-processor
      bindings:
        events:
          destination: queue
          binder: consumerrabbit
          group: events-processor
        events-processor:
          destination: activity-events
          binder: producerrabbit
      binders:
        producerrabbit:
          defaultCandidate: false
          inheritEnvironment: false
          type: rabbit
          environment:
            spring:
              rabbitmq:
                host: host
                port: port
                username: username
                password: password
                virtual-host: virtual-host
        consumerrabbit:
          defaultCandidate: true
          inheritEnvironment: false
          type: rabbit
          environment:
            spring:
              rabbitmq:
                host: host
                port: port
                username: username
                password: password
                virtual-host: virtual-host

以下是启动日志以及应用程序何时接收事件的日志。我不确定为什么在启动过程中该频道有一个订阅者，但是一旦收到一条消息，它说它没有订阅者。

2020-09-07 12:02:04.076  INFO 10652 --- [           main] o.s.c.s.m.DirectWithAttributesChannel    : Channel 'application-1.events' has 1 subscriber(s).
2020-09-07 18:02:08.848  INFO 10652 --- [           main] o.s.c.s.m.DirectWithAttributesChannel    : Channel 'application-1.events-processor' has 1 subscriber(s).
2020-09-07 12:02:05.081  INFO 10652 --- [           main] o.s.i.monitor.IntegrationMBeanExporter   : Registering MessageChannel events
2020-09-07 12:02:05.145  INFO 10652 --- [           main] o.s.i.monitor.IntegrationMBeanExporter   : Registering MessageChannel errorChannel
2020-09-07 12:02:05.178  INFO 10652 --- [           main] o.s.i.monitor.IntegrationMBeanExporter   : Registering MessageChannel events-processor
...................other startup log messages...............
2020-09-07 18:05:09.896  INFO 10652 --- [nts-processor-1] o.s.c.s.m.DirectWithAttributesChannel    : Channel 'application-1.events' has 0 subscriber(s).
ACKING MESSAGE
2020-09-07 18:05:12.900 ERROR 10652 --- [nts-processor-1] o.s.integration.handler.LoggingHandler   : org.springframework.messaging.MessageDeliveryException: Dispatcher has no subscribers for channel 'application-1.events'.; nested exception is org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers,failedMessage=GenericMessage [payload={
  "data": {
  
  }
},headers={amqp_receivedDeliveryMode=NON_PERSISTENT,amqp_receivedExchange=source,amqp_deliveryTag=1,deliveryAttempt=3,amqp_consumerQueue=source.events-processor,amqp_channel=Cached Rabbit Channel: AMQChannel(amqp:),conn: Proxy@43e869ea Shared Rabbit Connection: SimpleConnection@213451a4 [delegate=amqp:],amqp_redelivered=false,id=35cdfa7f-7f75-5502-aede-6ec9569145f0,amqp_consumerTag=amq.ctag-Z87p6nKN4PLJKDbZQpXnVw,sourceData=(Body:'[B@20dc391b(byte[401])' MessageProperties [headers={},contentLength=0,receivedDeliveryMode=NON_PERSISTENT,redelivered=false,receivedExchange=source,receivedRoutingKey=,deliveryTag=1,consumerTag=amq.ctag-Z87p6nKN4PLJKDbZQpXnVw,consumerQueue=source.events-processor]),contentType=application/json,timestamp=1599501912897}],amqp_channel=Cached Rabbit Channel: AMQChannel(amqp://H),conn: Proxy@43e869ea Shared Rabbit Connection: SimpleConnection@213451a4 [delegate=amqp://],timestamp=1599501912897}]

解决方法

弄清楚了如何使函数在失败时将消息发送到DLQ。我也添加了一个消费者，因为它们是相关的。

我相信我们需要确认或拒绝该消息，但是在拒绝时，我们想返回一个Flux.empty（），因此没有任何内容发布到下游交易所。

代码拒绝任何失败的消息作为有效负载，并终止其他任何消息。

@SpringBootApplication
public class DemoApplication {

  public static void main(String[] args) {
    SpringApplication.run(DemoApplication.class,args);
  }

  @Bean
  public Function<Flux<Message<String>>,Flux<Message<String>>> transform() {
    return inbound -> inbound.flatMap(msg -> {
      try {
        Channel channel = msg.getHeaders().get(AmqpHeaders.CHANNEL,Channel.class);
        Long deliveryTag = msg.getHeaders().get(AmqpHeaders.DELIVERY_TAG,Long.class);

        if (msg.getPayload().startsWith("fail")) {
          channel.basicReject(deliveryTag,false);
        } else {
          channel.basicAck(deliveryTag,false);
          return Flux.just(msg);
        }
      } catch (IOException e) {
        e.printStackTrace();
      }

      return Flux.empty();
    });
  }

  @Bean
  public Consumer<Flux<Message<String>>> consumer() {
    return inbound -> inbound
        .map(msg -> {
          try {
            Channel channel = msg.getHeaders().get(AmqpHeaders.CHANNEL,Channel.class);
            Long deliveryTag = msg.getHeaders().get(AmqpHeaders.DELIVERY_TAG,Long.class);

            if (msg.getPayload().startsWith("fail")) {
              channel.basicReject(deliveryTag,false);
            } else {
              channel.basicAck(deliveryTag,false);
            }
          } catch (IOException e) {
            e.printStackTrace();
          }
          return msg.getPayload();
        })
        .subscribe(System.out::println);
  }
}

spring:
  profiles: consumer
  cloud:
    stream:
      function:
        definition: consumer
      bindings:
        consumer-in-0:
          destination: consumer
          group: queue
      rabbit:
        bindings:
          consumer-in-0:
            consumer:
              acknowledge-mode: MANUAL
              autoBindDlq: true
              declareDlx: true
              bindQueue: true
              deadLetterExchange: dead-letter-exchange
              prefetch: 10
---
spring:
  profiles: processor
  cloud:
    stream:
      function:
        definition: transform
      bindings:
        transform-in-0:
          destination: processorIn
          group: queue
        transform-out-0:
          destination: processorOut
      rabbit:
        bindings:
          transform-in-0:
            consumer:
              acknowledge-mode: MANUAL
              autoBindDlq: true
              declareDlx: true
              bindQueue: true
              deadLetterExchange: dead-letter-exchange
              prefetch: 10
          transform-out-0:
            producer:
              bindQueue: true
              declareExchange: true

spring-cloud-function spring-cloud-stream spring-rabbitmq spring-reactive

Spring Cloud功能和死信交换

问题描述

解决方法

相关问答