共计 7560 个字符，预计需要花费 19 分钟才能阅读完成。
序
本文主要研究一下 flink 的 FencedAkkaInvocationHandler
FencedRpcGateway
flink-release-1.7.2/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/FencedRpcGateway.java
public interface FencedRpcGateway<F extends Serializable> extends RpcGateway {
/**
* Get the current fencing token.
*
* @return current fencing token
*/
F getFencingToken();
}
FencedRpcGateway 接口继承了 RpcGateway 接口，它定义一个泛型 F，即为 fencing token 的泛型
FencedMainThreadExecutable
flink-release-1.7.2/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/FencedMainThreadExecutable.java
public interface FencedMainThreadExecutable extends MainThreadExecutable {
/**
* Run the given runnable in the main thread without attaching a fencing token.
*
* @param runnable to run in the main thread without validating the fencing token.
*/
void runAsyncWithoutFencing(Runnable runnable);
/**
* Run the given callable in the main thread without attaching a fencing token.
*
* @param callable to run in the main thread without validating the fencing token.
* @param timeout for the operation
* @param <V> type of the callable result
* @return Future containing the callable result
*/
<V> CompletableFuture<V> callAsyncWithoutFencing(Callable<V> callable, Time timeout);
}
FencedMainThreadExecutable 接口继承了 MainThreadExecutable，它定义了 runAsyncWithoutFencing、callAsyncWithoutFencing 方法用于运行 unfenced runnable 或者 unfenced callable，之所以这样定义主要是因为 FencedMainThreadExecutable 继承了 MainThreadExecutable，因而 MainThreadExecutable 里头定义的 runAsync、callAsync、scheduleRunAsync 方法的语义就变成是 Fenced
FencedAkkaInvocationHandler
flink-release-1.7.2/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/FencedAkkaInvocationHandler.java
public class FencedAkkaInvocationHandler<F extends Serializable> extends AkkaInvocationHandler implements FencedMainThreadExecutable, FencedRpcGateway<F> {
private final Supplier<F> fencingTokenSupplier;
public FencedAkkaInvocationHandler(
String address,
String hostname,
ActorRef rpcEndpoint,
Time timeout,
long maximumFramesize,
@Nullable CompletableFuture<Void> terminationFuture,
Supplier<F> fencingTokenSupplier) {
super(address, hostname, rpcEndpoint, timeout, maximumFramesize, terminationFuture);
this.fencingTokenSupplier = Preconditions.checkNotNull(fencingTokenSupplier);
}
@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
Class<?> declaringClass = method.getDeclaringClass();
if (declaringClass.equals(FencedMainThreadExecutable.class) ||
declaringClass.equals(FencedRpcGateway.class)) {
return method.invoke(this, args);
} else {
return super.invoke(proxy, method, args);
}
}
@Override
public void runAsyncWithoutFencing(Runnable runnable) {
checkNotNull(runnable, “runnable”);
if (isLocal) {
getActorRef().tell(
new UnfencedMessage<>(new RunAsync(runnable, 0L)), ActorRef.noSender());
} else {
throw new RuntimeException(“Trying to send a Runnable to a remote actor at ” +
getActorRef().path() + “. This is not supported.”);
}
}
@Override
public <V> CompletableFuture<V> callAsyncWithoutFencing(Callable<V> callable, Time timeout) {
checkNotNull(callable, “callable”);
checkNotNull(timeout, “timeout”);
if (isLocal) {
@SuppressWarnings(“unchecked”)
CompletableFuture<V> resultFuture = (CompletableFuture<V>) FutureUtils.toJava(
Patterns.ask(
getActorRef(),
new UnfencedMessage<>(new CallAsync(callable)),
timeout.toMilliseconds()));
return resultFuture;
} else {
throw new RuntimeException(“Trying to send a Runnable to a remote actor at ” +
getActorRef().path() + “. This is not supported.”);
}
}
@Override
public void tell(Object message) {
super.tell(fenceMessage(message));
}
@Override
public CompletableFuture<?> ask(Object message, Time timeout) {
return super.ask(fenceMessage(message), timeout);
}
@Override
public F getFencingToken() {
return fencingTokenSupplier.get();
}
private <P> FencedMessage<F, P> fenceMessage(P message) {
if (isLocal) {
return new LocalFencedMessage<>(fencingTokenSupplier.get(), message);
} else {
if (message instanceof Serializable) {
@SuppressWarnings(“unchecked”)
FencedMessage<F, P> result = (FencedMessage<F, P>) new RemoteFencedMessage<>(fencingTokenSupplier.get(), (Serializable) message);
return result;
} else {
throw new RuntimeException(“Trying to send a non-serializable message ” + message + ” to a remote ” +
“RpcEndpoint. Please make sure that the message implements java.io.Serializable.”);
}
}
}
}
FencedAkkaInvocationHandler 继承了 AkkaInvocationHandler，实现了 FencedMainThreadExecutable、FencedRpcGateway 接口；runAsyncWithoutFencing、callAsyncWithoutFencing 发送的均为 UnfencedMessage
FencedAkkaInvocationHandler 的 invoke 方法针对 FencedRpcGateway、FencedMainThreadExecutable 的方法则对当前对象进行对应方法调用，其他的就转为调用父类的 invoke 方法
父类的 runAsync、scheduleRunAsync、callAsync 最后调用的是 tell 或者 ask 方法，而 FencedAkkaInvocationHandler 覆盖了父类的 tell 及 ask 方法，将 runAsync、scheduleRunAsync、callAsync 方法的语义变为 Fenced；这里的 tell 及 ask 方法通过 fenceMessage 方法构造 FencedMessage，而 fenceMessage 方法通过 getFencingToken 方法获取 fencing token；getFencingToken 方法调用的是 fencingTokenSupplier.get()，fencingTokenSupplier 由构造器传入
UnfencedMessage
flink-release-1.7.2/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/messages/UnfencedMessage.java
public class UnfencedMessage<P> {
private final P payload;
public UnfencedMessage(P payload) {
this.payload = Preconditions.checkNotNull(payload);
}
public P getPayload() {
return payload;
}
@Override
public String toString() {
return “UnfencedMessage(” + payload + ‘)’;
}
}
UnfencedMessage 即不需要 fencing token 的 message
FencedMessage
flink-release-1.7.2/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/messages/FencedMessage.java
public interface FencedMessage<F extends Serializable, P> {
F getFencingToken();
P getPayload();
}
FencedMessage 接口定义了 getFencingToken 及 getPayload 方法；它有两个子类，分别是 LocalFencedMessage 及 RemoteFencedMessage
LocalFencedMessage
flink-release-1.7.2/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/messages/LocalFencedMessage.java
public class LocalFencedMessage<F extends Serializable, P> implements FencedMessage<F, P> {
private final F fencingToken;
private final P payload;
public LocalFencedMessage(@Nullable F fencingToken, P payload) {
this.fencingToken = fencingToken;
this.payload = Preconditions.checkNotNull(payload);
}
@Override
public F getFencingToken() {
return fencingToken;
}
@Override
public P getPayload() {
return payload;
}
@Override
public String toString() {
return “LocalFencedMessage(” + fencingToken + “, ” + payload + ‘)’;
}
}
LocalFencedMessage 实现了 FencedMessage 接口，其中 fencingToken 的类型要求实现 Serializable 接口，它有 fencingToken 及 payload 两个属性
RemoteFencedMessage
flink-release-1.7.2/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/messages/RemoteFencedMessage.java
public class RemoteFencedMessage<F extends Serializable, P extends Serializable> implements FencedMessage<F, P>, Serializable {
private static final long serialVersionUID = 4043136067468477742L;
private final F fencingToken;
private final P payload;
public RemoteFencedMessage(@Nullable F fencingToken, P payload) {
this.fencingToken = fencingToken;
this.payload = Preconditions.checkNotNull(payload);
}
@Override
public F getFencingToken() {
return fencingToken;
}
@Override
public P getPayload() {
return payload;
}
@Override
public String toString() {
return “RemoteFencedMessage(” + fencingToken + “, ” + payload + ‘)’;
}
}
RemoteFencedMessage 实现了 FencedMessage 及 Serializable 接口，同时 payload 的类型也要求实现 Serializable 接口，它有 fencingToken 及 payload 两个属性
小结
FencedRpcGateway 接口继承了 RpcGateway 接口，它定义一个泛型 F，即为 fencing token 的泛型；FencedMainThreadExecutable 接口继承了 MainThreadExecutable，它定义了 runAsyncWithoutFencing、callAsyncWithoutFencing 方法用于运行 unfenced runnable 或者 unfenced callable
FencedAkkaInvocationHandler 继承了 AkkaInvocationHandler，实现了 FencedMainThreadExecutable、FencedRpcGateway 接口；runAsyncWithoutFencing、callAsyncWithoutFencing 发送的均为 UnfencedMessage；父类的 runAsync、scheduleRunAsync、callAsync 最后调用的是 tell 或者 ask 方法，而 FencedAkkaInvocationHandler 覆盖了父类的 tell 及 ask 方法，将 runAsync、scheduleRunAsync、callAsync 方法的语义变为 Fenced；这里的 tell 及 ask 方法通过 fenceMessage 方法构造 FencedMessage
UnfencedMessage 即不需要 fencing token 的 message；FencedMessage 接口定义了 getFencingToken 及 getPayload 方法；它有两个子类，分别是 LocalFencedMessage 及 RemoteFencedMessage；LocalFencedMessage 与 RemoteFencedMessage 的区别在于 RemoteFencedMessage 实现了 Serializable 接口，同时 payload 的类型也要求实现 Serializable 接口
doc
FencedAkkaInvocationHandler