Spring XD 自定义 class 处理器输入类型

Question

我编写了一个处理器模块，它获取 String 有效负载和 return 结果作为 LogFileTuple 类型。这个模块很好用。
我写了另一个模块，它获取 LogFileTuple 作为输入。但是当部署流时：Source | Module1 | Module2 | Sink get MessageDeliveryException.

我的转换方法：

public Event transform(LogFileTuple payload) {
    if (payload == null) {
      return null;
    }

    try {
      Event event = logParser.parse(payload);
      if (event != null) {
        return event;
      } else {
        logger.warn("LogParser result is null: {}.", payload);
        return null;
      }
    } catch (Exception e) {
      logger.warn("Cannot parse logFileTuple: {}.", payload, e);
      return null;
    }
  }

我的 xml 模块 2 配置是：

<int:channel id="input" datatype="com.company.threatanalyzer.core.commons.tuple.LogFileTuple"/>

<int:transformer input-channel="input" output-channel="output">
    <beans:bean class="com.company.threatanalyzer.xd.plugin.XdLogParser" />
</int:transformer>

<int:channel id="output"/>

我得到这个异常：

Caused by: org.springframework.messaging.MessageDeliveryException: Channel 'input' expected one of the following datataypes [class com.company.threatanalyzer.core.commons.tuple.LogFileTuple], but received [class com.company.threatanalyzer.core.commons.tuple.LogFileTuple]
    at org.springframework.integration.channel.AbstractMessageChannel.convertPayloadIfNecessary(AbstractMessageChannel.java:499) ~[spring-integration-core-4.2.5.RELEASE.jar:na]
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:426) ~[spring-integration-core-4.2.5.RELEASE.jar:na]
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:392) ~[spring-integration-core-4.2.5.RELEASE.jar:na]
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:115) ~[spring-messaging-4.2.4.RELEASE.jar:4.2.4.RELEASE]
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:45) ~[spring-messaging-4.2.4.RELEASE.jar:4.2.4.RELEASE]
    at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:105) ~[spring-messaging-4.2.4.RELEASE.jar:4.2.4.RELEASE]
    at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutput(AbstractMessageProducingHandler.java:231) ~[spring-integration-core-4.2.5.RELEASE.jar:na]
    at org.springframework.integration.handler.AbstractMessageProducingHandler.produceOutput(AbstractMessageProducingHandler.java:154) ~[spring-integration-core-4.2.5.RELEASE.jar:na]
    at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutputs(AbstractMessageProducingHandler.java:102) ~[spring-integration-core-4.2.5.RELEASE.jar:na]
    at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:105) ~[spring-integration-core-4.2.5.RELEASE.jar:na]
    at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:127) ~[spring-integration-core-4.2.5.RELEASE.jar:na]
    at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:116) ~[spring-integration-core-4.2.5.RELEASE.jar:na]
    at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:147) ~[spring-integration-core-4.2.5.RELEASE.jar:na]
    at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:120) ~[spring-integration-core-4.2.5.RELEASE.jar:na]
    at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:77) ~[spring-integration-core-4.2.5.RELEASE.jar:na]
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:442) ~[spring-integration-core-4.2.5.RELEASE.jar:na]
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:392) ~[spring-integration-core-4.2.5.RELEASE.jar:na]
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:115) ~[spring-messaging-4.2.4.RELEASE.jar:4.2.4.RELEASE]
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:45) ~[spring-messaging-4.2.4.RELEASE.jar:4.2.4.RELEASE]
    at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:105) ~[spring-messaging-4.2.4.RELEASE.jar:4.2.4.RELEASE]
    at org.springframework.integration.endpoint.MessageProducerSupport.sendMessage(MessageProducerSupport.java:105) ~[spring-integration-core-4.2.5.RELEASE.jar:na]
    at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter.access0(AmqpInboundChannelAdapter.java:45) ~[spring-integration-amqp-4.2.5.RELEASE.jar:na]
    at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter.onMessage(AmqpInboundChannelAdapter.java:93) ~[spring-integration-amqp-4.2.5.RELEASE.jar:na]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:757) ~[spring-rabbit-1.5.4.RELEASE.jar:na]
    ... 23 common frames omitted

Answer 1

这是一个 class 加载器问题；每个模块加载它自己的 LogFileTuple 版本。在 Java 中，classes 由加载它们的 classloader 限定范围，因此，在这种情况下，它是一个不同的 Class 实例。

您需要将包含 class 的 jar 提升到 /xd/lib，以便它由共享的 class 加载器加载。

Spring XD 自定义 class 处理器输入类型

Spring XD custom class type for processor input

java

spring

spring-integration

spring-xd