java - SpringXD 和 Spring Integration : Read from kafka topic every X minutes, 然后发送到另一个主题

Question

我正在尝试实现一个解决方案来创建由 kafka 源、桥接模块和 kafka 接收器组成的 SpringXD 流。

所以我有这样的东西:

<channel id="pollable">
    <queue />
</channel>

<bridge input-channel="pollable" output-channel="executorChannel">
    <poller max-messages-per-poll="5" fixed-rate="5000" />
</bridge>

我的问题是我想以某种方式避免轮询器。基本上是因为我想避免当这些消息在队列中时将消息保留在内存中。我更愿意每 X 分钟从 kafka 读取一次，然后从队列中取出 Y 条消息并将这些消息发送到下一个主题。

看起来我无法摆脱队列，但我的问题是:还有其他选择吗？我不喜欢将内容保留在内存中，但我也不想使用此选项:http://docs.spring.io/spring-integration/reference/html/system-management-chapter.html#message-store

Answer 1

将数据保存在内存中并不是一个好主意。

您可以根据需要stop()和start() channel 适配器(KafkaMessageDrivenChannelAdapter)；重新启动时它将从中断处继续。

但是，kafka 源使用非常旧版本的 spring-integration-kafka (1.3.x)。

如果您创建自定义源以使用 spring-integration-kafka 2.1.0(使用 kafka 0.10.1.x 客户端)，则可以设置 kafka 属性 max.poll.records限制获取的记录数量。