はじめに

PlayFrameworkを使って勉強シリーズ第三弾です。今回はRedisにアクセスしてみます。

ScalaからRedisへアクセスするライブラリはいろいろあるのですが、今回は扱いやすさからscredisを使用しました。

なお、Scala、PlayFramework共に勉強中なため、至らぬ点がありましたら平にご容赦願いたてまつりまつる。

実装

前提

今回使用した各種バージョンは以下の通りです。

名前	バージョン
Scala	2.12.4
PlayFramework	2.6.7
Redis	3.0.6
scredis	2.1.1

Redis

Redisは、今回、非clusterのMaster-Read Replica構成を想定します。

build.sbt

build.sbtに、必要なライブラリを追記します。

build.sbt

// scredis
resolvers += Resolver.bintrayRepo("jastice", "maven")

libraryDependencies ++= Seq(
  "com.github.scredis" %% "scredis" % "2.1.1"
)

Redisアクセス

Redisへの接続方法ですが、今回は、Factoryパターンを使ってscredis.Redisオブジェクトを生成するような形にします。

Factory自体は、Moduleクラスでシングルトンを登録しておき、Injectionして使うような形にします。
シングルトンは、以下のように定義します。

services/RedisConnector.scala

package services

import javax.inject.{Inject, Singleton}

import play.api.Configuration
import scredis.Redis

import scala.util.Random

// PrimaryかRead Replicaへのアクセスかを識別するためのtype
sealed trait RedisType
object RedisTypes {
  class WriterType extends RedisType
  class ReaderType extends RedisType

  implicit case object RedisWriterObject extends WriterType
  implicit case object RedisReaderObject extends ReaderType
}

// 公開されたtrait
trait RedisConnector {
  def redis[T <: RedisType](implicit redisType: T): Redis
}

// 実装クラス
@Singleton
class DefaultRedisConnector @Inject()(config: Configuration)
  extends RedisConnector
{
  // configの読み込み
  private val defaultPort = config.getOptional[Int]("redis.port").getOrElse(6379)
  private val writerHost = config.getOptional[String]("redis.writer.host").get
  private val (writerHostName, writerPort) = splitHostAndPort(writerHost)
  private val readerHosts =
    config.getOptional[String]("redis.reader.host").getOrElse(writerHost).split(",").map(splitHostAndPort)

  // read replicaが複数あるときのrandomizer
  private lazy val randomizer = Random

  // ReaderType, WriterTypeの読み込み
  import RedisTypes._

  override def redis[T <: RedisType](implicit redisType: T): Redis = {
    redisType match {
      case _: WriterType => writerClient
      case _: ReaderType => selectReaderClient
    }
  }

  private def splitHostAndPort(host: String): (String, Int) =
    host.split(":") match {
      case Array(hostname, port) => (hostname, port.toInt)
      case Array(hostname)       => (hostname, defaultPort)
    }

  private def writerClient: Redis = Redis(writerHostName, writerPort)

  private def selectReaderClient: Redis = {
    readerHosts.length match {
      // To-Do: case 0
      case 1 => readerClient(0)
      case n if n > 1 => readerClient(randomizer.nextInt(n))
    }
  }

  private def readerClient(n: Int): Redis = {
    val (hostName, port) = readerHosts(n)
    Redis(hostName, port)
  }
}

Module

Moduleクラスに、上記のクラスを登録します。

Module.scala

import services._

class Module extends AbstractModule {

  override def configure(): Unit = {
    bind(classOf[RedisConnector]).to(classOf[DefaultRedisConnector])
  }
}

applicaton.conf

接続先のRedisホストは、application.confに記載することになります。
なお、Read Replicaについては、複数指定が可能になっています。(複数指定されている場合、ランダムに選択する)。

application.conf

redis {
  port: 6379
  writer {
    host: "redis.hogehoge.ng.0001.apne1.cache.amazonaws.com"
  }
  reader {
    host: "redis-002.hogehoge.0001.apne1.cache.amazonaws.com,redis-003.hogehoge.0001.apne1.cache.amazonaws.com"
  }
}

Controllerからの使い方

controllers/HogeController.scala

package controllers

import java.nio.charset.StandardCharsets
import javax.inject._

import play.api.mvc._
import scredis.serialization.{Reader, Writer}
import services.RedisConnector

import scala.concurrent.duration.Duration
import scala.concurrent.{ExecutionContext, Future}

@Singleton
class HogeController @Inject()(cc: ControllerComponents,
                               rc: RedisConnector,
                               implicit val ec: ExecutionContext) extends AbstractController(cc)
{
  import services.RedisTypes._

  // Redisに格納されている(String, Int)をGetするためのReader。なお、文字列に","は含まれていないものとする(手抜き)
  implicit val stringIntReader: Reader[(String, Int)] = { (bytes: Array[Byte]) =>
    new String(bytes, StandardCharsets.UTF_8).split(",") match {
      case Array(strValue, intValue) => (strValue, intValue) 
    }
  }

  // Redisに(String, Int)をPutするためのWriter
  implicit val stringIntWriter: Writer[(String, Int)] = { (value: (String, Int)) =>
    (value._1.toString + "," + value._2.toString).getBytes(StandardChasets.UTF_8)

  // ReadReplicaからの読み込み(stringIntReaderが暗黙的に参照される)
  private def getValue: Future[Option[(String, Int)]] =
    rc.redis[ReaderType].get[(String, Int)]("hogehoge_key")

  // Masterへの書き込み(stringIntWriterが暗黙的に参照される)
  private def setValue(textValue: String, intValue: Int): Future[Unit] =
    rc.redis[WriterType].set("hogehoge_key", (textValue, intValue))
}

DNSキャッシュ(おまけ)

AWS環境では、短期間にDNSクエリが集中すると、途中からエラーを返すようになります (PHPからElastiCacheを参照してて痛い目にあいました)。

(参考) https://dev.classmethod.jp/cloud/aws/amazon-dns-threshold-exceeded-action/

しかし、scredisのベースとなっているakka.ioおよびjavaのInetAddressでは、ローカルにキャッシュするのでこの問題は回避できると思われます。ただ、インターバル時間がデフォルトの30秒では長すぎるといった場合には、これを短くする必要があります。

独自のApplicationLoaderを定義し、そこで定数を書き換えてやることにします。下の例ではDNSクエリ成功時・失敗時ともに2秒に変更しています。

MyApplicationLoader.scala

import java.security.Security

import play.api.ApplicationLoader
import play.api.inject.guice.{GuiceApplicationBuilder, GuiceApplicationLoader}
class MyApplicationLoader extends GuiceApplicationLoader {
  override protected def builder(context: ApplicationLoader.Context): GuiceApplicationBuilder = {
    Security.setProperty("networkaddress.cache.ttl", "2")
    Security.setProperty("networkaddress.cache.negative.ttl", "2")
    super.builder(context)
  }
}

application.conf

play.application {
  loader=MyApplicationLoader
}