3.精通Flink1.19—动态代理

 

1. 本节课目的

核心点 本节课核心:彻底理解Java动态代理，彻底理解Flink底层源码中哪些地方用到了动态代理。 这节课我们只聚焦动态代理。

2.开始本节内容

2.1.Java动态代理

动态代理是一种设计模式，它允许在运行时创建代理对象，并将方法调用重定向到不同的实际对象。它使我们能够在不修改现有代码的情况下增加或改变某个对象的行为。

InvocationHandler接口: 这个接口定义了一个invoke方法，该方法在代理实例上的方法被调用时被调用。 public interface InvocationHandler { public Object invoke(Object proxy, Method method, Object[] args) throws Throwable; } Proxy类:这个类提供了创建动态代理类和实例的静态方法。 public static Object newProxyInstance(ClassLoader loader, Class<?>[] interfaces, InvocationHandler h) throws IllegalArgumentException ClassLoader loader： 这个类加载器用于定义代理类的类加载器。通常，我们可以使用被代理对象的类加载器，即 targetObject.getClass().getClassLoader()。 代理类必须和它所表示的接口在同一个类加载器的命名空间中，以确保代理类能够访问被代理的接口。 总结：类加载器，targetObject.getClass().getClassLoader()。 Class<?>[] interfaces： 这是一个接口数组，表示代理类需要实现的接口。 代理类将实现这些接口，并可以在运行时动态地调用这些接口的方法。 总结：动态代理类会调用实现该接口的方法 InvocationHandler h： 这是一个调用处理程序，它负责实现接口中的方法调用。 当代理类的方法被调用时，实际上会调用这个 InvocationHandler 的 invoke 方法。invoke 方法会接收被调用的方法、方法的参数以及代理实例本身作为参数。 总结：动态代理类调用方法的时候，会流转到invoke方法中，在invoke方法中可以完成我们要做的操作，比如打印日志

2.2.Java 动态代理Demo

2.2.1.ResourceManagerGateway 接口

使用java动态代理定义一个接口。实际调用对象会实现该接口

package com.source.proxy; /** * 模拟Flink为代理目标对象定义一个接口 */ public interface ResourceManagerGateway { /** * 定义一个注册方法 */ void registerTaskExecutor(); }

2.2.1.ResourceManager 类

1.定义ResourceManager 类实现ResourceManagerGateway 接口

2.内部实现接口方法打印一句话

package com.source.proxy; /** * 创建实现该接口的目标对象 */ public class ResourceManager implements ResourceManagerGateway{ /** * 实现方法中打印一句话 */ @Override public void registerTaskExecutor() { System.out.println("注册registerTaskExecutor"); } }

2.2.1.PekkoInvocationHandler类

1.定义PekkoInvocationHandler 类实现java InvocationHandler

2.内部顶一个Object目标对象

3.invoke方法中直接调用invokeRpc方法（为了模拟Flink内部代码）

4.invokeRpc方法内部调用目标方法前后打印两行话

package com.source.proxy; import java.lang.reflect.InvocationHandler; import java.lang.reflect.Method; /** * 创建实现InvocationHandler接口的类 */ public class PekkoInvocationHandler implements InvocationHandler { /**目标对象*/ private Object target; public PekkoInvocationHandler(Object target) { this.target = target; } /** * * 在invoke中调用内部方法invokeRpc * @return * @throws Throwable */ @Override public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { return invokeRpc(method,args); } /** * 在invokeRpc中实现自己的逻辑，比如向ResourceManager发送pekko的请求 * flink内部实现的时候会将调用的代理类和方法封装成RpcInvocation 调用ask方法发送给PekkoRpcActor接收到消息 * 内部调用HandlerMessage处理不同类型的请求然后通过java反射调用最终调用传递给ResourceManager.registerTaskExecutor方法 * * @param method * @param args * @return * @throws Exception */ private Object invokeRpc(Method method, Object[] args) throws Exception { System.out.println("调用pekko ask方法向ResourceManager发送调用的方法"); Object result = method.invoke(target,args); System.out.println("结束调用"); return result; } }

2.2.1.Demo接口

1.创建ResourceManager 对象

2.创建PekkoInvocationHandler

3.调用Proxy.newProxyInstance创建代理对象，内部参数为类加载器、定义的ResourceManagerGateway.class、PekkoInvocationHandler 实例对象

4.通过代理对象调用registerTaskExecutor()

5.此时程序就会跳转到PekkoInvocationHandler 的invoke方法，并进入invokeRpc

6.打印第一句话，之后method.invoke通过反射机制调用到目标类的ResourceManager.registerTaskExecutor方法

7.打印第二句话，返回并结束

