我从Akka Streams开始,到目前为止一切进展顺利.但是,我遇到了一个我不知道如何处理的用例.该场景是一个流,其中ActorPublisher作为源消耗来自Kafka的消息,订阅者作为更新Cassandra表的接收器.

Kafka~>一些映射操作〜>卡桑德拉

关键是我每次成功处理消息并插入Cassandra时都要向Kafka明确确认,以便在灾难发生且服务失败时重新读取消息,即至少某种消息一旦交货行为.我怎么能用Akka Streams来解决这个问题？是受支持的场景吗？

确实,我总是可以使用自动提交行为配置Kafka使用者,但我宁愿控制我如何阅读消息.

更新

关于这个主题,我们目前正在评估Reactive Kafka,他们已经在kafka中包含0.8版本的手动提交(这些人的称赞).此功能将允许我们实现我们需要的alod行为.

最佳答案 我认为您可以使用PushPullStage自定义流处理,如
akka streams documentation中所述