2

我正在尝试使用Typesafe Redis Play plugin连接到 Redis 以进行发布订阅。

我有以下测试场景，由每秒生成消息的参与者组成：

  // Initialization happens in Application.scala,
  private lazy val fakeStreamActor = Akka.system.actorOf(Props[FakeStreamActor])

  val actorPut = Akka.system.scheduler.schedule(
    Duration(1000, MILLISECONDS),
    Duration(1000, MILLISECONDS),
    fakeStreamActor,
    Put("This is a sample message"))

演员来源：

class FakeStreamActor extends Actor {
  implicit val timeout = Timeout(1, SECONDS)

  val CHANNEL = "channel1"
  val plugin = Play.application.plugin(classOf[RedisPlugin]).get
  val listener = new MyListener()

  val pool = plugin.sedisPool

  pool.withJedisClient{ client =>
    client.subscribe(listener, CHANNEL)
  }

  def receive = {

    case Put(msg: String) => {
      //send data to Redis
      Logger.info("Push %s".format(msg))
      pool.withJedisClient { client =>
        client.publish(CHANNEL, msg)
      }

    }
  }
}

/** Messages */
case class Put(msg: String)

还有一个订阅监听器：

case class MyListener() extends JedisPubSub {
  def onMessage(channel: String, message: String): Unit = {
    Logger.info("onMessage[%s, %s]".format(channel, message))
  }

  def onSubscribe(channel: String, subscribedChannels: Int): Unit = {
    Logger.info("onSubscribe[%s, %d]".format(channel, subscribedChannels))
  }

  def onUnsubscribe(channel: String, subscribedChannels: Int): Unit = {
    Logger.info("onUnsubscribe[%s, %d]".format(channel, subscribedChannels))
  }

  def onPSubscribe(pattern: String, subscribedChannels: Int): Unit = {
    Logger.info("onPSubscribe[%s, %d]".format(pattern, subscribedChannels))
  }

  def onPUnsubscribe(pattern: String, subscribedChannels: Int): Unit = {
    Logger.info("onPUnsubscribe[%s, %d]".format(pattern, subscribedChannels))
  }

  def onPMessage(pattern: String, channel: String, message: String): Unit = {
    Logger.info("onPMessage[%s, %s, %s]".format(pattern, channel, message))
  }
}

现在，理想情况下，我应该订阅定义的频道并在日志中查看监听器如何每秒处理接收到的消息。但这不会发生，因为订阅的行为会锁定线程。

我的问题是：

有什么方法可以利用 Play 异步特性进行非阻塞订阅？

4

1 回答 1

4

是的。这就是我在 Global.scala 中的做法：

Akka.future { 
  val j = new RedisPlugin(app).jedisPool.getResource
  j.subscribe(PubSub, "*")
}

我在实例化插件时遇到了麻烦，但是您实际上将 withJedisClient 位放在了 future 块中。

感谢您向我展示如何在 scala 中实例化插件！

于 2013-02-20T19:02:24.573 回答