Ваше первое сообщение — Открывая Акку

1. Клиентское приложение должно вызывать указанный выше URL не более чем из одного потока — запрещено одновременно получать случайные числа, используя несколько HTTP-соединений.
2. Мы должны загружать случайные числа в пакетах, а не по одному в каждом запросе. Запрос выше принимает num=20номера за один звонок.
3. Нас предупреждают о задержке, ответ может прийти даже через одну минуту
4. Клиент должен периодически проверять квоту случайных чисел (услуга бесплатна только до заданного количества случайных бит в день). Все эти требования делают интеграцию с random.orgнетривиальной. В этой серии, которую я только начал, мы будем постепенно улучшать наше приложение, шаг за шагом изучая новые функции Akka. Вскоре мы поймем, что довольно крутая кривая обучения окупится быстро, как только мы поймем основные концепции платформы. Итак, давайте код!

Сегодня мы попытаемся выполнить первые два требования: это не более одного соединения в любой данный момент времени и загрузка номеров в пакетах. Для этого шага нам не нужен Akka, достаточно простой синхронизации и буфера:

val buffer = new Queue[Int]
 
def nextRandom(): Int = {
  this.synchronized {
    if(buffer.isEmpty) {
      buffer ++= fetchRandomNumbers(50)
    }
    buffer.dequeue()
  }
}
 
def fetchRandomNumbers(count: Int) = {
  val url = new URL("https://www.random.org/integers/?num=" + count + "&min=0&max=65535&col=1&base=10&format=plain&rnd=new")
  val connection = url.openConnection()
  val stream = Source.fromInputStream(connection.getInputStream)
  val randomNumbers = stream.getLines().map(_.toInt).toList
  stream.close()
  randomNumbers
}

Этот код работает и эквивалентен
synchronizedключевому слову в Java. Способ
nextRandom()работы должен быть очевиден: если буфер пуст, заполните его 50 случайными числами, выбранными с сервера. В конце возьмите и верните первое значение из буфера. Этот код имеет несколько недостатков, начиная с
synchronizedблока на первом месте. Скорее дорогая синхронизация для каждого вызова кажется излишним. И мы даже не в кластере, где нам пришлось бы поддерживать одно активное соединение на весь кластер, а не только с одной JVM!

Начнем с реализации одного актера. Актер в основном класс, расширяющий
Actorчерты и реализующий
receiveметод. Этот метод отвечает за получение и обработку одного сообщения. Давайте повторим то, что мы уже сказали: каждый субъект может обрабатывать не более одного сообщения за раз, поэтому
receiveметод никогда не вызывается
одновременно. Если субъект уже обрабатывает какое-либо сообщение, остальные сообщения хранятся в очереди, выделенной для каждого субъекта. Благодаря этому строгому правилу мы можем избежать
любой синхронизации внутри актера, которая всегда поточно-ориентирована.

case object RandomRequest
 
class RandomOrgBuffer extends Actor {
 
  val buffer = new Queue[Int]
 
  def receive = {
    case RandomRequest =>
      if(buffer.isEmpty) {
        buffer ++= fetchRandomNumbers(50)
      }
      println(buffer.dequeue())
  }
 
}

fetchRandomNumbers()метод остается прежним. Однопоточный доступ
random.orgбыл достигнут бесплатно, поскольку субъект может обрабатывать только одно сообщение за раз. Говоря о сообщениях, в данном случае
RandomRequestэто наше сообщение — пустой объект, не передающий никакой информации, кроме его типа. В Akka сообщения почти всегда реализуются с использованием case-классов или других неизменяемых типов. Таким образом, если мы хотим поддерживать выборку произвольного числа случайных чисел, мы должны включить это как часть сообщения:

case class RandomRequest(howMany: Int)
 
class RandomOrgBuffer extends Actor with ActorLogging {
 
  val buffer = new Queue[Int]
 
  def receive = {
    case RandomRequest(howMany) =>
      if(buffer.isEmpty) {
        buffer ++= fetchRandomNumbers(50)
      }
      for(_ <- 1 to (howMany min 50)) {
        println(buffer.dequeue())
      }
  }

Теперь мы должны попытаться отправить сообщение нашему новому актеру. Очевидно, что мы не можем просто вызвать
receiveметод, передающий сообщение в качестве аргумента. Сначала мы должны запустить платформу Akka и попросить ссылку на актера. Эта ссылка позже используется для отправки сообщения с использованием слегка нелогичного первого
!метода, начиная с дней Эрланга:

object Bootstrap extends App {
  val system = ActorSystem("RandomOrgSystem")
  val randomOrgBuffer = system.actorOf(Props[RandomOrgBuffer], "buffer")
 
  randomOrgBuffer ! RandomRequest(10)  //sending a message
 
  system.shutdown()
}