如何让kafka在spring xd运行时使用http流数据？

Question

是否可以让 kafka 源模块在 spring xd 运行时作为处理器模块工作？有代码示例吗？

我正在尝试实现这样的目标：http (xd source) | kafka源码（xd处理器）| kafka 消费者（xd 接收器）

我正在尝试这样做，因为我有来自 http 的流数据，我想使用 kafka 消息总线进行管理。

我的流定义是这样的：

stream create kafkaSourceTest --definition "http --outputType=application/json | kafka --zkconnect=localhost:2181 --topic=kafkaTestTopic | log " --deploy

将kafka源模块实现的开箱即用的实现放到spring xd的处理器模块中会导致如下错误：

2015-05-12 11:18:52,914 1.1.1.RELEASE ERROR pool-13-thread-4 http.NettyHttpInboundChannelAdapter - Error sending message

org.springframework.messaging.MessageDeliveryException: Dispatcher 没有频道 'admin:default,admin,singlenode,hsqldbServer:9393.kafkaSourceTest.0' 的订阅者。嵌套异常是 org.springframework.integration.MessageDispatchingException：调度程序没有订阅者 在 org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:81) 在 org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:277) 在 org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:239) 在 org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:115) 在 org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:45)

Answer 1

我正在尝试这样做，因为我有来自 http 的流数据，我想使用 kafka 消息总线进行管理。

如果你使用kafka作为messagebus（设置transport后），那么像“http | log”这样的流会让http消息流过kafka messagebus。在这种情况下，Kafka 代理中的主题将由 XD 内部定义。

是否可以让kafka源模块在spring xd运行时作为处理器模块工作？

不，源模块不能充当处理器模块。如果您希望消息流经 Kafka 中的特定主题，那么您可以拥有一个流，该流具有一个 kafka 接收器模块，该模块从 http 源接收数据，另一个流配置具有 same 主题的 kafka 源模块。

而且，这可以通过这种方式实现：

stream create KafkaSink --definition "http --outputType=application/json | kafka --brokerList= --topic=kafkaTestTopic" --deploy

stream create KafkaSource --definition "kafka --zkconnect=localhost:2181 --topic=kafkaTestTopic | log" --deploy