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

Question

我正在使用spring-cloud-stream:1.3.0.发行版、spring-cloud-stream-binder-kafka:1.3.0.发行版开发spring引导应用程序。我使用spring integration 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中，我注意到一个pull请求https://github.com/spring-cloud/spring-cloud-stream/pull/1039，它在errorChannelEnabled设置为true时创建PublishSubscribeChannel，还有一个testcase检查是否为生产者错误通道创建了bean。

当我在应用程序http://localhost:8195/beans中检查spring Actuctor url时，会创建全局错误通道的“ErrorChannel”bean，但不会创建“WebMarketBasket.Errors”bean。当存在“org.springframework.messaging.messagehandlingexception”时，错误消息被发送到“errorchannel”kafka主题，并停止处理文件中的剩余行。Kafka主题“WebMarketBasketErrors”从未创建。你能帮忙吗，如果我错过了什么，请告诉我。

Answer 1

你好像混淆了两件事。

Spring Integration Error Channel支持是针对希望将某些内容发布到全局ErrorChannel时提供的，全局ErrorChannel通过Spring.Cloud.Stream.Bindings.Error.destination=MyErrors绑定到目标。

1.3中的新支持为每个侦听器创建一个错误通道；它是pub/sub，并且桥接到全局errorchannel。因此，该消息还将发布到...bindings.error.destination（如果已配置，则目标名为errorchannel)。专用错误通道的bean名称是webmarketbasket。USProductRemordationsGroup.Errors。框架没有卡夫卡主题绑定到那个频道。

如果您想自己处理错误，您可以从专用错误通道或全局错误通道中使用。