Блог

Я пытаюсь смоделировать систему, которая является генератором нагрузки для веб-сайта. На веб-сайте есть страницы, которые на данный момент являются строками.

 type Page = String

val pages: Vector[Page] = Stream.eval(Task.delay {
  new Random().alphanumeric.take(10).mkString
}).repeat.take(100).runLog.unsafeRun
 

На веб-сайте есть пользователи, которые моделируются как начинающие просматривать веб-сайт в случайное время:

 case class User(userName: String)

def newUser: Task[User] = Task.delay {
  val random = new Random
  val userName = random.alphanumeric.take(5).mkString
  val sleepTime = random.nextInt(10000)
  println(s"${new Date}: ${Thread.currentThread.getName} Generating user ${userName} after ${sleepTime}ms")
  Thread.sleep(sleepTime)
  User(userName)
}
 

Каждый пользователь переходит со страницы на страницу, и это моделируется как:

 case class Transition(from: Page, to: Page, thinkTime: Int)

def newTransitions(user: User): Stream[Task, Transition] = {
  val init = Transition(pages(0), pages(1), 100)
  Stream.iterateEval(init) { t => nextTransition(user)(t) }
}

def nextTransition(user: User)(curr: Transition): Task[Transition] = Task.delay {
  val random = new Random
  val from: Page = curr.to
  val to: Page = pages(random.nextInt(pages.size))
  val thinkTime: Int = random.nextInt(10000)
  Thread.sleep(thinkTime)
  println(s"${new Date}: ${Thread.currentThread.getName} ${user} transitioning from ${from} to ${to} after thinking for ${thinkTime}ms")
  Transition(from, to, thinkTime)
}
 

Я пытаюсь запустить это моделирование, чтобы 4 пользователя начинали просматривать этот веб-сайт через случайные промежутки времени, и каждый пользователь переходил на три страницы случайным образом:

 implicit val S = fs2.Strategy.fromFixedDaemonPool(8, threadName = "worker")
val users: Stream[Task, User] = Stream.eval(newUser).repeat
val transitions: Stream[Task, Stream[Task, Transition]] = users.map(newTransitions(_).take(3))

transitions.take(4).runLog.async.unsafeRun.map { _.runLog.async.unsafeRun }
 

Результат выглядит следующим образом:

 Sun Oct 16 12:58:04 PDT 2016: worker-1 Generating user 97qkE after 3700ms
Sun Oct 16 12:58:08 PDT 2016: worker-1 Generating user sMIJs after 2074ms
Sun Oct 16 12:58:10 PDT 2016: worker-1 Generating user CpWWf after 1334ms
Sun Oct 16 12:58:11 PDT 2016: worker-1 Generating user xlQmj after 8825ms
Sun Oct 16 12:58:28 PDT 2016: worker-7 User(97qkE) transitioning from sLBbTbXKhO to 7ROMneDkzl after thinking for 8040ms
Sun Oct 16 12:58:34 PDT 2016: worker-7 User(97qkE) transitioning from 7ROMneDkzl to AK66SuDENi after thinking for 6227ms
Sun Oct 16 12:58:41 PDT 2016: worker-6 User(sMIJs) transitioning from sLBbTbXKhO to U2Qu8Th1oH after thinking for 6599ms
Sun Oct 16 12:58:46 PDT 2016: worker-6 User(sMIJs) transitioning from U2Qu8Th1oH to 3bRBWYUeS0 after thinking for 4920ms
Sun Oct 16 12:58:52 PDT 2016: worker-4 User(CpWWf) transitioning from sLBbTbXKhO to ZVwRydhPe9 after thinking for 6395ms
Sun Oct 16 12:58:54 PDT 2016: worker-4 User(CpWWf) transitioning from ZVwRydhPe9 to yHtly5IrMC after thinking for 1963ms
Sun Oct 16 12:58:57 PDT 2016: worker-2 User(xlQmj) transitioning from sLBbTbXKhO to 6jn7TmxW0O after thinking for 2565ms
Sun Oct 16 12:59:05 PDT 2016: worker-2 User(xlQmj) transitioning from 6jn7TmxW0O to FFROVBvpHJ after thinking for 8535ms
 

т. е. валюты нет, все случайные переходы происходят в строгой последовательности.

Как я могу изменить эту программу таким образом, чтобы эти разные потоки (новые пользователи, начинающие просматривать веб-сайт, и переходы, выполняемые каждым пользователем) происходили одновременно?

Вот моя полная программа:

 import fs2._
import scala.util.Random
import java.util.Date

object Question {
  type Page = String

  val pages: Vector[Page] = Stream.eval(Task.delay {
    new Random().alphanumeric.take(10).mkString
  }).repeat.take(100).runLog.unsafeRun

  case class User(userName: String)

  case class Transition(from: Page, to: Page, thinkTime: Int)

  def newUser: Task[User] = Task.delay {
    val random = new Random
    val userName = random.alphanumeric.take(5).mkString
    val sleepTime = random.nextInt(10000)
    println(s"${new Date}: ${Thread.currentThread.getName} Generating user ${userName} after ${sleepTime}ms")
    Thread.sleep(sleepTime)
    User(userName)
  }

  def nextTransition(user: User)(curr: Transition): Task[Transition] = Task.delay {
    val random = new Random
    val from: Page = curr.to
    val to: Page = pages(random.nextInt(pages.size))
    val thinkTime: Int = random.nextInt(10000)
    Thread.sleep(thinkTime)
    println(s"${new Date}: ${Thread.currentThread.getName} ${user} transitioning from ${from} to ${to} after thinking for ${thinkTime}ms")
    Transition(from, to, thinkTime)
  }

  def newTransitions(user: User): Stream[Task, Transition] = {
    val init = Transition(pages(0), pages(1), 100)
    Stream.iterateEval(init) { t => nextTransition(user)(t) }
  }

  def main(args: Array[String]) {
    implicit val S = fs2.Strategy.fromFixedDaemonPool(8, threadName = "worker")
    val users: Stream[Task, User] = Stream.eval(newUser).repeat
    val transitions: Stream[Task, Stream[Task, Transition]] = users.map(newTransitions(_).take(3))

    transitions.take(4).runLog.async.unsafeRun.map { _.runLog.async.unsafeRun }
  }
}
 

Во-первых, на вашем object Question , похоже, есть опечатка. Не забудьте переключить:

 Thread.sleep(sleepTime)
println(s"${new Date}: ${Thread.currentThread.getName} Generating user ${userName} after ${sleepTime}ms")
 

Далее, вы хотели бы использовать Task { ... } вместо Task.delay , поскольку вы хотели бы запустить его в рабочем потоке.

Наконец, последняя строка в вашей основной функции должна быть примерно такой:

 fs2.concurrent.join(maxOpen = 3)(transitions).run.unsafeRun
 

Кроме того, я предполагаю, что Thread.sleep это просто заполнители — вы не хотели бы иметь их в реальном коде. И println они хороши для отладки, но лучше всего перенести ведение журнала в виде отдельных потоковых конструкций после завершения.