使用spring云流向错误通道发送错误消息

Question

我正在使用spring -cloud-stream:1.3.0开发spring启动应用程序。messages，spring-cloud-stream-binder-kafka:1.3.0.RELEASE.我使用spring集成dsl来拆分文件中的行，并使用beanio将行转换为json，要求将成功的json消息写入kafka主题，并将错误消息写入不同的kafka主题。下面是application.yml中的配置。

spring:
  cloud:
    stream:
      kafka:
        binder:
          autoAddPartitions: true
      bindings.webmarketbasket:
        destination: webmarketbasket
        group: usproductrecommendationsgroup
        producer:
          partitionCount: 5
          errorChannelEnabled: true
      bindings.webmarketbasket.errors:
        destination: webmarketbasketerrors
        group: usproductrecommendationsgroup
        producer:
          partitionCount: 5
      bindings.error:
        destination: errorchannel
        group: usproductrecommendationsgroup
        producer:
          partitionCount: 5

我注意到spring-cloud-stream-binder-kafka:1.3.0.RELEASE，中有一个拉取请求https://github.com/spring-cloud/spring-cloud-stream/pull/1039，它在errorChannelEnabled设置为true时创建PublishSubscribeChannel，还有一个测试用例检查是否为生产者错误通道创建了bean。

当我在我的应用程序http://localhost:8195/beans中检查弹簧执行器url时，创建了全局错误通道的"errorChannel“bean，但没有创建"webmarketbasket.errors”bean。当出现"org.springframework.messaging.MessageHandlingException"，错误时，会向"errorchannel“kafka主题发送错误消息，并停止处理文件中的剩余行。Kafka主题"webmarketbasketerrors“从未创建过。你能帮忙吗?如果我错过了什么，请告诉我。

Answer 1

你似乎把两件事搞混了。

当您想要将某些内容发布到全局errorChannel，而全局via通过spring.cloud.stream.bindings.error.destination=myErrors绑定到目的地时，就可以使用Spring Integration Error Channel Support。

new support in 1.3为每个侦听器创建一个错误通道；它是发布/订阅的，并且桥接到全局errorChannel。因此，消息也将发布到...bindings.error.destination (如果已配置；在您的示例中，目的地称为errorChannel)。专用错误通道的bean名称是webmarketbasket. usproductrecommendationsgroup.errors。框架没有绑定到该通道的kafka主题。

如果您希望自己处理错误，则可以使用专用错误通道或全局错误通道。