使用 Kafka 的 Spring Boot 云流

Question

我正在尝试使用带有 Kafka 的 Springboot cloud Stream 设置一个项目。我设法构建了一个简单的示例，其中侦听器从一个主题获取消息，并在处理后将输出发送到另一个主题。

我的监听器和频道是这样配置的：

@Component
public class FileEventListener {
    private FileEventProcessorService fileEventProcessorService;

    @Autowired
    public FileEventListener(FileEventProcessorService fileEventProcessorService) {
        this.fileEventProcessorService = fileEventProcessorService;
    }

    @StreamListener(target = FileEventStreams.INPUT)
    public void handleLine(@Payload(required = false) String jsonData) {
        this.fileEventProcessorService.process(jsonData);
    }
}

public interface FileEventStreams {
    String INPUT = "file_events";
    String OUTPUT = "raw_lines";

    @Input(INPUT)
    SubscribableChannel inboundFileEventChannel();

    @Output(OUTPUT)
    MessageChannel outboundRawLinesChannel();
}

这个例子的问题是，当服务启动时，它不会检查主题中已经存在的消息，它只处理启动后发送的那些消息。我对 Springboot 流和 kafka 很陌生，但就我所读到的而言，这种行为可能与我使用SubscribableChannel 的事实相对应。例如，我尝试使用QueueChannel 来查看它是如何工作的，但我发现了以下异常：

Error creating bean with name ... nested exception is java.lang.IllegalStateException: No factory found for binding target type: org.springframework.integration.channel.QueueChannel among registered factories: channelFactory,messageSourceFactory

所以，我的问题是：

1. 如果我想在应用程序启动后处理主题中存在的所有消息（并且消息也仅由一个消费者处理），那么我走在正确的道路上吗？
2. 即使 QueueChannel 不是实现 1 中所述行为的正确选择。）我必须向我的项目添加什么才能使用这种类型的通道？

谢谢！

Answer 1

1. 添加spring.cloud.stream.bindings.file_events.group=foo

   • 默认情况下，匿名组仅从主题末尾开始使用，与组的绑定从开头开始使用。

2. 您不能使用PollableChannel 进行绑定，它必须是SubscribableChannel。