ApacheKafka—编写用于在scala的listbuffer中存储数据的不可变代码

ss2ws0br 于 2021-06-07 发布在 Kafka

关注(0)|答案(2)|浏览(338)

我有下面的代码，我正在使用可变列表缓冲区来存储从kafka消费者收到的文件，然后当列表大小达到15时，我将它们插入cassandra。但是他们有没有办法用不变列表做同样的事情呢。

val filesList = ListBuffer[SystemTextFile]()
  storeservSparkService.configFilesTopicInBatch.subscribe.atLeastOnce(Flow[SystemTextFile].mapAsync(4) { file: SystemTextFile =>
    filesList += file
    if (filesList.size == 15) {
      storeServSystemRepository.config.insertFileInBatch(filesList.toList)
      filesList.clear()
    }
    Future(Done)
  })

scala apache-kafka lagom immutability

来源：https://stackoverflow.com/questions/43450993/write-immutable-code-for-storing-data-in-listbuffer-in-scala

2条答案

按热度按时间

osh3o9ms1#

沿着这些路线？

Flow[SystemTextFile].grouped(15).mapAsync(4){ files =>
  storeServSystemRepository.config.insertFileInBatch(files)
}

赞(0）回复(0）举报 2021-06-07

ymzxtsji2#

你试过使用向量吗？

val filesList = Vector[SystemTextFile]()
      storeservSparkService.configFilesTopicInBatch.subscribe.
          atLeastOnce(Flow[SystemTextFile].mapAsync(4) { file: SystemTextFile =>
       filesList = filesList :+ file
       if (filesList.length == 15) {
            storeServSystemRepository.config.insertFileInBatch(filesList.toList)
       }
       Future(Done)
     })

赞(0）回复(0）举报 2021-06-07

我来回答

ApacheKafka—编写用于在scala的listbuffer中存储数据的不可变代码

2条答案

相关问题

热门标签

最新问答