val stringSourceinFuture=Source.actorRef[String](100,OverflowStrategy.fail) // 缓存最大为100,超出的话,将以失败告终
val hahaStrSource=stringSourceinFuture.filter(str=>str.startsWith("haha")) //source数据流中把不是以"haha"开头的字符串过滤掉
val actor=hahaStrSource.to(Sink.foreach(println)).run()
actor!"asdsadasd"
actor!"hahaasd"
actor!Success("ok")// 数据流成功完成并关闭
”how to create a Source that can receive elements later via a method call?”在akka-http中经常遇见Source[T,N]的地方就是对文件上传和下载的功能的编码(文件IO)中,完成file=>Source[ByteString,_]的转化,或者Source(List(1,2,3,4,5))这种hello-world级别的玩具代码中,这些代码中在定义Source时,就已经确定流中数据是什么了。那么如何先定义流,而后给流传递数据呢?答案就是Source.actorRef。郑重说明:Source.actorRef没有背压策略(背压简单说就是生产者的生成速率大于消费者处理速率,导致数据积压)。
Sink.actorRef
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
classMyActorextendsActor{
overridedefreceive: Receive = {
case"FIN"=>
println("完成了哇!!!")
context.stop(self)
case str:String =>
println("msgStr:"+str)
}
}
......
val actor=system.actorOf(Props[MyActor],"myActor")
val sendToActor=Sink.actorRef(actor,onCompleteMessage = "FIN")
val hahaStringSource=Source.actorRef[String](100,OverflowStrategy.dropHead).filter(str=>str.startsWith("haha"))
val actorReceive=hahaStringSource.to(sendToActor).run()