正如标题所示，我正在寻找有关 DSL 流中错误处理的好示例。具体来说，我正在寻找处理来自服务激活器的错误。
示例:
IntegrationFlows.from(Amqp.inboundAdapter(simpleMessageListenerContainer()))
.transform(new JsonToObjectTransformer(AlbumDescriptor.class))
.handle(AlbumDescriptor.class, (p,h) -> transformXml(p))
.transform(new ObjectToJsonTransformer())
.handle(Amqp.outboundAdapter(rabbitTemplate).routingKey("xml-transformed"))
.get();
如果我的函数 transformXml(p) 抛出异常，我该如何让它转到特定的 errorChannel。
如果没有抛出异常，一切都会按照我想要的方式进行。
目前，如果从此函数引发异常，我会收到以下异常:
2014-09-17 15:26:28.566 WARN 16339 --- [cTaskExecutor-1] s.a.r.l.ConditionalRejectingErrorHandler : Execution of Rabbit message listener failed.
org.springframework.amqp.rabbit.listener.ListenerExecutionFailedException: Listener threw exception
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.wrapToListenerExecutionFailedExceptionIfNeeded(AbstractMessageListenerContainer.java:758)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:653)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:576)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$001(SimpleMessageListenerContainer.java:78)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$1.invokeListener(SimpleMessageListenerContainer.java:161)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.invokeListener(SimpleMessageListenerContainer.java:1177)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:559)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:950)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:934)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$600(SimpleMessageListenerContainer.java:78)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1045)
at java.lang.Thread.run(Thread.java:745)
Caused by: org.springframework.messaging.MessageHandlingException: java.lang.reflect.InvocationTargetException
at org.springframework.integration.dsl.LambdaMessageProcessor.processMessage(LambdaMessageProcessor.java:125)
at org.springframework.integration.handler.ServiceActivatingHandler.handleRequestMessage(ServiceActivatingHandler.java:71)
at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:170)
at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:78)
at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:116)
at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:101)
at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:97)
at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:77)
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:255)
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:223)
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:109)
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:44)
at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:94)
at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.sendMessage(AbstractReplyProducingMessageHandler.java:260)
at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.sendReplyMessage(AbstractReplyProducingMessageHandler.java:241)
at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.produceReply(AbstractReplyProducingMessageHandler.java:205)
at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleResult(AbstractReplyProducingMessageHandler.java:199)
at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:177)
at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:78)
at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:116)
at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:101)
at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:97)
at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:77)
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:255)
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:223)
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:109)
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:44)
at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:94)
at org.springframework.integration.endpoint.MessageProducerSupport.sendMessage(MessageProducerSupport.java:98)
at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter.access$400(AmqpInboundChannelAdapter.java:44)
at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$1.onMessage(AmqpInboundChannelAdapter.java:90)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:650)
... 10 common frames omitted
Caused by: java.lang.reflect.InvocationTargetException: null
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:483)
at org.springframework.integration.dsl.LambdaMessageProcessor.processMessage(LambdaMessageProcessor.java:122)
... 41 common frames omitted
Caused by: com.amazonaws.services.s3.model.AmazonS3Exception: The specified key does not exist. (Service: Amazon S3; Status Code: 404; Error Code: NoSuchKey; Request ID: 7029FF03D01CFB95)
at com.amazonaws.http.AmazonHttpClient.handleErrorResponse(AmazonHttpClient.java:1053)
at com.amazonaws.http.AmazonHttpClient.executeOneRequest(AmazonHttpClient.java:662)
at com.amazonaws.http.AmazonHttpClient.executeHelper(AmazonHttpClient.java:428)
at com.amazonaws.http.AmazonHttpClient.execute(AmazonHttpClient.java:290)
at com.amazonaws.services.s3.AmazonS3Client.invoke(AmazonS3Client.java:3655)
at com.amazonaws.services.s3.AmazonS3Client.getObject(AmazonS3Client.java:1132)
at com.amazonaws.services.s3.AmazonS3Client.getObject(AmazonS3Client.java:1004)
at com.emusic.indiegestion.integration.flows.AlbumTransformFlowConfig.transformXml(AlbumTransformFlowConfig.java:67)
at com.emusic.indiegestion.integration.flows.AlbumTransformFlowConfig.lambda$0(AlbumTransformFlowConfig.java:49)
at com.emusic.indiegestion.integration.flows.AlbumTransformFlowConfig$$Lambda$1/2142893855.handle(Unknown Source)
... 46 common frames omitted
最佳答案
实际上，无论是 DSL 还是原始 XML Spring Integration 配置，都没有区别。
Amqp.inboundAdapter()提供.errorChannel()选项以及 <amqp:inbound-channel-adapter>提供error-channel属性。
正如我在手册中所说，DSL 的作用与所有其他 Spring Integration 配置变体相同。所以，Error Handling这里也一样。
关于Spring Integration DSL 错误处理，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/25899044/