為什么80%的碼農都做不了架構師？>>> ??

序

本文主要研究一下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);}@Overridepublic 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);}}@Overridepublic 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.");}}@Overridepublic <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.");}}@Overridepublic void tell(Object message) {super.tell(fenceMessage(message));}@Overridepublic CompletableFuture<?> ask(Object message, Time timeout) {return super.ask(fenceMessage(message), timeout);}@Overridepublic 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;}@Overridepublic 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);}@Overridepublic F getFencingToken() {return fencingToken;}@Overridepublic P getPayload() {return payload;}@Overridepublic 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);}@Overridepublic F getFencingToken() {return fencingToken;}@Overridepublic P getPayload() {return payload;}@Overridepublic 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

轉載于:https://my.oschina.net/go4it/blog/3022817

總結

以上是生活随笔為你收集整理的聊聊flink的FencedAkkaInvocationHandler的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。