如何从Akka中提取实时的FileIO状态？
Akka是一个基于Actor模型的并发编程框架，它提供了强大的工具和库来构建分布式、可扩展和高可用的应用程序。在Akka中，FileIO是一个用于处理文件输入输出的模块，它提供了一种简单而高效的方式来处理大量数据。
要从Akka中提取实时的FileIO状态，可以通过以下步骤实现：
- 导入必要的Akka库和模块：
- 导入必要的Akka库和模块：
- 创建一个ActorSystem和ActorMaterializer：
- 创建一个ActorSystem和ActorMaterializer：
- 使用FileIO模块读取文件并处理数据：
- 使用FileIO模块读取文件并处理数据：
在上述代码中，我们首先使用FileIO.fromPath方法创建一个文件源，然后使用map操作对每个数据块进行处理。在处理过程中，可以提取实时的FileIO状态并进行相应的操作。最后，使用FileIO.toPath方法将处理后的数据块写入文件或其他目标。
需要注意的是，上述代码只是一个示例，具体的实现方式取决于具体的需求和业务逻辑。此外，Akka还提供了许多其他功能和模块，如流处理、集群管理等，可以根据实际情况进行选择和使用。
推荐的腾讯云相关产品：腾讯云云服务器（CVM）、腾讯云对象存储（COS）、腾讯云云数据库MySQL等。你可以通过访问腾讯云官方网站获取更多关于这些产品的详细信息和介绍。
相关·内容
这是我之前的问题：
我已经成功地使用FileIO.fromPath(Paths.get(file.toURI()在Akka流上发送了文件，它工作得很好。但是，我想在发送文件之前对其进行压缩和加密。我已经创建了方法，打开FileInputStream，通过压缩流路由它，然后通过加密流路由它，现在我想使用Akka流将它定向到套接字中。
文件-> FileInputStream -> CompressedInputStream -> EncryptedInputStream -?> Spring/Akka流
关键是，我可以实时地压缩/加密文件，同时逐块读取它(我没有在磁盘上创
浏览 2提问于2017-02-25得票数 1
回答已采纳
在Scala中从FileIO.fromPath中获取akka Bytestring的惯用方法是什么？
浏览 0提问于2016-09-21得票数 1
回答已采纳
2回答
使用下面的代码，我使用Akka streams读取并打印文件的内容： package playground
import java.nio.file.Paths
import akka.actor.ActorSystem
import akka.stream.scaladsl.{FileIO, Framing, Sink, Source}
import akka.util.ByteString
import akka.stream.ActorMaterializer
object Greeter extends App {
implicit val system = ActorSy
浏览 47提问于2019-09-24得票数 0
回答已采纳
服务器接收包含一个文件和一个JSON对象的多部分表单。从请求中提取文件、元数据和其他JSON对象的最佳方式是什么？现在我有以下代码：
pathPrefix(""){
post {
extractRequestContext {
requestCtx => {
implicit val materializer = requestCtx.materializer
implicit val executionContext = requestCtx.executionContext
fileUpl
浏览 0提问于2018-09-04得票数 0
我有一个非常简单的流来处理一个文件--只是一个FileIO.FromFile源、一个通过流和一个Last接收器：
var source = FileIO.FromFile(new FileInfo(fileName));
var flow = Flow.FromFunction<ByteString, ByteString>(x =>
{
Log.Info(x.Count.ToString());
return x;
});
var sink = Sink.Last<ByteString>()
浏览 0提问于2018-05-17得票数 0
1回答
保持组合的意义？
、、
我试图在akka中的Keep组合下创建以下示例：
import java.nio.file.Paths
import akka.NotUsed
import akka.actor.ActorSystem
import akka.stream.{ActorMaterializer, IOResult}
import akka.stream.scaladsl.{FileIO, Flow, Keep, Sink, Source}
import akka.util.ByteString
import scala.concurrent.Future
import scala.util.{Failur
浏览 0提问于2019-04-12得票数 4
回答已采纳
1回答
对于下面的代码，我需要客户端将文件名发送到服务器，然后服务器可以向客户端回复指定文件的内容。如果我不指定文件名(只是服务器部分的硬编码)，但是客户机如何告诉服务器它需要指定的文件，它就可以工作。
Server.scala
package com.tst
import java.nio.file.Paths
import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.Tcp.{IncomingConnection, ServerBinding}
import akk
浏览 0提问于2018-02-02得票数 0
回答已采纳
2回答
我正在尝试运行一个简单的示例，但没有成功。我可以创建一个源代码，运行流，然后创建一个文件，但是ByteString不会被写入到文件中。然而，如果我试图打印流输出到控制台，我可以这样做。我是不是漏掉了什么？
import akka.stream._
import akka.stream.scaladsl._
import akka.{ NotUsed, Done}
import akka.actor.ActorSystem
import akka.util.ByteString
import scala.concurrent._
import scala.concurrent.duration
浏览 3提问于2016-09-19得票数 2
1回答
我试图从akka流中读取一个带有Source的avro文件。
在akka中，源读取像FileIO.FromPath(文件)这样的数据，它将根据(\n)字符读取和分隔行，至于它是如何工作的呢？
流动：
object AvroFlow {
def apply(jobDate: String): Flow[GenericRecord, GenericRecord, NotUsed] = {
Flow[GenericRecord].map { rec => rec.put("date", "20190812") rec}
浏览 0提问于2019-06-18得票数 2
回答已采纳
1回答
基本上，我希望允许用户从服务器下载csv文件。假设CSV文件已经存在于服务器上。API端点通过GET /export公开。如何将文件从Akka HTTP服务器流到客户端？这就是我目前所拥有的..。
服务：
def export(): Future[IOResult] = {
FileIO.fromPath(Paths.get("file.csv"))
.to(Sink.ignore)
.run()
}
路由：
pathPrefix("export") {
pathEndOrSingleSlash {
get {
浏览 1提问于2016-12-14得票数 5
回答已采纳
我正在处理将当前项目切换到Akka流时出现的性能问题。
在简化问题之后，似乎Akka流传递的消息比我预期的要少得多。
这里我有两段非常简单的代码，每次都只是向磁盘上的一个文件写入10个字节。
第一个使用两个线程和一个连接它们的ArrayBlockingQueue：
val bw = Files.newBufferedWriter(Paths.get("test.txt"))
val target = "0123456789".toCharArray
val abq = new ArrayBlockingQueue[Array[Char]](10000)
new
浏览 0提问于2018-05-15得票数 4
2回答
我尝试使用PlayWS向邮件枪发送一个文件和一些数据，但是我收到了这个错误：
Cannot write an instance of akka.stream.scaladsl.Source[play.api.mvc.MultipartFormData.Part[akka.stream.scaladsl.Source[akka.util.ByteString, Any]], Any] to HTTP response. Try to define a Writeable[akka.stream.scaladsl.Source[play.api.mvc.MultipartFormData.Part
浏览 0提问于2016-07-06得票数 4
1回答
下面的代码段读取CSV文件，并将某些内容打印到控制台：
def readUsingAkkaStreams = {
import java.io.File
import akka.stream.scaladsl._
import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import java.security.MessageDigest
implicit val system = ActorSystem("Sys")
implicit va
浏览 0提问于2018-01-26得票数 1
回答已采纳
1回答
我是阿克卡流的学习者。我通过TCP制作了一个简单的文件服务器和一个简单的文件接收器。当服务器和接收方都在同一台主机上时，两者都能正常工作。但是，当接收方位于与服务器主机不同的主机时，接收方无法访问服务器。所以，，我的问题是，为什么不同主机的接收方不能访问服务器。
例如，如果我在192.168.1.20运行服务器，192.168.1.20的接收器可以访问服务器，但是192.168.11.22的接收方不能访问服务器。
我确认港口是开放的。我创建了一个使用相同端口的简单系统，可以使用telnet命令访问端口。
源代码在这里
Server.scala (文件服务器)
import java.io.Fi
浏览 1提问于2016-03-21得票数 0
回答已采纳
5回答
阿克卡是以什么方式实时的？
、、、、
在几个地方，有一种说法说阿克卡是“实时的”。例如：
不幸的是，我无法找到更深层次的解释，阿克卡是“实时的”。这就是问题所在：
，akka是以哪种方式实时的？
我假设akka并不是一个真正的实时计算系统，从以下定义来说，不是吗？：
浏览 10提问于2013-01-24得票数 7
1回答
我试着考虑使用akka流来控制消息。
对我来说，最简单的方法是从文件中读取流行，处理它并提供给目标服务。
接下来的代码将介绍idea (source ~> lines ~> sink)：
import akka.NotUsed
import akka.actor.SupervisorStrategy.Restart
import akka.actor.{
Actor,
ActorLogging,
ActorRef,
ActorSystem,
OneForOneStrategy,
Props,
SupervisorStrategy
}
import akk
浏览 3提问于2019-09-23得票数 0
如何将tcp.bind()连接的Sink切换到TCP？例如，假设我有：
Sink<ByteString, CompletionStage<Void>> tmpFileSink = FileIO
.toPath(destination)
.mapMaterializedValue(this::noticeCompletion);
tcp
.bind(host, port)
.to(Sink.foreach(incomingConnection -> wire(incoming
浏览 16提问于2021-05-29得票数 0
3回答
我是Akka流的新手，我想知道如何实现某种中间流验证。示例：
FileIO
.fromPath(file)
.via(Framing.delimiter(...)
.map(_.utf8String)
.map(_.split("\t", -1))
.validate(arr => arr.length == 10) // or similar
...
我假设此场景非常常见，因此必须有一个预先定义的功能来动态验证流。然而，我没能找到任何关于它的东西。我是不是站在了错误的轨道上，在Akka流中，验证是不应该这样做的？
在我的特定场景中，我将逐行处理文
浏览 3提问于2017-06-11得票数 2
回答已采纳
1回答
我已经构建了一个定义流的akka图。我的目标是重新格式化我未来的响应并将其保存到一个文件中。这个流程可以概括如下：
val g = RunnableGraph.fromGraph(GraphDSL.create() { implicit builder: GraphDSL.Builder[NotUsed] =>
import GraphDSL.Implicits._
val balancer = builder.add(Balance[(HttpRequest, String)](6, waitForAllDownstreams = false))
v
浏览 0提问于2017-02-06得票数 10
回答已采纳
1回答
我在使用alpakka的XmlParsing Flow时遇到问题 val files: List[String] = ... // file paths locally on disk
// simple source emitting the contents of 2 XML files
val documentSource = FileIO.fromPath(Paths.get(files.head))
.concat(FileIO.fromPath(Paths.get(files(1))))
val contentFlow: Flow[ParseEvent, CustomCo
浏览 16提问于2018-12-21得票数 0
回答已采纳
2回答
读取文件时无输出
、、
我使用Akka FileIO逐行读取文件，如下所示：
def main(args: Array[String]): Unit = {
implicit val system = ActorSystem("MySys")
implicit val materializer = ActorMaterializer()
FileIO.fromPath(Paths.get("/foo/1.csv")).via(Framing.delimiter(ByteString("\n"), 256, true).map(_.utf8St
浏览 0提问于2016-11-04得票数 0
我正在尝试使用akka-stream-alpakka连接器从S3下载pdf文件。我有s3路径，并尝试在alpakka s3Client上使用包装器方法下载pdf。
def getSource(s3Path: String): Source[ByteString, NotUsed] = {
val (source, _) = s3Client.download(s3Bucket, s3Path)
source
}
从我的主代码中，我调用上面的方法，获得pdf源代码，并能够下载pdf。但是，我现在需要将pdf作为一个字节数组放在内存中，这样我就可以执行一些其他处理，而不是下载文
浏览 2提问于2020-02-06得票数 0
2回答
我有一个关于使用Akka Streams和Akka集群的问题。我正在尝试做一个使用Akka Streams和Akka Cluster的分布式字数统计版本。
我想构建一个Akka Streams客户端，它将文本文件作为流I/O读取，并将单词流发送到远程集群。这是客户端的代码：
final Path file = Paths.get("example.txt");
final Source<ByteString, CompletionStage<IOResult>> read = FileIO.fromPath(file);
final Source<
浏览 5提问于2018-11-10得票数 2
1回答
我使用sttp和akka后端从服务器加载一个文件。下列任何一种方法都会导致加载1Gb文件的大量内存占用：
import com.softwaremill.sttp._
val file: File = new File(...)
sttp.response(asStream[Source[ByteString, Any]])
.mapResponse { src =>
src.runWith(FileIO.toPath(file.toPath, options, 0))
}
sttp.response(asFile(file, false))
1Gb文件顺序加载的Visua
浏览 2提问于2019-11-28得票数 3
回答已采纳
了解所有AKKA演员都是从/user守护演员派生出来的。为了处理异常，我想定义MyAppParent执行元，并将我的所有执行元置于此执行元之下。有没有可能用AKKA做到这一点。在实时中，我需要直接发送消息到MyAppParentActor下的子Actor邮箱，而不是通过父Actor。任何关于如何做到这一点的指针，如果它可以与AKKA。
谢谢，卡熊
浏览 2提问于2015-06-25得票数 1
我是Akka HTTP的新手，我想知道如何从Akka HTTP请求中提取cookie。如果可能的话，有人可以发布一个如何使用Scala来完成这个任务的例子吗？
浏览 1提问于2017-12-01得票数 1
回答已采纳
我试图张贴pdf文件到外部服务与多部分/表单数据请求。我已经用示例Java脚本客户端完成了这个任务，这样外部服务就可以正常工作了。在scala中使用以下代码的Hovewer :错误请求：
import akka.stream.scaladsl.FileIO
import akka.stream.scaladsl.Source
import play.api.libs.ws.WSClient
import play.api.mvc.MultipartFormData._
val pathToFile = "./sampleCV.pdf"
val fileName = "
浏览 4提问于2020-08-25得票数 3
回答已采纳
在火花流中，我们设置了几乎实时的微批处理的批处理间隔。在Flink (DataStream)或Storm中，流是实时的，所以我想没有这样的批处理间隔的概念。
在卡夫卡，消费者在拉，我想象星火使用批参数从卡夫卡经纪人提取消息，那么Flink和Storm是如何做到的呢？我设想Flink和Storm在一个快速循环中拉出Kafka消息形成实时流源，如果是这样的话，如果我将火花批处理间隔设置为100 as、50 as甚至更小，那么我们是否与Flink或Storm之间有显著差异？
同时，在Spark中，如果流数据很大，批处理间隔太小，我们可能会遇到一种情况，即需要处理的数据很多，因此会发生OutOfMem
浏览 5提问于2016-10-24得票数 7
回答已采纳
我正在尝试为流分析设计一个流架构。要求：
RT和NRT流数据输入
实现某些财务分析的流处理器
RT和NRT分析输出流
流处理过程中的参考数据请求
我正在探索Kafka和Kafka流，用于流处理和RT/NRT实时消息传递。我的问题是:在流处理过程中，我需要对外部系统(信息提供者、MongoDB等)执行一些查询。根据外部系统特性，这些查询可以是同步查询和异步req响应。
我读过，解释了如何在处理过程中加入KStream和KTable，这很有趣，但在这个场景中，KTable并不依赖于来自KStream的输入参数，它只是表的流表示。
需要查询KStream消息的外部系统，传递一
浏览 5提问于2017-02-06得票数 9
回答已采纳
1回答
Akka集群文档没有指定如何自动扩展集群-添加或删除节点。这应该如何在Akka生态系统中实现呢？例如，在两种情况下-当我们使用VM作为节点，或者Kubernetes中的容器组成集群时。 如果特定节点的负载较低，Akka Cluster是否会将Actor从该节点移动到其他节点，并自动关闭未充分利用的节点？ 如果Akka集群没有弹性能力，并且依赖于Kubernetes (如here所述)，那么如果Kubernetes决定删除未充分利用的Akka集群节点，如何确保保留在未充分利用节点中的Actor将“优雅地”移动到其他集群节点？
浏览 67提问于2020-08-20得票数 0
1回答
我正在使用akka-stream来解析文件。在我的文件中，我有标题行在第一位，没有其他地方。如何在我的流中先跳过这个头文件？我不想每次都检查行是否为表头。有没有办法做到这一点？ FileIO.fromPath(path).
via(delimiter(ByteString("\n"), Int.MaxValue)). // split into lines
map(_.decodeString("UTF-8")). // convert ByteStrings to Strings
filter(!_.startsWith("#")
浏览 9提问于2019-05-03得票数 0
我正在尝试创建一个示例akka流，该流接受CSV文件，将其更改为XML (使用已经存在的具有toXml函数的对象)，然后将其发布到端点。我创建的代码如下所示：
val poolClientFlow =
Http().cachedHostConnectionPool[Thing]("localhost",5000)
val file = new File("./example.csv")
val uploadPipeline =
FileIO.fromFile(file)
.via(Framing.delimiter(ByteString(&#
浏览 1提问于2017-05-18得票数 1
回答已采纳
Java DSL我尝试使用客户端api实时地将字节流发送到服务器(作为主体)，但不知道请求开始之前的长度。
我无法从akka-http文档中找出如何做到这一点，有人尝试过吗？
浏览 0提问于2016-10-21得票数 0
1回答
我正在考虑将大型代码库迁移到Scala，并利用Akka Actor模型。我几乎立即遇到了以下概念问题：
在现有的代码库中，使用离散事件模拟(DES)来测试业务逻辑，比如具有确定性结果的SimPy。当实时程序运行时，DES需要几分钟.实时系统可以是异步的，不需要遵循测试设置的确切顺序。我想利用相同的Akka代码对测试设置和实时设置。这能在Scala + Akka中实现吗？
我已经尝试了一个中心消息队列执行者的想法--但是我觉得这不是正确的方法。
浏览 1提问于2016-05-05得票数 2
回答已采纳
我正在用Akka制作一个文件传输系统。我已经看了一段时间的文件了。当前进度状态为Actor2收到Actor1发送的文件并写入Actor2的本地系统(Actor1 = sender，Actor2 = receiver)。
但是我找不到一种方法来知道我在写的时候实时收到了多少字节。
我对其进行了测试，结果表明，使用runWith应用程序接口，可以在本地写入文件。通过runForeach接口，实时传递了多少字节。但是，如果同时创建这两个文件，则无法写入文件。
这是我的简单来源。请给我一些建议。
public static Behavior<Command> create() {
浏览 26提问于2020-05-11得票数 0
回答已采纳
我正在尝试构建使用S3密钥的代码，然后从S3下载这些文件，然后将数据保存到磁盘上的一个具有密钥名的文件中(流程更远的过程所需的)，并且输出返回键/文件名。到目前为止，我拥有的是；
val x: Sink[String, Future[IOResult]] =
Flow[String].flatMapConcat(key => S3.download("somebucket", key)).
withAttributes(S3Attributes.settings(useVersion1Api)).
collect{ ca
浏览 1提问于2019-11-21得票数 0
回答已采纳
我对Akka比较陌生，正在尝试学习基础知识。我的用例是不断地从JMS队列中读取消息，并将每条消息输出到一个新文件。我的基本设置是：
Source<String, NotUsed> jmsSource =
JmsSource
.textSource(JmsSourceSettings
.create(connectionFactory)
.withQueue("myQueue")
.withBufferSize(10));
Sink<ByteString, CompletionStage<IOResult>>
浏览 3提问于2017-07-19得票数 7
2回答
-基于这个问题。我用Akka流读取CSV。现在我需要逐行执行过程，但我也需要知道标题的名称是什么。有什么选择吗？
UPD。澄清一点。
FileIO.fromPath(Paths.get("a.csv))
.via(Framing.delimiter(ByteString("\n"), 256, true).map(_.utf8String))
.runForeach(println /* header + current line for each line*/)
浏览 5提问于2017-03-09得票数 2
回答已采纳
在过去的几天里，我一直试图找到使用Akka和HTTP将HTTP资源下载到文件中的最佳方法。
最初，我是从开始的，看起来如下所示：
def downloadViaFutures(uri: Uri, file: File): Future[Long] = {
val request = Get(uri)
val responseFuture = Http().singleRequest(request)
responseFuture.flatMap { response =>
val source = response.entity.dataBytes
sourc
浏览 5提问于2016-01-20得票数 12
回答已采纳
2回答
我现在正忙着用VB.NET编写一个挂人游戏。作为一个单词列表，我有一个包含1520个单词的文本文件，每个文本文件都被一个新的行分隔开来。
我所能想到的得到一个随机单词的最好方法是使用随机化()函数。然后从随机生成的行#中获取单词。
刚才才发现，这个方法：
Using parser As New Microsoft.VisualBasic.FileIO.TextFieldParser_
("filepath")
parser.TextFieldType = FileIO.FieldType.Delimited
不允许我用新行作为分隔符..。
考虑到所有的单词都有不
浏览 1提问于2012-05-14得票数 0
回答已采纳
2回答
我是Akka的新手，并开发了一个示例Akka WebSocket服务器，该服务器使用BroadcastHub (基于Akka 的示例)将文件内容流到客户端。
假设客户端消耗的速度与服务器一样快，我如何度量吞吐量(消息/秒)？
// file source
val fileSource = FileIO.fromPath(Paths.get(path)
// Akka file source
val theFileSource = fileSource
.toMat(BroadcastHub.sink)(Keep.right)
.run
//Akka kafka file source
浏览 2提问于2018-03-09得票数 4
回答已采纳
1回答
我正在尝试提取我用Coq编写的一些文件系统代码。我想把我的FileIO Monad换成Haskell的IO Monad。我的代码如下所示：
Variable FileIO : Type -> Type.
Variable sync : FileIO unit.
Extract Inlined Constant sync => "synchronize".
Extract Inlined Constant FileIO => "IO".
Recursive Extraction append.
替换sync没有问题，但是当我试图用FileIO
浏览 5提问于2015-01-26得票数 3
回答已采纳
我的Akka演员获得了一些资源，这些资源应该在作业完成后释放，如下面的代码片段所示。
问题是，在出现异常时，如何确保资源被释放。
我想到的第一个想法是将获得的资源分配给var，并在def postStop()钩子中释放它们作为一种保障。
由于这是一个常见的要求，我想知道这是否有最佳实践？
override def receive: Receive = {
case FILEIO(path) => {
val ios = fs.create(new Path(path))
// ios.read/write ..
// exception c
浏览 2提问于2019-11-21得票数 0
回答已采纳
我很难弄清楚如何从非参与者代码中知道我的参与者什么时候结束了。
尝试使用akka角色进行并发执行，这看起来真的很好，但我需要知道何时所有这些都完成了。
我已经添加了一个supervisor actor，这个actor衍生了actor来做工作，从supervisor/parent actor中，我可以观察到孩子的终止消息。
我需要帮助来弄清楚我的监督演员是否完成了，从akka演员系统之外的常规java代码。
这是可能的，请提供指示。
谢谢Hari
浏览 0提问于2018-01-06得票数 1
1回答
我正在尝试使用.NET Core1.0的单元测试上下文中的HOCON配置块来配置我的AKKA.NET运行时环境。显然，配置不是从App.config文件中提取的。
有人能解释一下在单元测试环境(xunit)中运行时，AKKA.NET是如何获取HOCON配置的吗？
如果上面的问题没有提供足够的信息，下面是一些详细信息：
我正在努力避免众所周知的警告。
NewtonSoftJsonSerializer has been detected as a default serializer. It will be obsoleted in Akka.NET starting from version 1
浏览 3提问于2016-08-26得票数 0
1回答
我使用Akka FileIO (在scala中)创建一个文件解析器，用于从输入文件中读取每一行并应用一个简单的接收器。除文件中以EOF结尾的最后一行外，每一行都用换行符('\n')分隔。
如何同时处理换行符和eof分隔符，这样就可以可靠地读取最后一行，而不必依赖于最终的'/n‘字符？
var rowNum = 0
val simpleMsgSink: Sink[String, Future[Done]] =
Sink.foreach {
case row: String => {
println(s&
浏览 3提问于2017-05-15得票数 1
回答已采纳
我希望在使用波束的内置FileIO.matchAll()变换时保留上一阶段的数据。 我有一个来自前一个转换的结构，看起来像这样： public class MyStruct {
public String filePattern;
// Down below are some other metadata fields
// I'd like to use in later transforms
public String tableId
....
} 问题是，为了使用FileIO.matchAll()，我需要截断结构中的其他字段，只提取fil
浏览 21提问于2021-08-09得票数 0
我在Play Framework2.8上下载了一个简单的alpakka s3文件，代码如下：
在S3服务中：
def download(link: String): Source[Option[(Source[ByteString, NotUsed], ObjectMetadata)], NotUsed] = {
S3.download(TraktrainBucket.DOWNLOAD_BUCKET, link)
}
在控制器中：
val source = s3Service.download(link).map(s => s.map(_._1))
浏览 2提问于2020-07-28得票数 2
1回答
我有一个愚蠢的问题，但不知道原因：
import akka.{Done, NotUsed}
import akka.actor.Status.Success
import akka.actor.{ActorRef, ActorSystem}
import akka.stream.scaladsl.{Flow, RunnableGraph, Sink, Source}
import akka.stream.{ActorMaterializer, OverflowStrategy}
import scala.concurrent.Future
object Generic {
def m
浏览 1提问于2017-07-19得票数 0
回答已采纳
我使用Source.queue对HttpRequests进行排队，并在客户端限制它，以便从远程服务器下载文件。我知道Source.queue不是线程安全的，我们需要使用MergeHub来使它成为线程安全。下面是使用Source.queue和cachedHostConnectionPool的代码片段。
import java.io.File
import akka.actor.Actor
import akka.event.Logging
import akka.http.scaladsl.Http
import akka.http.scaladsl.client.RequestBuildin
浏览 0提问于2017-09-22得票数 3
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