package com.source.proxy; import java.lang.reflect.Proxy; public class Demo { public static void main(String[] args) { /** 创建目标对象 */ ResourceManager myObject = new ResourceManager(); /** 创建InvocationHandler */ PekkoInvocationHandler handler = new PekkoInvocationHandler(myObject); /** 调用 Proxy.newProxyInstance静态方法创建动态代理类 */ ResourceManagerGateway proxy = (ResourceManagerGateway) Proxy.newProxyInstance( ResourceManagerGateway.class.getClassLoader(), new Class<?>[] { ResourceManagerGateway.class }, handler); /** 调用registerTaskExecutor 注册方法最终会调用PekkoInvocationHandler的invoke方法 */ proxy.registerTaskExecutor(); } }

运行结果

调用pekko ask方法向ResourceManager发送调用的方法 注册registerTaskExecutor 结束调用

2.3.Flink RPC中的动态代理详解

这里以Flink ResourceManager组件为例阅读源码

2.3.1 RpcGateway 接口

1.Flink底层通信用的就是java动态代理。

2.因此站在动态代理的角度考虑，目标类实现的接口在Flink 内部最终都是RpcGateway

3.比如ResourceManager目标类实现的接口ResourceManagerGateway(ResourceManagerGateway继承自FencedRpcGateway)

public interface ResourceManagerGateway extends FencedRpcGateway<ResourceManagerId>, ClusterPartitionManager, BlocklistListener { CompletableFuture<RegistrationResponse> registerTaskExecutor( TaskExecutorRegistration taskExecutorRegistration, @RpcTimeout Time timeout); }

2.3.2.ResourceManager实现类

1.动态代理的角度考虑ResourceManager就是目标类

public abstract class ResourceManager<WorkerType extends ResourceIDRetrievable> extends FencedRpcEndpoint<ResourceManagerId> implements DelegationTokenManager.Listener, ResourceManagerGateway { /** * ResourceManager真正注册的接口方法 * @param taskExecutorRegistration the task executor registration. * @param timeout The timeout for the response. * @return */ @Override public CompletableFuture<RegistrationResponse> registerTaskExecutor( final TaskExecutorRegistration taskExecutorRegistration, final Time timeout) { } }

2.3.3.PekkoInvocationHandler类

1.PekkoInvocationHandler类实现了java InvocationHandler

2.那也就是说Flink 底层通信的时候调用目标方法的时候会跳转到PekkoInvocationHandler类的invoke方法中

3.invoke方法内部

1)如果本地调用直接反射method.invoke调用方法。

2）远程调用invokeRpc方法，先发送RPC请求，在目标类中才会调用method.invoke执行方法的调用

