Thread Pool Executors

gistfile1.txt

package pellucid.dataloader.datascope

import com.google.common.util.concurrent.ThreadFactoryBuilder
import java.util.concurrent._
import java.util.concurrent.atomic.AtomicInteger
import org.slf4j.LoggerFactory
import scala.concurrent.duration._
import scala.concurrent.{Await, Future}
import scala.util.Random

object TestExecutors extends App {

  val log = LoggerFactory.getLogger(this.getClass)

  object BlockCallerPolicy extends RejectedExecutionHandler {
    def rejectedExecution(r: Runnable, executor: ThreadPoolExecutor) {
      try executor.getQueue.put(r)
      catch {
        case e: InterruptedException =>
          throw new RejectedExecutionException("Unexpected InterruptedException", e)
      }
    }
  }

  def namedExecutor(nThreads: Int, name: String, daemon: Boolean = true) = {
    val threadFactory = new ThreadFactoryBuilder().
      setNameFormat(s"$name-%d").
      setDaemon(daemon).
      build()

    /*new ThreadPoolExecutor(2, nThreads,
      1, TimeUnit.SECONDS,
      new SynchronousQueue[Runnable],
      threadFactory, BlockCallerPolicy)
*/
    new ThreadPoolExecutor(0, nThreads,
      1, TimeUnit.SECONDS,
      new LinkedBlockingQueue[Runnable],
      threadFactory)
  }

  implicit val ec = scala.concurrent.ExecutionContext.fromExecutor(namedExecutor(10, "test-pool"))

  val cnt = new AtomicInteger(0)
  val random = new Random()

  val futures = for (i <- 1 to 100) yield {
    scala.concurrent.future {
      Thread.sleep(100 + random.nextInt(300))
      val taskN = cnt.incrementAndGet()
      log.info(s"Run task $taskN")
      taskN
    }
  }

  val allDone = Future.sequence(futures)

  Await.result(allDone, 1.minute)

  println(allDone)
}