import akka.dispatch.Recover; //导入依赖的package包/类
public void receiveHandler(Object message) {
  log.debug("Master client received: {}",message);
  Timeout timeout = new Timeout(120, TimeUnit.SECONDS);
  Future<Object> future = ask(masterProxy, message, timeout);
  final ExecutionContext ec = getContext().system().dispatcher();

  Future<Object> res = future.map(new Mapper<Object, Object>() {
    @Override
    public Object apply(Object msg) {
        return new Ok(msg);
    }
  }, ec).recover(new Recover<Object>() {
    @Override
    public Object recover(Throwable failure) throws Throwable {
      return new NotOk(null);
    }
  }, ec);

  pipe(res, ec).to(getSender());
}
 

import akka.dispatch.Recover; //导入依赖的package包/类
public Future<Object> queryStateFutureWithFailover(final int tries, final QueryState<K> queryState) {
	@SuppressWarnings("unchecked")
	Future<Object> result = queryStateFuture(queryState).recoverWith(new Recover<Future<Object>>() {
		@Override
		public Future<Object> recover(Throwable failure) throws Throwable {
			if (tries > 0) {
				LOG.debug("Query state failed with {}. Try to recover. #{} left.", failure, tries - 1);
				return queryStateFutureWithFailover(tries - 1, queryState);
			} else {
				return Futures.failed(failure);
			}
		}
	}, executor);

	return result;
}
 

import akka.dispatch.Recover; //导入依赖的package包/类
public void onReceive(Object message) {

    Timeout timeout = new Timeout(5, TimeUnit.SECONDS);
    Future<Object> f = ask(masterProxy, message, timeout);


    final ExecutionContext ec = getContext().system().dispatcher();

    Future<Object> res = f.map(new Mapper<Object, Object>() {
      @Override
      public Object apply(Object msg) {
        if (msg instanceof Master.Ack)
          return Ok.getInstance();
        else
          return super.apply(msg);
      }
    }, ec).recover(new Recover<Object>() {
      @Override
      public Object recover(Throwable failure) throws Throwable {
        return NotOk.getInstance();
      }
    }, ec);

    pipe(res, ec).to(getSender());
  }
 

import akka.dispatch.Recover; //导入依赖的package包/类
/**
 * Returns a future holding the serialized request result.
 *
 * <p>If the server does not serve a KvState instance with the given ID,
 * the Future will be failed with a {@link UnknownKvStateID}.
 *
 * <p>If the KvState instance does not hold any data for the given key
 * and namespace, the Future will be failed with a {@link UnknownKeyOrNamespace}.
 *
 * <p>All other failures are forwarded to the Future.
 *
 * @param jobId                     JobID of the job the queryable state
 *                                  belongs to
 * @param queryableStateName        Name under which the state is queryable
 * @param keyHashCode               Integer hash code of the key (result of
 *                                  a call to {@link Object#hashCode()}
 * @param serializedKeyAndNamespace Serialized key and namespace to query
 *                                  KvState instance with
 * @return Future holding the serialized result
 */
@SuppressWarnings("unchecked")
public Future<byte[]> getKvState(
		final JobID jobId,
		final String queryableStateName,
		final int keyHashCode,
		final byte[] serializedKeyAndNamespace) {

	return getKvState(jobId, queryableStateName, keyHashCode, serializedKeyAndNamespace, false)
			.recoverWith(new Recover<Future<byte[]>>() {
				@Override
				public Future<byte[]> recover(Throwable failure) throws Throwable {
					if (failure instanceof UnknownKvStateID ||
							failure instanceof UnknownKvStateKeyGroupLocation ||
							failure instanceof UnknownKvStateLocation ||
							failure instanceof ConnectException) {
						// These failures are likely to be caused by out-of-sync
						// KvStateLocation. Therefore we retry this query and
						// force look up the location.
						return getKvState(
								jobId,
								queryableStateName,
								keyHashCode,
								serializedKeyAndNamespace,
								true);
					} else {
						return Futures.failed(failure);
					}
				}
			}, executionContext);
}
 

import akka.dispatch.Recover; //导入依赖的package包/类
@Override
public <R> Future<R> exceptionallyAsync(final ApplyFunction<Throwable, ? extends R> exceptionallyFunction, Executor executor) {
	Preconditions.checkNotNull(scalaFuture);
	Preconditions.checkNotNull(exceptionallyFunction);
	Preconditions.checkNotNull(executor);

	scala.concurrent.Future<R> recoveredFuture = scalaFuture.recover(new Recover<R>() {
		@Override
		public R recover(Throwable failure) throws Throwable {
			return exceptionallyFunction.apply(failure);
		}
	}, createExecutionContext(executor));

	return new FlinkFuture<>(recoveredFuture);
}
 

