Removing the AtomicReference from Dispatcher and restructured the code a bit

akka-actor/src/main/scala/akka/dispatch/BalancingDispatcher.scala

@@ -87,7 +87,7 @@ class BalancingDispatcher(
     @tailrec def scheduleOne(i: Iterator[ActorCell] = team.iterator): Unit =
       if (messageQueue.hasMessages
         && i.hasNext
-        && (executorService.get().executor match {
+        && (executorService match {
           case lm: LoadMetrics ⇒ lm.atFullThrottle == false
           case other           ⇒ true
         })

akka-actor/src/main/scala/akka/dispatch/Dispatcher.scala

@@ -33,11 +33,15 @@ class Dispatcher(
   val shutdownTimeout: Duration)
   extends MessageDispatcher(_prerequisites) {
 
-  protected val executorServiceFactory: ExecutorServiceFactory =
+  private class LazyExecutorServiceDelegate(factory: ExecutorServiceFactory) extends ExecutorServiceDelegate {
-    executorServiceFactoryProvider.createExecutorServiceFactory(id, prerequisites.threadFactory)
+    lazy val executor: ExecutorService = factory.createExecutorService
+    def copy(): LazyExecutorServiceDelegate = new LazyExecutorServiceDelegate(factory)
+  }
 
-  protected val executorService = new AtomicReference[ExecutorServiceDelegate](
+  @volatile private var executorServiceDelegate: LazyExecutorServiceDelegate =
-    new ExecutorServiceDelegate { lazy val executor = executorServiceFactory.createExecutorService })
+    new LazyExecutorServiceDelegate(executorServiceFactoryProvider.createExecutorServiceFactory(id, prerequisites.threadFactory))
+
+  protected final def executorService: ExecutorService = executorServiceDelegate
 
   /**
    * INTERNAL USE ONLY
@@ -62,11 +66,11 @@ class Dispatcher(
    */
   protected[akka] def executeTask(invocation: TaskInvocation) {
     try {
-      executorService.get() execute invocation
+      executorService execute invocation
     } catch {
       case e: RejectedExecutionException ⇒
         try {
-          executorService.get() execute invocation
+          executorService execute invocation
         } catch {
           case e2: RejectedExecutionException ⇒
             prerequisites.eventStream.publish(Error(e, getClass.getName, getClass, "executeTask was rejected twice!"))
@@ -83,10 +87,15 @@ class Dispatcher(
   /**
    * INTERNAL USE ONLY
    */
-  protected[akka] def shutdown: Unit =
+  protected[akka] def shutdown: Unit = {
-    Option(executorService.getAndSet(new ExecutorServiceDelegate {
+    val newDelegate = executorServiceDelegate.copy() // Doesn't matter which one we copy
-      lazy val executor = executorServiceFactory.createExecutorService
+    val es = synchronized { // FIXME getAndSet using ARFU or Unsafe
-    })) foreach { _.shutdown() }
+      val service = executorServiceDelegate
+      executorServiceDelegate = newDelegate // just a quick getAndSet
+      service
+    }
+    es.shutdown()
+  }
 
   /**
    * Returns if it was registered
@@ -97,12 +106,12 @@ class Dispatcher(
     if (mbox.canBeScheduledForExecution(hasMessageHint, hasSystemMessageHint)) { //This needs to be here to ensure thread safety and no races
       if (mbox.setAsScheduled()) {
         try {
-          executorService.get() execute mbox
+          executorService execute mbox
           true
         } catch {
           case e: RejectedExecutionException ⇒
             try {
-              executorService.get() execute mbox
+              executorService execute mbox
               true
             } catch { //Retry once
               case e: RejectedExecutionException ⇒