flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActor.java
/**
	 * Handle asynchronous {@link Runnable}. This method simply executes the given {@link Runnable}
	 * in the context of the actor thread.
	 *
	 * @param runAsync Run async message
	 */
Handle asynchronous  Runnable. This method simply executes the given  Runnable
in the context of the actor thread.

@param runAsync Run async message
Handle asynchronous Runnable . This method simply executes the given Runnable in the context of the actor thread . @param runAsync Run async message
private void handleRunAsync(RunAsync runAsync) {
		if (runAsync.getRunnable() == null) {
			log.warn("Received a {} message with an empty runnable field. This indicates " +
				"that this message has been serialized prior to sending the message. The " +
				"{} is only supported with local communication.",
				runAsync.getClass().getName(),
				runAsync.getClass().getName());
		}
		else {
			final long timeToRun = runAsync.getTimeNanos();
			final long delayNanos;
			if (timeToRun == 0 || (delayNanos = timeToRun - System.nanoTime()) <= 0) {
				try {
					runAsync.getRunnable().run();
				} catch (Throwable t) {
					log.error("Caught exception while executing runnable in main thread.", t);
					ExceptionUtils.rethrowIfFatalErrorOrOOM(t);
				}
			}
			else {
				FiniteDuration delay = new FiniteDuration(delayNanos, TimeUnit.NANOSECONDS);
				RunAsync message = new RunAsync(runAsync.getRunnable(), timeToRun);
				getContext().system().scheduler().scheduleOnce(delay, getSelf(), message,
						getContext().dispatcher(), ActorRef.noSender());
			}
		}
	}
private void handleRunAsync ( RunAsync runAsync ) { if ( runAsync . getRunnable ( ) == null ) { log . warn ( "Received a {} message with an empty runnable field. This indicates " + "that this message has been serialized prior to sending the message. The " + "{} is only supported with local communication." , runAsync . getClass ( ) . getName ( ) , runAsync . getClass ( ) . getName ( ) ) ; } else { final long timeToRun = runAsync . getTimeNanos ( ) ; final long delayNanos ; if ( timeToRun == 0 || ( delayNanos = timeToRun - System . nanoTime ( ) ) <= 0 ) { try { runAsync . getRunnable ( ) . run ( ) ; } catch ( Throwable t ) { log . error ( "Caught exception while executing runnable in main thread." , t ) ; ExceptionUtils . rethrowIfFatalErrorOrOOM ( t ) ; } } else { FiniteDuration delay = new FiniteDuration ( delayNanos , TimeUnit . NANOSECONDS ) ; RunAsync message = new RunAsync ( runAsync . getRunnable ( ) , timeToRun ) ; getContext ( ) . system ( ) . scheduler ( ) . scheduleOnce ( delay , getSelf ( ) , message , getContext ( ) . dispatcher ( ) , ActorRef . noSender ( ) ) ; } } }
Handle asynchronous {@link Runnable}.
Handle asynchronous { @link Runnable } .
// run immediately // schedule for later. send a new message after the delay, which will then be immediately executed