import akka.dispatch.Recover; //导入依赖的package包/类
@SuppressWarnings("unchecked")
private static Future<byte[]> getKvStateWithRetries(
		final QueryableStateClient client,
		final JobID jobId,
		final String queryName,
		final int key,
		final byte[] serializedKey,
		final FiniteDuration retryDelay,
		final boolean failForUknownKeyOrNamespace) {

	return client.getKvState(jobId, queryName, key, serializedKey)
			.recoverWith(new Recover<Future<byte[]>>() {
				@Override
				public Future<byte[]> recover(Throwable failure) throws Throwable {
					if (failure instanceof AssertionError) {
						return Futures.failed(failure);
					} else if (failForUknownKeyOrNamespace &&
							(failure instanceof UnknownKeyOrNamespace)) {
						return Futures.failed(failure);
					} else {
						// At startup some failures are expected
						// due to races. Make sure that they don't
						// fail this test.
						return Patterns.after(
								retryDelay,
								TEST_ACTOR_SYSTEM.scheduler(),
								TEST_ACTOR_SYSTEM.dispatcher(),
								new Callable<Future<byte[]>>() {
									@Override
									public Future<byte[]> call() throws Exception {
										return getKvStateWithRetries(
												client,
												jobId,
												queryName,
												key,
												serializedKey,
												retryDelay,
												failForUknownKeyOrNamespace);
									}
								});
					}
				}
			}, TEST_ACTOR_SYSTEM.dispatcher());
}
 

import akka.dispatch.Recover; //导入依赖的package包/类
/**
 * Returns a future holding the {@link KvStateLocation} for the given job
 * and KvState registration name.
 *
 * <p>If there is currently no JobManager registered with the service, the
 * request is retried. The retry behaviour is specified by the
 * {@link LookupRetryStrategy} of the lookup service.
 *
 * @param jobId               JobID the KvState instance belongs to
 * @param registrationName    Name under which the KvState has been registered
 * @param lookupRetryStrategy Retry strategy to use for retries on UnknownJobManager failures.
 * @return Future holding the {@link KvStateLocation}
 */
@SuppressWarnings("unchecked")
private Future<KvStateLocation> getKvStateLookupInfo(
		final JobID jobId,
		final String registrationName,
		final LookupRetryStrategy lookupRetryStrategy) {

	return jobManagerFuture
			.flatMap(new Mapper<ActorGateway, Future<Object>>() {
				@Override
				public Future<Object> apply(ActorGateway jobManager) {
					// Lookup the KvStateLocation
					Object msg = new KvStateMessage.LookupKvStateLocation(jobId, registrationName);
					return jobManager.ask(msg, askTimeout);
				}
			}, actorSystem.dispatcher())
			.mapTo(ClassTag$.MODULE$.<KvStateLocation>apply(KvStateLocation.class))
			.recoverWith(new Recover<Future<KvStateLocation>>() {
				@Override
				public Future<KvStateLocation> recover(Throwable failure) throws Throwable {
					// If the Future fails with UnknownJobManager, retry
					// the request. Otherwise all Futures will be failed
					// during the start up phase, when the JobManager did
					// not notify this service yet or leadership is lost
					// intermittently.
					if (failure instanceof UnknownJobManager && lookupRetryStrategy.tryRetry()) {
						return Patterns.after(
								lookupRetryStrategy.getRetryDelay(),
								actorSystem.scheduler(),
								actorSystem.dispatcher(),
								new Callable<Future<KvStateLocation>>() {
									@Override
									public Future<KvStateLocation> call() throws Exception {
										return getKvStateLookupInfo(
												jobId,
												registrationName,
												lookupRetryStrategy);
									}
								});
					} else {
						return Futures.failed(failure);
					}
				}
			}, actorSystem.dispatcher());
}
 

import akka.dispatch.Recover; //导入依赖的package包/类
public Future<Object> queryStateFuture(final QueryState<K> queryState) {
	LOG.debug("Try to get ActorRef future for key {}.", queryState.getKey());

	Future<ActorRef> actorRefFuture = getActorRefFuture(queryState.getKey());

	@SuppressWarnings("unchecked")
	Future<Object> result =  actorRefFuture.flatMap(new Mapper<ActorRef, Future<Object>>() {
		public Future<Object> apply(ActorRef actorRef) {
			LOG.debug("Ask response actor for state for key {}.", queryState.getKey());
			return Patterns.ask(actorRef, queryState, new Timeout(askTimeout));
		}
	}, executor).recoverWith(new Recover<Future<Object>>() {
		@Override
		public Future<Object> recover(final Throwable failure) throws Throwable {
			if (failure instanceof WrongKeyPartitionException || failure instanceof ActorNotFound) {
				// wait askTimeout because we communicated with the wrong actor. This usually
				// indicates that not all actors have registered at the registry.
				return Patterns.after(
					askTimeout,
					getContext().system().scheduler(),
					executor,
					new Callable<Future<Object>>() {
						@Override
						public Future<Object> call() throws Exception {
							refreshCache();
							return Futures.failed(failure);
						}
					});
			} else if (failure instanceof AskTimeoutException) {
				LOG.debug("Ask timed out.", failure);
				handleAskTimeout();
				return Futures.failed(failure);
			} else {
				LOG.debug("State query failed with.", failure);
				refreshCache();
				return Futures.failed(failure);
			}
		}
	}, executor);

	return result;
}
 

import akka.dispatch.Recover; //导入依赖的package包/类
public static <T> Recover<T> recoverF( Function<Throwable, T> f ){		
	
	return new Recover<T>() {

		@Override
		public T recover(Throwable t ) throws Throwable {
			
			return f.apply( t );
			
		}
	};
	
}