无法使用 spring 云流执行任务
Not able to execute task using spring cloud stream
我有三个应用程序作为源、处理器和接收器。源能够将消息传递给处理器，但处理器无法将消息发送到接收器并抛出异常。尝试在本地使用 spring 云流执行任务，所以我的接收器 pom 文件中有 spring-cloud-deployer-local 并且主 class 也用 @EnableTaskLauncher 注释。如果有人可以提供相同的 samole，那也会有所帮助。
处理器
@SpringBootApplication
@EnableBinding(Processor.class)
public class ProcessorAppApplication {
@Autowired
Processor channel;
public static void main(String[] args) {
SpringApplication.run(ProcessorAppApplication.class, args);
}
@Transformer(inputChannel=Processor.INPUT,outputChannel=Processor.OUTPUT)
public void trans(String st)
{
System.out.println(st);
String url = "maven://demo:springcloud-m3-task:jar:0.0.1-SNAPSHOT";
TaskLaunchRequest request = new TaskLaunchRequest(url, null, null, null);
System.out.println("created task launch request ...");
GenericMessage<TaskLaunchRequest> message = new GenericMessage<>(request);
this.channel.output().send(message);
//return "request sent";
}
}
来源错误
2017-08-23 10:48:33.019 INFO 8632 --- [ main] o.s.integration.channel.DirectChannel : Channel 'stream.source.stream.bridge' has 1 subscriber(s).
2017-08-23 10:48:33.088 INFO 8632 --- [ main] o.s.i.a.i.AmqpInboundChannelAdapter : started inbound.stream.source.stream
2017-08-23 10:48:33.088 INFO 8632 --- [ main] o.s.c.support.DefaultLifecycleProcessor : Starting beans in phase 2147483647
2017-08-23 10:48:33.089 INFO 8632 --- [ main] o.s.c.support.DefaultLifecycleProcessor : Starting beans in phase 0
2017-08-23 10:48:33.089 INFO 8632 --- [ main] o.s.c.support.DefaultLifecycleProcessor : Starting beans in phase 2147482647
2017-08-23 10:48:33.090 INFO 8632 --- [ main] o.s.c.support.DefaultLifecycleProcessor : Starting beans in phase 2147483647
2017-08-23 10:48:33.211 INFO 8632 --- [ main] s.b.c.e.t.TomcatEmbeddedServletContainer : Tomcat started on port(s): 23381 (http)
2017-08-23 10:48:33.215 INFO 8632 --- [ main] c.example.demo.ProcessorAppApplication : Started ProcessorAppApplication in 38.865 seconds (JVM running for 40.371)
this is from source to processor..
created task launch request ...
2017-08-23 11:05:45.176 INFO 8632 --- [source.stream-1] o.s.i.codec.kryo.CompositeKryoRegistrar : registering [40, java.io.File] with serializer org.springframework.integration.codec.kryo.FileSerializer
this is from source to processor..
created task launch request ...
this is from source to processor..
created task launch request ...
2017-08-23 11:05:48.272 WARN 8632 --- [source.stream-1] o.s.a.r.r.RejectAndDontRequeueRecoverer : Retries exhausted for message (Body:'this is from source to processor..' MessageProperties [headers={contentType=text/plain}, timestamp=null, messageId=null, userId=null, receivedUserId=null, appId=null, clusterId=null, type=null, correlationId=null, correlationIdString=null, replyTo=null, contentType=text/plain, contentEncoding=null, contentLength=0, deliveryMode=null, receivedDeliveryMode=PERSISTENT, expiration=null, priority=0, redelivered=false, receivedExchange=stream.source, receivedRoutingKey=stream.source, receivedDelay=null, deliveryTag=1, messageCount=0, consumerTag=amq.ctag-7vJ-EEX8fImoqdXZRYClNw, consumerQueue=stream.source.stream])
org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException: Listener threw exception
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.wrapToListenerExecutionFailedExceptionIfNeeded(AbstractMessageListenerContainer.java:915) [spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:825) [spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:745) [spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access[=12=]1(SimpleMessageListenerContainer.java:97) [spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.invokeListener(SimpleMessageListenerContainer.java:189) ~[spring-rabbit-1.7.3.RELEASE.jar!/:na]
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:1.8.0_101]
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[na:1.8.0_101]
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:1.8.0_101]
at java.lang.reflect.Method.invoke(Method.java:498) ~[na:1.8.0_101]
at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:333) ~[spring-aop-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:190) [spring-aop-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:157) [spring-aop-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at org.springframework.retry.interceptor.RetryOperationsInterceptor.doWithRetry(RetryOperationsInterceptor.java:91) ~[spring-retry-1.2.1.RELEASE.jar!/:na]
at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:287) [spring-retry-1.2.1.RELEASE.jar!/:na]
at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:180) [spring-retry-1.2.1.RELEASE.jar!/:na]
at org.springframework.retry.interceptor.RetryOperationsInterceptor.invoke(RetryOperationsInterceptor.java:115) [spring-retry-1.2.1.RELEASE.jar!/:na]
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:179) [spring-aop-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:213) [spring-aop-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at com.sun.proxy.$Proxy103.invokeListener(Unknown Source) [na:na]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.invokeListener(SimpleMessageListenerContainer.java:1276) [spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:726) [spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:1219) [spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:1189) [spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access00(SimpleMessageListenerContainer.java:97) [spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1421) [spring-rabbit-1.7.3.RELEASE.jar!/:na]
at java.lang.Thread.run(Thread.java:745) [na:1.8.0_101]
Caused by: org.springframework.integration.handler.ReplyRequiredException: No reply produced by handler 'processorAppApplication.trans.transformer.handler', and its 'requiresReply' property is set to true.
at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:119) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:127) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:116) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:148) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:121) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:89) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:423) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:373) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:115) ~[spring-messaging-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:45) ~[spring-messaging-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:105) ~[spring-messaging-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutput(AbstractMessageProducingHandler.java:360) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.handler.AbstractMessageProducingHandler.produceOutput(AbstractMessageProducingHandler.java:271) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutputs(AbstractMessageProducingHandler.java:188) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:115) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:127) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:116) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:148) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:121) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:89) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:423) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:373) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:115) ~[spring-messaging-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:45) ~[spring-messaging-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:105) ~[spring-messaging-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at org.springframework.integration.endpoint.MessageProducerSupport.sendMessage(MessageProducerSupport.java:188) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter.access00(AmqpInboundChannelAdapter.java:56) ~[spring-integration-amqp-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.processMessage(AmqpInboundChannelAdapter.java:246) ~[spring-integration-amqp-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.onMessage(AmqpInboundChannelAdapter.java:203) ~[spring-integration-amqp-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:822) [spring-rabbit-1.7.3.RELEASE.jar!/:na]
... 24 common frames omitted
2017-08-23 11:05:48.276 WARN 8632 --- [source.stream-1] s.a.r.l.ConditionalRejectingErrorHandler : Execution of Rabbit message listener failed.
org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException: Retry Policy Exhausted
at org.springframework.amqp.rabbit.retry.RejectAndDontRequeueRecoverer.recover(RejectAndDontRequeueRecoverer.java:45) ~[spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.amqp.rabbit.config.StatelessRetryOperationsInterceptorFactoryBean.recover(StatelessRetryOperationsInterceptorFactoryBean.java:66) ~[spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.amqp.rabbit.config.StatelessRetryOperationsInterceptorFactoryBean.recover(StatelessRetryOperationsInterceptorFactoryBean.java:59) ~[spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.retry.interceptor.RetryOperationsInterceptor$ItemRecovererCallback.recover(RetryOperationsInterceptor.java:141) ~[spring-retry-1.2.1.RELEASE.jar!/:na]
at org.springframework.retry.support.RetryTemplate.handleRetryExhausted(RetryTemplate.java:512) ~[spring-retry-1.2.1.RELEASE.jar!/:na]
at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:351) ~[spring-retry-1.2.1.RELEASE.jar!/:na]
at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:180) ~[spring-retry-1.2.1.RELEASE.jar!/:na]
at org.springframework.retry.interceptor.RetryOperationsInterceptor.invoke(RetryOperationsInterceptor.java:115) ~[spring-retry-1.2.1.RELEASE.jar!/:na]
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:179) ~[spring-aop-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:213) ~[spring-aop-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at com.sun.proxy.$Proxy103.invokeListener(Unknown Source) ~[na:na]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.invokeListener(SimpleMessageListenerContainer.java:1276) [spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:726) ~[spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:1219) [spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:1189) [spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access00(SimpleMessageListenerContainer.java:97) [spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1421) [spring-rabbit-1.7.3.RELEASE.jar!/:na]
at java.lang.Thread.run(Thread.java:745) [na:1.8.0_101]
Caused by: org.springframework.amqp.AmqpRejectAndDontRequeueException: org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException: Listener threw exception
... 18 common frames omitted
Caused by: org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException: Listener threw exception
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.wrapToListenerExecutionFailedExceptionIfNeeded(AbstractMessageListenerContainer.java:915) ~[spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:825) ~[spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:745) ~[spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access[=12=]1(SimpleMessageListenerContainer.java:97) [spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.invokeListener(SimpleMessageListenerContainer.java:189) ~[spring-rabbit-1.7.3.RELEASE.jar!/:na]
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:1.8.0_101]
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[na:1.8.0_101]
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:1.8.0_101]
at java.lang.reflect.Method.invoke(Method.java:498) ~[na:1.8.0_101]
at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:333) ~[spring-aop-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:190) ~[spring-aop-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:157) ~[spring-aop-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at org.springframework.retry.interceptor.RetryOperationsInterceptor.doWithRetry(RetryOperationsInterceptor.java:91) ~[spring-retry-1.2.1.RELEASE.jar!/:na]
at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:287) ~[spring-retry-1.2.1.RELEASE.jar!/:na]
... 12 common frames omitted
Caused by: org.springframework.integration.handler.ReplyRequiredException: No reply produced by handler 'processorAppApplication.trans.transformer.handler', and its 'requiresReply' property is set to true.
at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:119) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:127) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:116) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:148) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:121) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:89) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:423) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:373) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:115) ~[spring-messaging-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:45) ~[spring-messaging-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:105) ~[spring-messaging-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutput(AbstractMessageProducingHandler.java:360) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.handler.AbstractMessageProducingHandler.produceOutput(AbstractMessageProducingHandler.java:271) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutputs(AbstractMessageProducingHandler.java:188) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:115) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:127) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:116) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:148) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:121) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:89) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:423) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:373) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:115) ~[spring-messaging-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:45) ~[spring-messaging-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:105) ~[spring-messaging-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at org.springframework.integration.endpoint.MessageProducerSupport.sendMessage(MessageProducerSupport.java:188) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter.access00(AmqpInboundChannelAdapter.java:56) ~[spring-integration-amqp-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.processMessage(AmqpInboundChannelAdapter.java:246) ~[spring-integration-amqp-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.onMessage(AmqpInboundChannelAdapter.java:203) ~[spring-integration-amqp-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:822) ~[spring-rabbit-1.7.3.RELEASE.jar!/:na]
... 24 common frames omitted
请有人调查一下，需要更多信息请告诉我
我想您可以在您的用例中使用这个现成的 task launcher transformer 处理器。或者，您想实现此处理器无法实现的任何特定功能吗？
我有三个应用程序作为源、处理器和接收器。源能够将消息传递给处理器，但处理器无法将消息发送到接收器并抛出异常。尝试在本地使用 spring 云流执行任务，所以我的接收器 pom 文件中有 spring-cloud-deployer-local 并且主 class 也用 @EnableTaskLauncher 注释。如果有人可以提供相同的 samole，那也会有所帮助。
处理器
@SpringBootApplication
@EnableBinding(Processor.class)
public class ProcessorAppApplication {
@Autowired
Processor channel;
public static void main(String[] args) {
SpringApplication.run(ProcessorAppApplication.class, args);
}
@Transformer(inputChannel=Processor.INPUT,outputChannel=Processor.OUTPUT)
public void trans(String st)
{
System.out.println(st);
String url = "maven://demo:springcloud-m3-task:jar:0.0.1-SNAPSHOT";
TaskLaunchRequest request = new TaskLaunchRequest(url, null, null, null);
System.out.println("created task launch request ...");
GenericMessage<TaskLaunchRequest> message = new GenericMessage<>(request);
this.channel.output().send(message);
//return "request sent";
}
}
来源错误
2017-08-23 10:48:33.019 INFO 8632 --- [ main] o.s.integration.channel.DirectChannel : Channel 'stream.source.stream.bridge' has 1 subscriber(s).
2017-08-23 10:48:33.088 INFO 8632 --- [ main] o.s.i.a.i.AmqpInboundChannelAdapter : started inbound.stream.source.stream
2017-08-23 10:48:33.088 INFO 8632 --- [ main] o.s.c.support.DefaultLifecycleProcessor : Starting beans in phase 2147483647
2017-08-23 10:48:33.089 INFO 8632 --- [ main] o.s.c.support.DefaultLifecycleProcessor : Starting beans in phase 0
2017-08-23 10:48:33.089 INFO 8632 --- [ main] o.s.c.support.DefaultLifecycleProcessor : Starting beans in phase 2147482647
2017-08-23 10:48:33.090 INFO 8632 --- [ main] o.s.c.support.DefaultLifecycleProcessor : Starting beans in phase 2147483647
2017-08-23 10:48:33.211 INFO 8632 --- [ main] s.b.c.e.t.TomcatEmbeddedServletContainer : Tomcat started on port(s): 23381 (http)
2017-08-23 10:48:33.215 INFO 8632 --- [ main] c.example.demo.ProcessorAppApplication : Started ProcessorAppApplication in 38.865 seconds (JVM running for 40.371)
this is from source to processor..
created task launch request ...
2017-08-23 11:05:45.176 INFO 8632 --- [source.stream-1] o.s.i.codec.kryo.CompositeKryoRegistrar : registering [40, java.io.File] with serializer org.springframework.integration.codec.kryo.FileSerializer
this is from source to processor..
created task launch request ...
this is from source to processor..
created task launch request ...
2017-08-23 11:05:48.272 WARN 8632 --- [source.stream-1] o.s.a.r.r.RejectAndDontRequeueRecoverer : Retries exhausted for message (Body:'this is from source to processor..' MessageProperties [headers={contentType=text/plain}, timestamp=null, messageId=null, userId=null, receivedUserId=null, appId=null, clusterId=null, type=null, correlationId=null, correlationIdString=null, replyTo=null, contentType=text/plain, contentEncoding=null, contentLength=0, deliveryMode=null, receivedDeliveryMode=PERSISTENT, expiration=null, priority=0, redelivered=false, receivedExchange=stream.source, receivedRoutingKey=stream.source, receivedDelay=null, deliveryTag=1, messageCount=0, consumerTag=amq.ctag-7vJ-EEX8fImoqdXZRYClNw, consumerQueue=stream.source.stream])
org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException: Listener threw exception
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.wrapToListenerExecutionFailedExceptionIfNeeded(AbstractMessageListenerContainer.java:915) [spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:825) [spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:745) [spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access[=12=]1(SimpleMessageListenerContainer.java:97) [spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.invokeListener(SimpleMessageListenerContainer.java:189) ~[spring-rabbit-1.7.3.RELEASE.jar!/:na]
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:1.8.0_101]
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[na:1.8.0_101]
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:1.8.0_101]
at java.lang.reflect.Method.invoke(Method.java:498) ~[na:1.8.0_101]
at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:333) ~[spring-aop-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:190) [spring-aop-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:157) [spring-aop-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at org.springframework.retry.interceptor.RetryOperationsInterceptor.doWithRetry(RetryOperationsInterceptor.java:91) ~[spring-retry-1.2.1.RELEASE.jar!/:na]
at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:287) [spring-retry-1.2.1.RELEASE.jar!/:na]
at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:180) [spring-retry-1.2.1.RELEASE.jar!/:na]
at org.springframework.retry.interceptor.RetryOperationsInterceptor.invoke(RetryOperationsInterceptor.java:115) [spring-retry-1.2.1.RELEASE.jar!/:na]
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:179) [spring-aop-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:213) [spring-aop-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at com.sun.proxy.$Proxy103.invokeListener(Unknown Source) [na:na]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.invokeListener(SimpleMessageListenerContainer.java:1276) [spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:726) [spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:1219) [spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:1189) [spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access00(SimpleMessageListenerContainer.java:97) [spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1421) [spring-rabbit-1.7.3.RELEASE.jar!/:na]
at java.lang.Thread.run(Thread.java:745) [na:1.8.0_101]
Caused by: org.springframework.integration.handler.ReplyRequiredException: No reply produced by handler 'processorAppApplication.trans.transformer.handler', and its 'requiresReply' property is set to true.
at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:119) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:127) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:116) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:148) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:121) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:89) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:423) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:373) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:115) ~[spring-messaging-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:45) ~[spring-messaging-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:105) ~[spring-messaging-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutput(AbstractMessageProducingHandler.java:360) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.handler.AbstractMessageProducingHandler.produceOutput(AbstractMessageProducingHandler.java:271) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutputs(AbstractMessageProducingHandler.java:188) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:115) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:127) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:116) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:148) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:121) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:89) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:423) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:373) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:115) ~[spring-messaging-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:45) ~[spring-messaging-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:105) ~[spring-messaging-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at org.springframework.integration.endpoint.MessageProducerSupport.sendMessage(MessageProducerSupport.java:188) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter.access00(AmqpInboundChannelAdapter.java:56) ~[spring-integration-amqp-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.processMessage(AmqpInboundChannelAdapter.java:246) ~[spring-integration-amqp-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.onMessage(AmqpInboundChannelAdapter.java:203) ~[spring-integration-amqp-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:822) [spring-rabbit-1.7.3.RELEASE.jar!/:na]
... 24 common frames omitted
2017-08-23 11:05:48.276 WARN 8632 --- [source.stream-1] s.a.r.l.ConditionalRejectingErrorHandler : Execution of Rabbit message listener failed.
org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException: Retry Policy Exhausted
at org.springframework.amqp.rabbit.retry.RejectAndDontRequeueRecoverer.recover(RejectAndDontRequeueRecoverer.java:45) ~[spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.amqp.rabbit.config.StatelessRetryOperationsInterceptorFactoryBean.recover(StatelessRetryOperationsInterceptorFactoryBean.java:66) ~[spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.amqp.rabbit.config.StatelessRetryOperationsInterceptorFactoryBean.recover(StatelessRetryOperationsInterceptorFactoryBean.java:59) ~[spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.retry.interceptor.RetryOperationsInterceptor$ItemRecovererCallback.recover(RetryOperationsInterceptor.java:141) ~[spring-retry-1.2.1.RELEASE.jar!/:na]
at org.springframework.retry.support.RetryTemplate.handleRetryExhausted(RetryTemplate.java:512) ~[spring-retry-1.2.1.RELEASE.jar!/:na]
at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:351) ~[spring-retry-1.2.1.RELEASE.jar!/:na]
at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:180) ~[spring-retry-1.2.1.RELEASE.jar!/:na]
at org.springframework.retry.interceptor.RetryOperationsInterceptor.invoke(RetryOperationsInterceptor.java:115) ~[spring-retry-1.2.1.RELEASE.jar!/:na]
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:179) ~[spring-aop-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:213) ~[spring-aop-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at com.sun.proxy.$Proxy103.invokeListener(Unknown Source) ~[na:na]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.invokeListener(SimpleMessageListenerContainer.java:1276) [spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:726) ~[spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:1219) [spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:1189) [spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access00(SimpleMessageListenerContainer.java:97) [spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1421) [spring-rabbit-1.7.3.RELEASE.jar!/:na]
at java.lang.Thread.run(Thread.java:745) [na:1.8.0_101]
Caused by: org.springframework.amqp.AmqpRejectAndDontRequeueException: org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException: Listener threw exception
... 18 common frames omitted
Caused by: org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException: Listener threw exception
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.wrapToListenerExecutionFailedExceptionIfNeeded(AbstractMessageListenerContainer.java:915) ~[spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:825) ~[spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:745) ~[spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access[=12=]1(SimpleMessageListenerContainer.java:97) [spring-rabbit-1.7.3.RELEASE.jar!/:na]
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.invokeListener(SimpleMessageListenerContainer.java:189) ~[spring-rabbit-1.7.3.RELEASE.jar!/:na]
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:1.8.0_101]
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[na:1.8.0_101]
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:1.8.0_101]
at java.lang.reflect.Method.invoke(Method.java:498) ~[na:1.8.0_101]
at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:333) ~[spring-aop-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:190) ~[spring-aop-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:157) ~[spring-aop-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at org.springframework.retry.interceptor.RetryOperationsInterceptor.doWithRetry(RetryOperationsInterceptor.java:91) ~[spring-retry-1.2.1.RELEASE.jar!/:na]
at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:287) ~[spring-retry-1.2.1.RELEASE.jar!/:na]
... 12 common frames omitted
Caused by: org.springframework.integration.handler.ReplyRequiredException: No reply produced by handler 'processorAppApplication.trans.transformer.handler', and its 'requiresReply' property is set to true.
at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:119) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:127) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:116) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:148) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:121) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:89) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:423) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:373) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:115) ~[spring-messaging-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:45) ~[spring-messaging-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:105) ~[spring-messaging-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutput(AbstractMessageProducingHandler.java:360) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.handler.AbstractMessageProducingHandler.produceOutput(AbstractMessageProducingHandler.java:271) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutputs(AbstractMessageProducingHandler.java:188) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:115) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:127) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:116) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:148) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:121) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:89) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:423) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:373) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:115) ~[spring-messaging-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:45) ~[spring-messaging-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:105) ~[spring-messaging-4.3.10.RELEASE.jar!/:4.3.10.RELEASE]
at org.springframework.integration.endpoint.MessageProducerSupport.sendMessage(MessageProducerSupport.java:188) ~[spring-integration-core-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter.access00(AmqpInboundChannelAdapter.java:56) ~[spring-integration-amqp-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.processMessage(AmqpInboundChannelAdapter.java:246) ~[spring-integration-amqp-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.onMessage(AmqpInboundChannelAdapter.java:203) ~[spring-integration-amqp-4.3.11.RELEASE.jar!/:4.3.11.RELEASE]
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:822) ~[spring-rabbit-1.7.3.RELEASE.jar!/:na]
... 24 common frames omitted
请有人调查一下，需要更多信息请告诉我
我想您可以在您的用例中使用这个现成的 task launcher transformer 处理器。或者，您想实现此处理器无法实现的任何特定功能吗？