WebSockets を使用してデータをクライアントにストリーミングする Play Framework 2.2.0-scala を使用して構築している例があります。私が抱えている問題は、何らかの理由で、親アクタの子の 1 つが適切にシャットダウンされていないことです。すべてのログは、停止中であり、シャットダウンしたことを示していますが、データを公開することで、実際にはダウンしていないことがわかります。最初にコントローラーアクションを使用したコードを次に示します。
def scores(teamIds: String) = WebSocket.async[JsValue] { request =>
val teamIdsArr:Array[String] = teamIds.split(",").distinct.map { el =>
s"nfl-streaming-scores-${el}"
}
val scoresStream = Akka.system.actorOf(Props(new ScoresStream(teamIdsArr)))
ScoresStream.join(scoresStream)
}
そのため、クライアントが接続するたびに参加ScoresStreamし、WebSocket.async が必要とするそれぞれの Iteratee,Enumerator を返します。実際の ScoresStream オブジェクトは次のようになります。
object ScoresStream {
implicit val timeout = Timeout(5 seconds)
def join(scoresStream:ActorRef):scala.concurrent.Future[(Iteratee[JsValue,_],Enumerator[JsValue])] = {
(scoresStream ? BeginStreaming).map {
case Connected(enumerator) =>
val iteratee = Iteratee.foreach[JsValue] { _ =>
Logger.info("Ignore iteratee input.")
}.map { _ =>
Logger.info("Client quitting - killing Actor.")
scoresStream ! UnsubscribeAll
scoresStream ! PoisonPill
}
(iteratee,enumerator)
}
ここでの考え方はScoresStream、クライアントが切断されたときにメインのアクタ を kill することです。を使用してそれを行いscoresStream ! PoisonPillます。
ScoresStream次に、メッセージの公開/書き込みのために Redis に接続するラッパーであるインスタンスを作成PubしSubます。アクター コードは次のとおりです。
class ScoresStream(teamIds: Array[String]) extends Actor with CreatePubSub with akka.actor.ActorLogging {
val (scoresEnumerator, scoresChannel) = Concurrent.broadcast[JsValue]
case class Message(kind: String, user: String, message: String)
implicit val messageReads = Json.reads[Message]
implicit val messageWrites = Json.writes[Message]
val sub = context.child("sub") match {
case None => createSub(scoresChannel)
case Some(c) => c
}
val pub = context.child("pub") match {
case None => createPub(teamIds)
case Some(c) => c
}
def receive = {
case BeginStreaming => {
log.info("hitting join...")
sub ! RegisterCallback
sub ! SubscribeChannel(teamIds)
sender ! Connected(scoresEnumerator)
}
case UnsubscribeAll => {
sub ! UnsubscribeChannel(teamIds)
}
}
}
trait CreatePubSub { self:Actor =>
def createSub(pChannel: Concurrent.Channel[JsValue]) = context.actorOf(Props(new Sub(pChannel)), "sub")
def createPub(teamIds: Array[String]) = context.actorOf(Props(new Pub(teamIds)), "pub")
}
最後に、実際のサブ アクター コードを次に示します (Pub正常にシャットダウンしているため、ここでは関係ないようです)。
class Sub(pChannel: Concurrent.Channel[JsValue]) extends Actor with CreatePublisherSubscriber with ActorLogging {
val s = context.child("subscriber") match {
case None => createSubscriber
case Some(c) => c
}
def callback(pubsub: PubSubMessage) = pubsub match {
case E(exception) => println("Fatal error caused consumer dead. Please init new consumer reconnecting to master or connect to backup")
case S(channel, no) => println("subscribed to " + channel + " and count = " + no)
case U(channel, no) => println("unsubscribed from " + channel + " and count = " + no)
case M(channel, msg) =>
msg match {
// exit will unsubscribe from all channels and stop subscription service
case "exit" =>
println("unsubscribe all ..")
pChannel.end
r.unsubscribe
// message "+x" will subscribe to channel x
case x if x startsWith "+" =>
val s: Seq[Char] = x
s match {
case Seq('+', rest @ _*) => r.subscribe(rest.toString){ m => }
}
// message "-x" will unsubscribe from channel x
case x if x startsWith "-" =>
val s: Seq[Char] = x
s match {
case Seq('-', rest @ _*) => r.unsubscribe(rest.toString)
pChannel.end
}
case x =>
try {
log.info("Just got a message: " + x)
pChannel.push(Json.parse(x))
}
catch {
case ex: com.fasterxml.jackson.core.JsonParseException => {
log.info("Malformed JSON sent.")
}
}
}
}
def receive = {
case RegisterCallback => {
log.info("Creating a subscriber and registering callback")
s ! Register(callback)
}
case SubscribeChannel(teamIds) => {
teamIds.foreach { x => log.info("subscribing to channel " + x + " ") }
//sub ! Subscribe(Array("scores-5","scores-6"))
s ! Subscribe(teamIds)
}
case UnsubscribeChannel(teamIds) => {
teamIds.foreach { x => log.info("unsubscribing from channel " + x + " ") }
s ! Unsubscribe(teamIds)
}
case true => println("Subscriber successfully received message.")
case false => println("Something went wrong.")
}
}
trait CreatePublisherSubscriber { self:Actor =>
def r = new RedisClient("localhost", 6379)
def createSubscriber = context.actorOf(Props(new Subscriber(r)), "subscriber")
def createPublisher = context.actorOf(Props(new Publisher(r)), "publisher")
}
クライアントが接続すると、起動メッセージは正常に見えます。
[DEBUG] [10/20/2013 00:35:53.618] [application-akka.actor.default-dispatcher-12] [akka://application/user] now supervising Actor[akka://application/user/$c#-54456921]
[DEBUG] [10/20/2013 00:35:53.619] [application-akka.actor.default-dispatcher-12] [akka://application/user/$c] started (com.example.stream.models.ScoresStream@131a9310)
[DEBUG] [10/20/2013 00:35:53.620] [application-akka.actor.default-dispatcher-12] [akka://application/user/$c] now supervising Actor[akka://application/user/$c/sub#1376180991]
[DEBUG] [10/20/2013 00:35:53.621] [application-akka.actor.default-dispatcher-17] [akka://application/user/$c/pub/publisher] started (com.redis.Publisher@3b34c0a6)
[DEBUG] [10/20/2013 00:35:53.622] [application-akka.actor.default-dispatcher-17] [akka://application/user/$c/sub/subscriber] started (com.redis.Subscriber@453f0a8)
Subscriber successfully received message.
Subscriber successfully received message.
[DEBUG] [10/20/2013 00:35:53.699] [application-akka.actor.default-dispatcher-19] [akka://application/user/$c/sub] started (com.example.stream.models.Sub@6165ab39)
[DEBUG] [10/20/2013 00:35:53.699] [application-akka.actor.default-dispatcher-19] [akka://application/user/$c/sub] now supervising Actor[akka://application/user/$c/sub/subscriber#-1562348862]
subscribed to nfl-streaming-scores-5 and count = 1
[DEBUG] [10/20/2013 00:35:53.699] [application-akka.actor.default-dispatcher-12] [akka://application/user/$c] now supervising Actor[akka://application/user/$c/pub#-707418539]
[INFO] [10/20/2013 00:35:53.700] [application-akka.actor.default-dispatcher-12] [akka://application/user/$c] hitting join...
[INFO] [10/20/2013 00:35:53.700] [application-akka.actor.default-dispatcher-23] [akka://application/user/$c/sub] Creating a subscriber and registering callback
[INFO] [10/20/2013 00:35:53.700] [application-akka.actor.default-dispatcher-23] [akka://application/user/$c/sub] subscribing to channel nfl-streaming-scores-5
[DEBUG] [10/20/2013 00:35:53.700] [application-akka.actor.default-dispatcher-18] [akka://application/user/$c/pub] started (com.example.stream.models.Pub@48007a17)
[DEBUG] [10/20/2013 00:35:53.703] [application-akka.actor.default-dispatcher-18] [akka://application/user/$c/pub] now supervising Actor[akka://application/user/$c/pub/publisher#1509054514]
そして、切断は正常に見えます:
[info] application - Client quitting - killing Actor.
unsubscribed from nfl-streaming-scores-5 and count = 0
[DEBUG] [10/20/2013 00:37:51.696] [application-akka.actor.default-dispatcher-17] [akka://application/user/$c] received AutoReceiveMessage Envelope(PoisonPill,Actor[akka://application/deadLetters])
[INFO] [10/20/2013 00:37:51.696] [application-akka.actor.default-dispatcher-25] [akka://application/user/$c/sub] unsubscribing from channel nfl-streaming-scores-5
[DEBUG] [10/20/2013 00:37:51.696] [application-akka.actor.default-dispatcher-17] [akka://application/user/$c] stopping
[DEBUG] [10/20/2013 00:37:51.697] [application-akka.actor.default-dispatcher-25] [akka://application/user/$c/sub] stopping
[DEBUG] [10/20/2013 00:37:51.697] [application-akka.actor.default-dispatcher-25] [akka://application/user/$c/pub/publisher] stopped
[DEBUG] [10/20/2013 00:37:51.697] [application-akka.actor.default-dispatcher-17] [akka://application/user/$c/sub/subscriber] stopped
[DEBUG] [10/20/2013 00:37:51.697] [application-akka.actor.default-dispatcher-17] [akka://application/user/$c/sub] stopped
[INFO] [10/20/2013 00:37:51.697] [application-akka.actor.default-dispatcher-17] [akka://application/user/$c/sub] Message [java.lang.Boolean] from Actor[akka://application/user/$c/sub/subscriber#-1562348862] to Actor[akka://application/user/$c/sub#1376180991] was not delivered. [2] dead letters encountered. This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.
[DEBUG] [10/20/2013 00:37:51.699] [application-akka.actor.default-dispatcher-26] [akka://application/user/$c/pub] stopping
[DEBUG] [10/20/2013 00:37:51.699] [application-akka.actor.default-dispatcher-26] [akka://application/user/$c/pub] stopped
[DEBUG] [10/20/2013 00:37:51.699] [application-akka.actor.default-dispatcher-17] [akka://application/user/$c] stopped
ここで問題があります。クライアントが切断された後、現在シャットダウンしているアクタがサブスクライブされているというメッセージを送信します。
redis-cli publish "nfl-streaming-scores-5" "{\"test\":\"message\"}"
ここでは表示されていますが、表示されるべきではない場合、このアクタは技術的には停止している必要があります。$a/$b のラベルが付いた、メッセージを受信する前に存在した他のアクタも同様です。他のクライアントが接続されていないことを確認できます。
[INFO] [10/20/2013 00:38:33.097] [Thread-7] [akka://application/user/$c/sub] Just got a message: {"test":"message"}
また、奇妙な指標は、アドレス名が再利用されないことです。切断/接続すると、次のような名前が生成される傾向が見られます。
akka://application/user/$c
akka://application/user/$d
akka://application/user/$e
古い参照が再利用されることはありません。
ここでの私の仮定は、Redis への接続がきれいに閉じられていないということです。アクターが停止したにもかかわらずまだ存在しているというログが表示される理由は説明されていませんが、netstatすべてのアクターがおそらく停止した後でも、実行後に redis への接続が確立されていることは確かです。アプリケーションの実行を完全に停止すると、それらの接続がクリアされます。それはあたかもサブスクライブ解除が静かに失敗し、それがアクタと接続を維持しているようです。これは、最終的にシステムがファイル記述子を使い果たしたり、メモリ リークが発生したりするため、複数の理由で非常に悪いことです。ここで私が間違っていることは明らかですか？