前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >聊聊flink的RpcServer

聊聊flink的RpcServer

原创
作者头像
code4it
发布2019-03-14 09:43:33
8660
发布2019-03-14 09:43:33
举报
文章被收录于专栏:码匠的流水账

本文主要研究一下flink的RpcServer

RpcGateway

flink-release-1.7.2/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcGateway.java

代码语言:javascript
复制
public interface RpcGateway {
​
    /**
     * Returns the fully qualified address under which the associated rpc endpoint is reachable.
     *
     * @return Fully qualified (RPC) address under which the associated rpc endpoint is reachable
     */
    String getAddress();
​
    /**
     * Returns the fully qualified hostname under which the associated rpc endpoint is reachable.
     *
     * @return Fully qualified hostname under which the associated rpc endpoint is reachable
     */
    String getHostname();
}
  • RpcGateway定义了getAddress、getHostname两个方法

MainThreadExecutable

flink-release-1.7.2/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/MainThreadExecutable.java

代码语言:javascript
复制
public interface MainThreadExecutable {
​
    /**
     * Execute the runnable in the main thread of the underlying RPC endpoint.
     *
     * @param runnable Runnable to be executed
     */
    void runAsync(Runnable runnable);
​
    /**
     * Execute the callable in the main thread of the underlying RPC endpoint and return a future for
     * the callable result. If the future is not completed within the given timeout, the returned
     * future will throw a {@link TimeoutException}.
     *
     * @param callable Callable to be executed
     * @param callTimeout Timeout for the future to complete
     * @param <V> Return value of the callable
     * @return Future of the callable result
     */
    <V> CompletableFuture<V> callAsync(Callable<V> callable, Time callTimeout);
​
    /**
     * Execute the runnable in the main thread of the underlying RPC endpoint, with
     * a delay of the given number of milliseconds.
     *
     * @param runnable Runnable to be executed
     * @param delay    The delay, in milliseconds, after which the runnable will be executed
     */
    void scheduleRunAsync(Runnable runnable, long delay);
}
  • MainThreadExecutable定义了runAsync、callAsync、scheduleRunAsync三个方法

StartStoppable

flink-release-1.7.2/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/StartStoppable.java

代码语言:javascript
复制
public interface StartStoppable {
​
    /**
     * Starts the processing of remote procedure calls.
     */
    void start();
​
    /**
     * Stops the processing of remote procedure calls.
     */
    void stop();
}
  • StartStoppable定义了start、stop方法

RpcServer

flink-release-1.7.2/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/RpcServer.java

代码语言:javascript
复制
public interface RpcServer extends StartStoppable, MainThreadExecutable, RpcGateway {
​
    /**
     * Return a future which is completed when the rpc endpoint has been terminated.
     *
     * @return Future indicating when the rpc endpoint has been terminated
     */
    CompletableFuture<Void> getTerminationFuture();
}
  • RpcServer接口继承了RpcGateway、MainThreadExecutable、StartStoppable三个接口,另外定义了getTerminationFuture方法;它有两个实现类,分别是AkkaInvocationHandler、FencedAkkaInvocationHandler;其中FencedAkkaInvocationHandler继承了AkkaInvocationHandler

AkkaBasedEndpoint

flink-release-1.7.2/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaBasedEndpoint.java

代码语言:javascript
复制
interface AkkaBasedEndpoint extends RpcGateway {
​
    /**
     * Returns the {@link ActorRef} of the underlying RPC actor.
     *
     * @return the {@link ActorRef} of the underlying RPC actor
     */
    ActorRef getActorRef();
}
  • AkkaBasedEndpoint接口继承了RpcGateway接口,它另外定义了getActorRef()方法用于获取ActorRef

AkkaInvocationHandler

flink-release-1.7.2/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaInvocationHandler.java

