- 使用 Spring Initializr 创建 Spring Boot 应用程序
- 在Spring Boot中配置Cassandra
- 在 Spring Boot 上配置 Tomcat 连接池
- 将Camel消息路由到嵌入WildFly的Artemis上
相关文章:
【Flink】Flink 源码之RPC调用
Flink】FLink 通讯组件 RPC
作为一个分布式系统,Flink 内部不同组件之间通信依赖于 RPC 机制。这篇文章将对 Flink 的 RPC 框架加以分析。
先来看一个简单的例子,了解 Flink 内部的 RPC 框架是如何使用的。
public class RpcTest {
private static final Time TIMEOUT = Time.seconds(10L);
private static ActorSystem actorSystem = null;
private static RpcService rpcService = null;
// 定义通信协议
public interface HelloGateway extends RpcGateway {
String hello();
}
public interface HiGateway extends RpcGateway {
String hi();
}
// 具体实现
public static class HelloRpcEndpoint extends RpcEndpoint implements HelloGateway {
protected HelloRpcEndpoint(RpcService rpcService) {
super(rpcService);
}
@Override
public String hello() {
return "hello";
}
}
public static class HiRpcEndpoint extends RpcEndpoint implements HiGateway {
protected HiRpcEndpoint(RpcService rpcService) {
super(rpcService);
}
@Override
public String hi() {
return "hi";
}
}
@BeforeClass
public static void setup() {
actorSystem = AkkaUtils.createDefaultActorSystem();
// 创建 RpcService, 基于 AKKA 的实现
rpcService = new AkkaRpcService(actorSystem, AkkaRpcServiceConfiguration.defaultConfiguration());
}
@AfterClass
public static void teardown() throws Exception {
final CompletableFuture<Void> rpcTerminationFuture = rpcService.stopService();
final CompletableFuture<Terminated> actorSystemTerminationFuture = FutureUtils.toJava(actorSystem.terminate());
FutureUtils
.waitForAll(Arrays.asList(rpcTerminationFuture, actorSystemTerminationFuture))
.get(TIMEOUT.toMilliseconds(), TimeUnit.MILLISECONDS);
}
@Test
public void test() throws Exception {
HelloRpcEndpoint helloEndpoint = new HelloRpcEndpoint(rpcService);
HiRpcEndpoint hiEndpoint = new HiRpcEndpoint(rpcService);
helloEndpoint.start();
//获取 endpoint 的 self gateway
HelloGateway helloGateway = helloEndpoint.getSelfGateway(HelloGateway.class);
String hello = helloGateway.hello();
assertEquals("hello", hello);
hiEndpoint.start();
// 通过 endpoint 的地址获得代理
HiGateway hiGateway = rpcService.connect(hiEndpoint.getAddress(),HiGateway.class).get();
String hi = hiGateway.hi();
assertEquals("hi", hi);
}
}
基本的使用流程就是1)定义协议,提供 RPC 方法的实现;2)获得服务对象的代理对象,调用 RPC 方法。
RpcEndpoint
是对 RPC 框架中提供具体服务的实体的抽象,所有提供远程调用方法的组件都需要继承该抽象类
。另外,对于同一个 RpcEndpoint 的所有 RPC 调用都会在同一个线程(RpcEndpoint 的“主线程”)
中执行,因此无需担心并发执行的线程安全问题。
RpcGateway
接口是用于远程调用的代理接口
。 RpcGateway 提供了获取其所代理的 RpcEndpoint 的地址的方法。在实现一个提供 RPC 调用的组件时,通常需要先定一个接口,该接口继承 RpcGateway 并约定好提供的远程调用的方法。
RpcService
是 RpcEndpoint 的运行时环境
, RpcService
提供了启动 RpcEndpoint
, 连接到远端 RpcEndpoint
并返回远端 RpcEndpoint
的代理对象等方法。此外, RpcService
还提供了某些异步任务或者周期性调度任务
的方法。
RpcServer
相当于 RpcEndpoint
自身的的代理对象(self gateway)。RpcServer
是 RpcService
在启动了 RpcEndpoint
之后返回的对象,每一个 RpcEndpoint
对象内部都有一个 RpcServer
的成员变量,通过 getSelfGateway
方法就可以获得自身的代理,然后调用该Endpoint 提供的服务。
FencedRpcEndpoint
和 FencedRpcGate
要求在调用 RPC 方法时携带 token 信息,只有当调用方提供了 token 和 endpoint 的 token 一致时才允许调用。
前面介绍了 Flink 内部 RPC 框架的基本抽象,主要就是 RpcService, RpcEndpoint, RpcGateway, RpcServer 等接口。至于具体的实现,则可以有多种不同的方式,如 Akka, Netty 等。Flink 目前提供了一套基于 Akka 的实现。
AkkaRpcService
实现了 RpcService
接口, AkkaRpcService
会启动 Akka actor 来接收来自 RpcGateway
的 RPC 调用。
首先,在 RpcEndpoint
的构造函数中,会调用 AkkaRpcService#startServer
方法来初始化服务,AkkaRpcService#startServer
的主要工作包括: - 创建一个 Akka actor (AkkaRpcActor 或 FencedAkkaRpcActor)
- 通过动态代理创建代理对象
class AkkaRpcService {
@Override
public <C extends RpcEndpoint & RpcGateway> RpcServer startServer(C rpcEndpoint) {
checkNotNull(rpcEndpoint, "rpc endpoint");
CompletableFuture<Void> terminationFuture = new CompletableFuture<>();
final Props akkaRpcActorProps;
if (rpcEndpoint instanceof FencedRpcEndpoint) {
akkaRpcActorProps = Props.create(
FencedAkkaRpcActor.class,
rpcEndpoint,
terminationFuture,
getVersion(),
configuration.getMaximumFramesize());
} else {
akkaRpcActorProps = Props.create(
AkkaRpcActor.class,
rpcEndpoint,
terminationFuture,
getVersion(),
configuration.getMaximumFramesize());
}
ActorRef actorRef;
// 创建 Akka actor
synchronized (lock) {
checkState(!stopped, "RpcService is stopped");
actorRef = actorSystem.actorOf(akkaRpcActorProps, rpcEndpoint.getEndpointId());
actors.put(actorRef, rpcEndpoint);
}
LOG.info("Starting RPC endpoint for {} at {} .", rpcEndpoint.getClass().getName(), actorRef.path());
final String akkaAddress = AkkaUtils.getAkkaURL(actorSystem, actorRef);
final String hostname;
Option<String> host = actorRef.path().address().host();
if (host.isEmpty()) {
hostname = "localhost";
} else {
hostname = host.get();
}
// 代理的接口
Set<Class<?>> implementedRpcGateways = new HashSet<>(RpcUtils.extractImplementedRpcGateways(rpcEndpoint.getClass()));
implementedRpcGateways.add(RpcServer.class);
implementedRpcGateways.add(AkkaBasedEndpoint.class);
final InvocationHandler akkaInvocationHandler;
//创建 InvocationHandler
if (rpcEndpoint instanceof FencedRpcEndpoint) {
// a FencedRpcEndpoint needs a FencedAkkaInvocationHandler
akkaInvocationHandler = new FencedAkkaInvocationHandler<>(
akkaAddress,
hostname,
actorRef,
configuration.getTimeout(),
configuration.getMaximumFramesize(),
terminationFuture,
((FencedRpcEndpoint<?>) rpcEndpoint)::getFencingToken);
implementedRpcGateways.add(FencedMainThreadExecutable.class);
} else {
akkaInvocationHandler = new AkkaInvocationHandler(
akkaAddress,
hostname,
actorRef,
configuration.getTimeout(),
configuration.getMaximumFramesize(),
terminationFuture);
}
// 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();
//通过动态代理创建代理对象
@SuppressWarnings("unchecked")
RpcServer server = (RpcServer) Proxy.newProxyInstance(
classLoader,
implementedRpcGateways.toArray(new Class<?>[implementedRpcGateways.size()]),
akkaInvocationHandler);
return server;
}
}
在 RpcEndpoint
对象创建后,下一步操作是启动它,实际上调用的是 RpcServer.start()
方法。RpcServer
是通过 AkkaInvocationHandler
创建的动态代理对象:
class AkkaInvocationHandler {
private final ActorRef rpcEndpoint;
public void start() {
//向 Akka actor 发送 START 消息
rpcEndpoint.tell(ControlMessages.START, ActorRef.noSender());
}
}
所以启动 RpcEndpoint 实际上就是向当前 endpoint 绑定的 Actor 发送一条 START 消息,通知服务启动。
在 RpcEndpoint
创建的过程中,实际上已经通过动态代理生成了一个可供本地使用的代理对象,通过 RpcEndpoint#getSelfGateway
方法可以直接获取。
class RpcEndpoint {
public <C extends RpcGateway> C getSelfGateway(Class<C> selfGatewayType) {
//rpcServer 是通过动态代理创建的
if (selfGatewayType.isInstance(rpcServer)) {
@SuppressWarnings("unchecked")
C selfGateway = ((C) rpcServer);
return selfGateway;
} else {
throw new RuntimeException("RpcEndpoint does not implement the RpcGateway interface of type " + selfGatewayType + '.');
}
}
}
如果需要获取一个远程 RpcEndpoint
的代理,就需要通过 RpcService#connect
方法,需要提供远程 endpoint 的地址:
class AkkaRpcService {
private <C extends RpcGateway> CompletableFuture<C> connectInternal(
final String address,
final Class<C> clazz,
Function<ActorRef, InvocationHandler> invocationHandlerFactory) {
checkState(!stopped, "RpcService is stopped");
LOG.debug("Try to connect to remote RPC endpoint with address {}. Returning a {} gateway.",
address, clazz.getName());
final ActorSelection actorSel = actorSystem.actorSelection(address);
final Future<ActorIdentity> identify = Patterns
.ask(actorSel, new Identify(42), configuration.getTimeout().toMilliseconds())
.<ActorIdentity>mapTo(ClassTag$.MODULE$.<ActorIdentity>apply(ActorIdentity.class));
final CompletableFuture<ActorIdentity> identifyFuture = FutureUtils.toJava(identify);
//获取 actor 的引用 ActorRef
final CompletableFuture<ActorRef> actorRefFuture = identifyFuture.thenApply(
(ActorIdentity actorIdentity) -> {
if (actorIdentity.getRef() == null) {
throw new CompletionException(new RpcConnectionException("Could not connect to rpc endpoint under address " + address + '.'));
} else {
return actorIdentity.getRef();
}
});
//发送握手消息
final CompletableFuture<HandshakeSuccessMessage> handshakeFuture = actorRefFuture.thenCompose(
(ActorRef actorRef) -> FutureUtils.toJava(
Patterns
.ask(actorRef, new RemoteHandshakeMessage(clazz, getVersion()), configuration.getTimeout().toMilliseconds())
.<HandshakeSuccessMessage>mapTo(ClassTag$.MODULE$.<HandshakeSuccessMessage>apply(HandshakeSuccessMessage.class))));
// 创建 InvocationHandler,并通过动态代理生成代理对象
return actorRefFuture.thenCombineAsync(
handshakeFuture,
(ActorRef actorRef, HandshakeSuccessMessage ignored) -> {
InvocationHandler invocationHandler = invocationHandlerFactory.apply(actorRef);
// 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();
@SuppressWarnings("unchecked")
C proxy = (C) Proxy.newProxyInstance(
classLoader,
new Class<?>[]{clazz},
invocationHandler);
return proxy;
},
actorSystem.dispatcher());
}
}
上述方法主要的功能包括:
RpcEndpoint
绑定的 actor 的引用 ActorRef
AkkaRpcActor
发送握手消息AkkaInvocationHandler
对象,并通过动态代理生成代理对象在获取了本地或者远端 RpcEndpoint
的代理对象后,就可以通过代理对象发起 RPC 调用了。由于代理对象是通过动态代理创建的,因而所以的方法都会转化为 AkkaInvocationHandler#invoke
方法,并传入 RPC 调用的方法以及参数信息。
class AkkaInvocationHandler {
@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;
}
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);
//将 RPC 调用封装为 RpcInvocation(会根据RpcEndpoint是本地还是远程的)
final RpcInvocation rpcInvocation = createRpcInvocationMessage(methodName, parameterTypes, args);
Class<?> returnType = method.getReturnType();
final Object result;
//根据RPC方法是否有返回值决定调用 tell 还是 ask
if (Objects.equals(returnType, Void.TYPE)) {
//akka actor tell
tell(rpcInvocation);
result = null;
} else {
// execute an asynchronous call
//akka actor ask
CompletableFuture<?> resultFuture = ask(rpcInvocation, futureTimeout);
CompletableFuture<?> completableFuture = resultFuture.thenApply((Object o) -> {
if (o instanceof SerializedValue) {
try {
return ((SerializedValue<?>) o).deserializeValue(getClass().getClassLoader());
} catch (IOException | ClassNotFoundException e) {
throw new CompletionException(
new RpcException("Could not deserialize the serialized payload of RPC method : "
+ methodName, e));
}
} else {
return o;
}
});
if (Objects.equals(returnType, CompletableFuture.class)) {
result = completableFuture;
} else {
try {
result = completableFuture.get(futureTimeout.getSize(), futureTimeout.getUnit());
} catch (ExecutionException ee) {
throw new RpcException("Failure while obtaining synchronous RPC result.", ExceptionUtils.stripExecutionException(ee));
}
}
}
return result;
}
}
对于 RPC 调用,需要将 RPC 调用的方法名、参数类型和参数值封装为一个 RpcInvocation 对象,根据 RpcEndpoint 是本地的还是远端,具体的 有 LocalRpcInvocation 和 RemoteRpcInvocation 两类,它们的区别在于是否需要序列化。
然后根据 RPC 方法是否有返回值,决定调用 tell 或 ask 方法,然后通过 Akka 的 ActorRef 向对应的 AkkaRpcActor 发送请求,如果带有返回值,则等待 actor 的响应。
AkkaRpcActor 负责接受 RPC 调用的请求,并通过反射调用 RpcEndpoint 的对应方法来完成 RPC 调用。
class AkkaRpcActor<T extends RpcEndpoint & RpcGateway> extends AbstractActor {
protected final T rpcEndpoint;
@Override
public Receive createReceive() {
//不同类型消息的处理方法
return ReceiveBuilder.create()
.match(RemoteHandshakeMessage.class, this::handleHandshakeMessage)
.match(ControlMessages.class, this::handleControlMessage)
.matchAny(this::handleMessage)
.build();
}
//处理 RPC 调用
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.getClass().getName());
sendErrorIfSender(new AkkaRpcException(
String.format("Discard message, because the rpc endpoint %s has not been started yet.", rpcEndpoint.getAddress())));
}
}
private void handleRpcInvocation(RpcInvocation rpcInvocation) {
Method rpcMethod = null;
try {
String methodName = rpcInvocation.getMethodName();
Class<?>[] parameterTypes = rpcInvocation.getParameterTypes();
//获去需要调用的方法
rpcMethod = lookupRpcMethod(methodName, parameterTypes);
} catch (ClassNotFoundException e) {
log.error("Could not load method arguments.", e);
RpcConnectionException rpcException = new RpcConnectionException("Could not load method arguments.", e);
getSender().tell(new Status.Failure(rpcException), getSelf());
} catch (IOException e) {
log.error("Could not deserialize rpc invocation message.", e);
RpcConnectionException rpcException = new RpcConnectionException("Could not deserialize rpc invocation message.", e);
getSender().tell(new Status.Failure(rpcException), getSelf());
} 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);
if (rpcMethod.getReturnType().equals(Void.TYPE)) {
// No return value to send back
rpcMethod.invoke(rpcEndpoint, rpcInvocation.getArgs());
}
else {
final Object result;
try {
result = rpcMethod.invoke(rpcEndpoint, rpcInvocation.getArgs());
}
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();
//向调用方发送执行结果
if (result instanceof CompletableFuture) {
final CompletableFuture<?> responseFuture = (CompletableFuture<?>) result;
sendAsyncResponse(responseFuture, methodName);
} else {
sendSyncResponse(result, methodName);
}
}
} 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 框架。首先,通过 RpcService, RpcEndpoint, RpcGateway, RpcServer 等接口和抽象类,确定了 RPC 服务的基本框架;在这套框架的基础上, Flink 借助 Akka 和动态代理等技术提供了 RPC 调用的具体实现。
我正在尝试读取一个大型日志文件,该文件已使用不同的分隔符(遗留更改)进行了解析。 此代码有效 import os, subprocess, time, re import pandas as pd f
我试图理解在 Linux 下以 Turbo 模式(特别是 fpc -Mtp -vw)编译的 Free Pascal 中看到的有点神奇的行为。代码来自 Jack Crenshaw 的“让我们构建一个编译
我有一个具有以下结构的 txt 文件: NAME DATA1 DATA2 a 10 1,2,3 b 6 8,9 c 2
我试图理解在 Linux 下以 Turbo 模式(特别是 fpc -Mtp -vw)编译的 Free Pascal 中看到的有点神奇的行为。代码来自 Jack Crenshaw 的“让我们构建一个编译
public class Bug1 { private String s; public void Bug1(){ s = "hello"; } public Stri
我们有这样一种情况,我们的应用程序需要处理一系列文件,而不是同步执行此功能,我们希望采用多线程将工作负载分配给不同的线程。 每一项工作是: 1.以只读方式打开文件 2.处理文件中的数据 3.将处理后的
我正在尝试读取 .php 文件并替换十六进制字符。php文件格式如下: 问题是它弄乱了转义字符 (\") 到目前为止我的代码: while(i=48 && str[i+2]=97 && str[i+
我正在用 C# 开发一个程序,我需要一些帮助。我正在尝试创建一个数组或项目列表,显示在某个网站上。我想要做的是阅读 anchor 文本,它是 href。例如,这是 HTML:
我有一个偏好设置,它控制我的应用程序是否在用户单击按钮时播放声音(这种情况经常发生,想想计算器)。每次用户单击按钮时,都会调用以下方法: private void playButtonClickSou
我正在尝试在我的标签末尾创建一个阅读更多按钮。我希望它默认显示 3 行。我正在用 swift 而不是 objective c 编写代码。只有当用户点击标签的阅读更多部分时,标签才会展开。它的外观和工作
当您获得第三方库(c、c++)、开源(LGPL 说)但没有很好的文档时,了解它以便能够集成到您的应用程序中的最佳方法是什么? 该库通常有一些示例程序,我最终使用 gdb 浏览了代码。还有其他建议/最佳
同时从 2 个或更多不同线程对同一个文件描述符使用 pread 是否有问题? 最佳答案 pread 本身是线程安全的,因为它不在 list of unsafe functions 上.所以调用它是安全
当您使用命令 pd.read_csv 读取 csv 时,如何跳过连续包含特定值的行?如果在第 50、55 行,第一列的值为 100,那么我想在读取 csv 文件时跳过这些行。我如何将这些命令放入像 p
我迫切需要在 C# 中使用 T4 生成 HTML 输出。 我正在使用 Runtime-T4-Files 并选择“TextTemplatingFilePreprocessor”而不是“TextTempl
今年夏天我在实习期间一直在学习 ERP 应用程序。由于我是一名即将毕业的程序员,我希望有一个可靠的软件分支可以帮助我完成工作,直到我确定下一步该做什么(直到我对大局有一个很好的了解)。到现在为止,我刚
将包含列(例如“a”、“b”)的数据帧保存为 parquet,然后在稍后的时间点读取 parquet 不会提供相同的列顺序(可能是“b”、“a”fe)文件保存为。 不幸的是,我无法弄清楚订单是如何受到
我正在开发一个使用谷歌表格作为数据库的应用程序,但我不知道如何让 Swift 从谷歌表格中读取。我浏览了 API 网站和一些问题,但刚开始我需要一些帮助。到目前为止,我有; 私有(private)让范
我打算阅读swing concept,如果值得一读,请推荐一些学习 Material 最佳答案 自 AWT 崩溃以来,Java 的 GUI 工具包太多了。即使是 Swing 也被评论家严重低估,但他们
我已经使用 J 几个月了,我发现阅读不熟悉的代码(例如,不是我自己写的)是该语言最具挑战性的方面之一,尤其是在默认情况下。过了一会儿,我想出了这个策略: 1)将代码段复制到word文档中 2)从(1)
很难说出这里问的是什么。这个问题是含糊的、模糊的、不完整的、过于宽泛的或修辞性的,无法以目前的形式得到合理的回答。如需帮助澄清此问题以便重新打开它,visit the help center 。 已关
我是一名优秀的程序员,十分优秀!