class PekkoInvocationHandler implements InvocationHandler, PekkoBasedEndpoint, RpcServer { /**动态代理的回调方法*/ @Override public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { Class<?> declaringClass = method.getDeclaringClass(); Object result; /** * 本地调用 * 判断是否是基本实现接口 */ if (declaringClass.equals(PekkoBasedEndpoint.class) || declaringClass.equals(Object.class) || declaringClass.equals(RpcGateway.class) || declaringClass.equals(StartStoppable.class) || declaringClass.equals(MainThreadExecutable.class) || declaringClass.equals(RpcServer.class)) { /** java 反射机制调用*/ result = method.invoke(this, args); } else if (declaringClass.equals(FencedRpcGateway.class)) { throw new UnsupportedOperationException( "InvocationHandler 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; } /** * 远程调用代理对象 * 1.将调用的方法名字、参数类型、参数封装为RpcInvocation * 2.调用Pekko tell、ask方法进行发送 */ private Object invokeRpc(Method method, Object[] args) throws Exception { String methodName = method.getName(); Class<?>[] parameterTypes = method.getParameterTypes(); final boolean isLocalRpcInvocation = method.getAnnotation(Local.class) != null; Annotation[][] parameterAnnotations = method.getParameterAnnotations(); Duration futureTimeout = RpcGatewayUtils.extractRpcTimeout(parameterAnnotations, args, timeout); /**构建RpcInvocation * 里面存放关键信息。方法名字，参数类型，参数 */ final RpcInvocation rpcInvocation = createRpcInvocationMessage( method.getDeclaringClass().getSimpleName(), methodName, isLocalRpcInvocation, parameterTypes, args); /** 获取方法返回类型 */ Class<?> returnType = method.getReturnType(); final Object result; /**如果方法返回void 类型，直接调用tell 否则ask*/ if (Objects.equals(returnType, Void.TYPE)) { tell(rpcInvocation); result = null; } else { // Capture the call stack. It is significantly faster to do that via an exception than // via Thread.getStackTrace(), because exceptions lazily initialize the stack trace, // initially only // capture a lightweight native pointer, and convert that into the stack trace lazily // when needed. final Throwable callStackCapture = captureAskCallStack ? new Throwable() : null; // execute an asynchronous call final CompletableFuture<?> resultFuture = ask(rpcInvocation, futureTimeout) .thenApply( resultValue -> deserializeValueIfNeeded( resultValue, method, flinkClassLoader)); final CompletableFuture<Object> completableFuture = new CompletableFuture<>(); resultFuture.whenComplete( (resultValue, failure) -> { if (failure != null) { completableFuture.completeExceptionally( resolveTimeoutException( ExceptionUtils.stripCompletionException(failure), callStackCapture, address, rpcInvocation)); } else { completableFuture.complete(resultValue); } }); if (Objects.equals(returnType, CompletableFuture.class)) { result = completableFuture; } else { try { result = completableFuture.get(futureTimeout.toMillis(), TimeUnit.MILLISECONDS); } catch (ExecutionException ee) { throw new RpcException( "Failure while obtaining synchronous RPC result.", ExceptionUtils.stripExecutionException(ee)); } } } return result; } }

2.3.4.PekkoRpcService类

1.PekkoRpcService类就是Flink RPC通信的服务类

2.通过startServer可以看出来内部用到了动态代理相关的代码，同时会创建动态代理对象

/** * 启动RpcEndpoint中的RpcServer后，RpcEndpoint组件仅仅有对外提供处理Rpc请求的能力 * 并不具备发送的能力。 * @param rpcEndpoint * @return * @param <C> */ @Override public <C extends RpcEndpoint & RpcGateway> RpcServer startServer(C rpcEndpoint) { checkNotNull(rpcEndpoint, "rpc endpoint"); /** * 创建Actor */ final SupervisorActor.ActorRegistration actorRegistration = registerRpcActor(rpcEndpoint); /** 获取Actor对应的RpcActor */ final ActorRef actorRef = actorRegistration.getActorRef(); final CompletableFuture<Void> actorTerminationFuture = actorRegistration.getTerminationFuture(); LOG.info( "Starting RPC endpoint for {} at {} .", rpcEndpoint.getClass().getName(), actorRef.path()); /** 获取Actor的路径*/ final String address = PekkoUtils.getRpcURL(actorSystem, actorRef); /** 获取hostname*/ final String hostname; Option<String> host = actorRef.path().address().host(); if (host.isEmpty()) { hostname = "localhost"; } else { hostname = host.get(); } /** * 解析RpcEndpoint是的所有RpcGateway */ Set<Class<?>> implementedRpcGateways = new HashSet<>(RpcUtils.extractImplementedRpcGateways(rpcEndpoint.getClass())); /**添加额外的RpcServer、PekkoBasedEndpoint类*/ implementedRpcGateways.add(RpcServer.class); implementedRpcGateways.add(PekkoBasedEndpoint.class); final InvocationHandler invocationHandler; /** 根据不同的RpcEndpoint类型创建动态代理对象的Handler */ if (rpcEndpoint instanceof FencedRpcEndpoint) { // a FencedRpcEndpoint needs a FencedPekkoInvocationHandler invocationHandler = new FencedPekkoInvocationHandler<>( address, hostname, actorRef, configuration.getTimeout(), configuration.getMaximumFramesize(), configuration.isForceRpcInvocationSerialization(), actorTerminationFuture, ((FencedRpcEndpoint<?>) rpcEndpoint)::getFencingToken, captureAskCallstacks, flinkClassLoader); } else { invocationHandler = new PekkoInvocationHandler( address, hostname, actorRef, configuration.getTimeout(), configuration.getMaximumFramesize(), configuration.isForceRpcInvocationSerialization(), actorTerminationFuture, captureAskCallstacks, flinkClassLoader); } // Rather than using the System ClassLoader directly, we derive the ClassLoader // from this class . That works better in cases where Flink runs embedded and all Flink // code is loaded dynamically (for example from an OSGI bundle) through a custom ClassLoader ClassLoader classLoader = getClass().getClassLoader(); /**生成RpcServer对象，然后根据改服务的调用就会金融到Handlerde invoke方法处理*/ @SuppressWarnings("unchecked") RpcServer server = (RpcServer) Proxy.newProxyInstance( classLoader, implementedRpcGateways.toArray( new Class<?>[implementedRpcGateways.size()]), invocationHandler); return server; }

2.3.5.PekkoRpcActor类

补充内容印象就行，后面会详细讲解

1.PekkoRpcActor是继承AbstractActor抽象类

2.所有的RPC通信都会被PekkoRpcActor.createReceive方法接收到

3.createReceive方法根据消息类型进入到对应的方法进行处理，最常用的就是handleMessage

@Override public Receive createReceive() { return ReceiveBuilder.create() /** 处理握手类的消息*/ .match(RemoteHandshakeMessage.class, this::handleHandshakeMessage) /** 控制类的消息，比如start 方法，stop方法*/ .match(ControlMessages.class, this::handleControlMessage) /** 处理其他所有类型的消息*/ .matchAny(this::handleMessage) .build(); } private void handleMessage(final Object message) { if (state.isRunning()) { mainThreadValidator.enterMainThread(); try { handleRpcMessage(message); } finally { mainThreadValidator.exitMainThread(); } } else { log.info( "The rpc endpoint {} has not been started yet. Discarding message {} until processing is started.", rpcEndpoint.getClass().getName(), message); sendErrorIfSender( new EndpointNotStartedException( String.format( "Discard message %s, because the rpc endpoint %s has not been started yet.", message, getSelf().path()))); } } protected void handleRpcMessage(Object message) { if (message instanceof RunAsync) { handleRunAsync((RunAsync) message); } else if (message instanceof CallAsync) { handleCallAsync((CallAsync) message); } else if (message instanceof RpcInvocation) { handleRpcInvocation((RpcInvocation) message); } } private void handleRpcInvocation(RpcInvocation rpcInvocation) { Method rpcMethod = null; try { String methodName = rpcInvocation.getMethodName(); Class<?>[] parameterTypes = rpcInvocation.getParameterTypes(); rpcMethod = lookupRpcMethod(methodName, parameterTypes); } catch (final NoSuchMethodException e) { log.error("Could not find rpc method for rpc invocation.", e); RpcConnectionException rpcException = new RpcConnectionException("Could not find rpc method for rpc invocation.", e); getSender().tell(new Status.Failure(rpcException), getSelf()); } if (rpcMethod != null) { try { // this supports declaration of anonymous classes rpcMethod.setAccessible(true); final Method capturedRpcMethod = rpcMethod; if (rpcMethod.getReturnType().equals(Void.TYPE)) { // No return value to send back runWithContextClassLoader( () -> capturedRpcMethod.invoke(rpcEndpoint, rpcInvocation.getArgs()), flinkClassLoader); } else { final Object result; try { result = runWithContextClassLoader( () -> capturedRpcMethod.invoke( rpcEndpoint, rpcInvocation.getArgs()), flinkClassLoader); } catch (InvocationTargetException e) { log.debug( "Reporting back error thrown in remote procedure {}", rpcMethod, e); // tell the sender about the failure getSender().tell(new Status.Failure(e.getTargetException()), getSelf()); return; } final String methodName = rpcMethod.getName(); final boolean isLocalRpcInvocation = rpcMethod.getAnnotation(Local.class) != null; if (result instanceof CompletableFuture) { final CompletableFuture<?> responseFuture = (CompletableFuture<?>) result; sendAsyncResponse(responseFuture, methodName, isLocalRpcInvocation); } else { sendSyncResponse(result, methodName, isLocalRpcInvocation); } } } catch (Throwable e) { log.error("Error while executing remote procedure call {}.", rpcMethod, e); // tell the sender about the failure getSender().tell(new Status.Failure(e), getSelf()); } } }

Flink RPC底层使用动态代理做什么

1.动态代理用到的所有类都是RpcGateway 实现，也就是说创建的是RpcGateway接口对应实现类的动态代理，比如ResourceManagerGateway接口

2.目标实现类ResourceManager实现了ResourceManagerGateway接口，实现了里面的方法

3.PekkoInvocationHandler类实现了InvocationHandler并实现invoke方法，是将代理类的方法，参数类型，参数封装为RpcInvocation对象，之后通过Pekko.tell 、Pekko.ask方法将RpcInvocation作为消息发送到代理接口所在的进程中。

4.Flink RPC动态代理调用目标方法后会进入到invoke方法

5.invoke方法会先判断是本地还是远程，如果是本地直接调用method.invoke执行方法，如果是远程将方法名、参数、参数类型封装为RpcInvocation对象，通过Pekko发送RPC通信

6.消息会被PekkoRpcActor类的createReceive接收到，然后进入handleMessage方法

7.handleMessage方法内部会调用method.invoke 反射执行目标方法

2.3.6Flink RPC debug调试

上节课我们讲解过TaskManager（TaskExecutor）启动后需要向ResourceManager进行注册

以TaskExecutor.registerTaskExecutor()为例查看最终是如何调用到ResourceManager.registerTaskExecutor()进行真正注册的

1.TaskExecutor.registerTaskExecutor调用方法

2.调用方法进入PekkoInvocationHandler.invoke方法

3.远程方法调用invoke()后流转到 PekkoRpcActor.createReceive()方法然后进入handleMessage方法

4.handleMessage方法内部会调用method.invoke 反射执行目标方法

5.最终是否进入ResourceManager.registerTaskExecutor()方法

课后要求：在理解的基础上知道Flink 哪些地方用到了动态代理相关的知识。理解一下动态代理Proxy、Invocation