代码语言:javascript
复制
class AkkaInvocationHandler implements InvocationHandler, AkkaBasedEndpoint, RpcServer {
    private static final Logger LOG = LoggerFactory.getLogger(AkkaInvocationHandler.class);
​
    /**
     * The Akka (RPC) address of {@link #rpcEndpoint} including host and port of the ActorSystem in
     * which the actor is running.
     */
    private final String address;
​
    /**
     * Hostname of the host, {@link #rpcEndpoint} is running on.
     */
    private final String hostname;
​
    private final ActorRef rpcEndpoint;
​
    // whether the actor ref is local and thus no message serialization is needed
    protected final boolean isLocal;
​
    // default timeout for asks
    private final Time timeout;
​
    private final long maximumFramesize;
​
    // null if gateway; otherwise non-null
    @Nullable
    private final CompletableFuture<Void> terminationFuture;
​
    AkkaInvocationHandler(
            String address,
            String hostname,
            ActorRef rpcEndpoint,
            Time timeout,
            long maximumFramesize,
            @Nullable CompletableFuture<Void> terminationFuture) {
​
        this.address = Preconditions.checkNotNull(address);
        this.hostname = Preconditions.checkNotNull(hostname);
        this.rpcEndpoint = Preconditions.checkNotNull(rpcEndpoint);
        this.isLocal = this.rpcEndpoint.path().address().hasLocalScope();
        this.timeout = Preconditions.checkNotNull(timeout);
        this.maximumFramesize = maximumFramesize;
        this.terminationFuture = terminationFuture;
    }
​
    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        Class<?> declaringClass = method.getDeclaringClass();
​
        Object result;
​
        if (declaringClass.equals(AkkaBasedEndpoint.class) ||
            declaringClass.equals(Object.class) ||
            declaringClass.equals(RpcGateway.class) ||
            declaringClass.equals(StartStoppable.class) ||
            declaringClass.equals(MainThreadExecutable.class) ||
            declaringClass.equals(RpcServer.class)) {
            result = method.invoke(this, args);
        } else if (declaringClass.equals(FencedRpcGateway.class)) {
            throw new UnsupportedOperationException("AkkaInvocationHandler does not support the call FencedRpcGateway#" +
                method.getName() + ". This indicates that you retrieved a FencedRpcGateway without specifying a " +
                "fencing token. Please use RpcService#connect(RpcService, F, Time) with F being the fencing token to " +
                "retrieve a properly FencedRpcGateway.");
        } else {
            result = invokeRpc(method, args);
        }
​
        return result;
    }
​
    @Override
    public ActorRef getActorRef() {
        return rpcEndpoint;
    }
​
    @Override
    public void runAsync(Runnable runnable) {
        scheduleRunAsync(runnable, 0L);
    }
​
    @Override
    public void scheduleRunAsync(Runnable runnable, long delayMillis) {
        checkNotNull(runnable, "runnable");
        checkArgument(delayMillis >= 0, "delay must be zero or greater");
​
        if (isLocal) {
            long atTimeNanos = delayMillis == 0 ? 0 : System.nanoTime() + (delayMillis * 1_000_000);
            tell(new RunAsync(runnable, atTimeNanos));
        } else {
            throw new RuntimeException("Trying to send a Runnable to a remote actor at " +
                rpcEndpoint.path() + ". This is not supported.");
        }
    }
​
    @Override
    public <V> CompletableFuture<V> callAsync(Callable<V> callable, Time callTimeout) {
        if (isLocal) {
            @SuppressWarnings("unchecked")
            CompletableFuture<V> resultFuture = (CompletableFuture<V>) ask(new CallAsync(callable), callTimeout);
​
            return resultFuture;
        } else {
            throw new RuntimeException("Trying to send a Callable to a remote actor at " +
                rpcEndpoint.path() + ". This is not supported.");
        }
    }
​
    @Override
    public void start() {
        rpcEndpoint.tell(Processing.START, ActorRef.noSender());
    }
​
    @Override
    public void stop() {
        rpcEndpoint.tell(Processing.STOP, ActorRef.noSender());
    }
​
    // ------------------------------------------------------------------------
    //  Private methods
    // ------------------------------------------------------------------------
​
    private Object invokeRpc(Method method, Object[] args) throws Exception {
        String methodName = method.getName();
        Class<?>[] parameterTypes = method.getParameterTypes();
        Annotation[][] parameterAnnotations = method.getParameterAnnotations();
        Time futureTimeout = extractRpcTimeout(parameterAnnotations, args, timeout);
​
        final RpcInvocation rpcInvocation = createRpcInvocationMessage(methodName, parameterTypes, args);
​
        Class<?> returnType = method.getReturnType();
​
        final Object result;
​
        if (Objects.equals(returnType, Void.TYPE)) {
            tell(rpcInvocation);
​
            result = null;
        } else if (Objects.equals(returnType, CompletableFuture.class)) {
            // execute an asynchronous call
            result = ask(rpcInvocation, futureTimeout);
        } else {
            // execute a synchronous call
            CompletableFuture<?> futureResult = ask(rpcInvocation, futureTimeout);
​
            result = futureResult.get(futureTimeout.getSize(), futureTimeout.getUnit());
        }
​
        return result;
    }
​
    protected RpcInvocation createRpcInvocationMessage(
            final String methodName,
            final Class<?>[] parameterTypes,
            final Object[] args) throws IOException {
        final RpcInvocation rpcInvocation;
​
        if (isLocal) {
            rpcInvocation = new LocalRpcInvocation(
                methodName,
                parameterTypes,
                args);
        } else {
            try {
                RemoteRpcInvocation remoteRpcInvocation = new RemoteRpcInvocation(
                    methodName,
                    parameterTypes,
                    args);
​
                if (remoteRpcInvocation.getSize() > maximumFramesize) {
                    throw new IOException("The rpc invocation size exceeds the maximum akka framesize.");
                } else {
                    rpcInvocation = remoteRpcInvocation;
                }
            } catch (IOException e) {
                LOG.warn("Could not create remote rpc invocation message. Failing rpc invocation because...", e);
                throw e;
            }
        }
​
        return rpcInvocation;
    }
​
    // ------------------------------------------------------------------------
    //  Helper methods
    // ------------------------------------------------------------------------
​
    private static Time extractRpcTimeout(Annotation[][] parameterAnnotations, Object[] args, Time defaultTimeout) {
        if (args != null) {
            Preconditions.checkArgument(parameterAnnotations.length == args.length);
​
            for (int i = 0; i < parameterAnnotations.length; i++) {
                if (isRpcTimeout(parameterAnnotations[i])) {
                    if (args[i] instanceof Time) {
                        return (Time) args[i];
                    } else {
                        throw new RuntimeException("The rpc timeout parameter must be of type " +
                            Time.class.getName() + ". The type " + args[i].getClass().getName() +
                            " is not supported.");
                    }
                }
            }
        }
​
        return defaultTimeout;
    }
​
    private static boolean isRpcTimeout(Annotation[] annotations) {
        for (Annotation annotation : annotations) {
            if (annotation.annotationType().equals(RpcTimeout.class)) {
                return true;
            }
        }
​
        return false;
    }
​
    protected void tell(Object message) {
        rpcEndpoint.tell(message, ActorRef.noSender());
    }
​
    protected CompletableFuture<?> ask(Object message, Time timeout) {
        return FutureUtils.toJava(
            Patterns.ask(rpcEndpoint, message, timeout.toMilliseconds()));
    }
​
    @Override
    public String getAddress() {
        return address;
    }
​
    @Override
    public String getHostname() {
        return hostname;
    }
​
    @Override
    public CompletableFuture<Void> getTerminationFuture() {
        return terminationFuture;
    }
}
  • AkkaInvocationHandler实现了RpcServer、AkkaBasedEndpoint、jdk的InvocationHandler接口;其构造器要求输入address、hostname、rpcEndpoint(ActorRef)、terminationFuture;getAddress、getHostname、getTerminationFuture均直接返回对应的属性
  • runAsync方法内部调用的是scheduleRunAsync;scheduleRunAsync方法使用的是tell方法,调用rpcEndpoint.tell传递RunAsync消息;callAsync方法使用的是ask方法,调用Patterns.ask,传递CallAsync消息
  • start方法执行rpcEndpoint.tell(Processing.START, ActorRef.noSender());stop方法执行rpcEndpoint.tell(Processing.STOP, ActorRef.noSender());invoke方法针对Object、RpcGateway、MainThreadExecutable、StartStoppable、AkkaBasedEndpoint、RpcServer的方法则对当前对象进行对应方法调用,针对FencedRpcGateway的方法抛出UnsupportedOperationException,其余的方法则内部调用invokeRpc方法,构造RpcInvocation消息进行调用

小结

  • RpcServer接口继承了RpcGateway、MainThreadExecutable、StartStoppable三个接口,另外定义了getTerminationFuture方法;它有两个实现类,分别是AkkaInvocationHandler、FencedAkkaInvocationHandler;其中FencedAkkaInvocationHandler继承了AkkaInvocationHandler
  • AkkaInvocationHandler实现了RpcServer、AkkaBasedEndpoint、jdk的InvocationHandler接口;其构造器要求输入address、hostname、rpcEndpoint(ActorRef)、terminationFuture;getAddress、getHostname、getTerminationFuture均直接返回对应的属性;runAsync方法内部调用的是scheduleRunAsync;scheduleRunAsync方法使用的是tell方法,调用rpcEndpoint.tell传递RunAsync消息;callAsync方法使用的是ask方法,调用Patterns.ask,传递CallAsync消息
  • AkkaInvocationHandler的start方法执行rpcEndpoint.tell(Processing.START, ActorRef.noSender());stop方法执行rpcEndpoint.tell(Processing.STOP, ActorRef.noSender());invoke方法针对Object、RpcGateway、MainThreadExecutable、StartStoppable、AkkaBasedEndpoint、RpcServer的方法则对当前对象进行对应方法调用,针对FencedRpcGateway的方法抛出UnsupportedOperationException,其余的方法则内部调用invokeRpc方法,构造RpcInvocation消息进行调用

doc

原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。

如有侵权,请联系 cloudcommunity@tencent.com 删除。

原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。

如有侵权,请联系 cloudcommunity@tencent.com 删除。

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • RpcGateway
  • MainThreadExecutable
  • StartStoppable
  • RpcServer
  • AkkaBasedEndpoint
  • AkkaInvocationHandler
  • 小结
  • doc
相关产品与服务
大数据
全栈大数据产品,面向海量数据场景,帮助您 “智理无数,心中有数”!
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档