在分布式计算，远程过程调用（Remote Procedure Call，缩写为 RPC）是一个计算机通信协议。该协议允许运行于一台计算机的程序调用另一个地址空间（通常为一个开放网络的一台计算机）的子程序，而程序员就像调用本地程序一样，无需额外地为这个交互作用编程（无需关注细节）。RPC是一种服务器-客户端（Client/Server）模式，是一个通过"发送请求-接受回应"进行信息交互的系统。
Zookeeper
它为大型分布式计算提供开源的分布式配置服务、同步服务和命名注册。
ZooKeeper的架构通过冗余服务实现高可用性。因此，如果第一次无应答，客户端就可以询问另一台ZooKeeper主机。ZooKeeper节点将它们的数据存储于一个分层的命名空间，非常类似于一个文件系统或一个前缀树结构。客户端可以在节点读写，从而以这种方式拥有一个共享的配置服务。
本项目通过zookeeper实现server端服务的注册和发现。
Akka
Akka实现了actor模型，为高并发分布式系统提供了可靠的消息传输机制。
Actor 的状态是本地的而不是共享的，更改和数据通过消息传播，每个Actor一次处理一条消息，不存在竞争，所以没有使用锁，发送者也不会阻塞。数百万个 Actor 可以有效地安排在十几个线程上，从而充分发挥现代 CPU 的潜力。
对于Rpc的调用请求，可以非常容易的使用akka的actor之间的消息机制实现。
著名的spark以及flink的RPC模块都是基于akka的。
本项目通过akka的actor集群实现RPC调用。
项目架构图
platform.jpg
RPC的client以及server端分别通过两个不同的Actor Cluster实现，client端的request通过作为负载均衡路由的Actor发送到不同的server端的Actor上处理，并且把结果返回。
核心代码
Actor
client 端
public class RpcClientActor extends UntypedActor {
private ActorRef rpc;
public RpcClientActor(ActorRef rpc) {
// 路由actor
this.rpc = rpc;
}
@Override
public void onReceive(Object message) throws Exception {
if (message instanceof RpcRequest) {
//五秒钟内获取不到结果，会报错
Future<Object> future = Patterns.ask(rpc, message, new Timeout(
Duration.create(5, TimeUnit.SECONDS)));
Object o = Await.result(future,
Duration.create(5, TimeUnit.SECONDS));
//返回结果
getSender().tell(o, getSelf());
}else
unhandled(message);
}
}
server 端
public class RpcServerActor extends UntypedActor {
private Map<String, Object> proxyBeans;
public RpcServerActor(Map<Class<?>, Object> beans) {
proxyBeans = new HashMap<String, Object>();
for (Iterator<Class<?>> iterator = beans.keySet().iterator(); iterator
.hasNext();) {
Class<?> inface = iterator.next();
proxyBeans.put(inface.getName(), beans.get(inface));
}
}
@Override
public void onReceive(Object message) throws Exception {
if (message instanceof RpcRequest) {
RpcRequest request=(RpcRequest)message;
Object serviceClass = proxyBeans.get(request.getClassName());
Object[] params = request.getParameters();
Class<?>[] paramerTypes = new Class<?>[params.length];
if (params != null) {
for (int i=0;i<params.length;i++) {
paramerTypes[i]=params[i].getClass();
}
}
//使用 jdk 执行反射调用
// Method method = serviceClass.getClass().getMethod(request.getMethodName(),
// paramerTypes);
// Object o = method.invoke(serviceClass, params);
//使用 CGLib 执行反射调用
FastClass serviceFastClass = FastClass.create(serviceClass.getClass());
FastMethod serviceFastMethod = serviceFastClass.getMethod(request.getMethodName(), paramerTypes);
Object o = serviceFastMethod.invoke(serviceClass, params);
RpcResponse response=new RpcResponse();
response.setData(o);
System.out.println(this.getSelf().path());
getSender().tell(response, getSelf());
}else
unhandled(message);
}
}
client 集群配置
public AkkaRpcClient() {
final Config config = ConfigFactory
.parseString("akka.remote.netty.tcp.port=" + 2552)
.withFallback(
ConfigFactory
.parseString("akka.cluster.roles = [RpcClient]"))
.withFallback(ConfigFactory.load());
//创建集群根节点
system = ActorSystem.create("RpcSystem", config);
int totalInstances = 100;
List<String> paths=null;
try { // 服务发现
ZooKeeperClient zkClient = new ZooKeeperClient("localhost:2181",null,null);
ZooKeeperServiceDiscovery discovery = new ZooKeeperServiceDiscovery(zkClient);
paths = discovery.discover("root");
} catch (Exception e) {
e.printStackTrace();
}
List<String> routeesPaths=new ArrayList<>();
for(String path:paths){
routeesPaths.add("/user/"+path);
}
boolean allowLocalRoutees = false;
ClusterRouterGroup clusterRouterGroup = new ClusterRouterGroup(
new AdaptiveLoadBalancingGroup(
HeapMetricsSelector.getInstance(),
Collections.<String> emptyList()),
new ClusterRouterGroupSettings(totalInstances, routeesPaths,
allowLocalRoutees, "RpcServer1"));
//创建集群负载路由Actor
rpc = system.actorOf(clusterRouterGroup.props(), "rpcCall");
//创建client
clientServer = system.actorOf(Props.create(RpcClientActor.class, rpc),
"rpcClient");
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
Cluster.get(system).registerOnMemberUp(new Runnable() {
@Override
public void run() {
synchronized (instance) {
System.out.println("notify");
instance.notify();
}
}
});
}
server 集群配置
final Config config = ConfigFactory
.parseString("akka.remote.netty.tcp.port=" + 2551)
.withFallback(
ConfigFactory
.parseString("akka.cluster.roles = [RpcServer1]"))
.withFallback(
ConfigFactory
.parseString("akka.remote.netty.tcp.hostname=127.0.0.1"))
.withFallback(ConfigFactory.load());
//创建server端集群根节点
ActorSystem system = ActorSystem.create("RpcSystem", config);
// Server 加入发布的服务
Map<Class<?>, Object> services = new HashMap<Class<?>, Object>();
services.put(Rzk.class, new Hello());
//新建RpcServerActor,并且加入集群,rpcServer1与AkkaRpcClient中对应routeesPaths
system.actorOf(Props.create(RpcServerActor.class, services), "rpcServer1");
system.actorOf(Props.create(RpcServerActor.class, services), "rpcServer2");
system.actorOf(Props.create(RpcServerActor.class, services), "rpcServer3");
try {
//服务发布
ZooKeeperClient zkClient = new ZooKeeperClient("localhost:2181",null,null);
ZooKeeperServiceRegistry registry = new ZooKeeperServiceRegistry(zkClient);
registry.register("rpcServer1");
registry.register("rpcServer2");
} catch (IOException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
项目运行
client 端
[INFO] [11/30/2020 11:16:45.064] [main] [Remoting] Remoting started; listening on addresses :[akka.tcp://RpcSystem@127.0.0.1:2552]
[INFO] [11/30/2020 11:16:45.078] [main] [Cluster(akka://RpcSystem)] Cluster Node [akka.tcp://RpcSystem@127.0.0.1:2552] - Starting up...
[WARN] [11/30/2020 11:16:45.079] [main] [Cluster(akka://RpcSystem)] [akka.cluster.auto-down] setting is replaced by [akka.cluster.auto-down-unreachable-after]
[INFO] [11/30/2020 11:16:45.182] [main] [Cluster(akka://RpcSystem)] Cluster Node [akka.tcp://RpcSystem@127.0.0.1:2552] - Registered cluster JMX MBean [akka:type=Cluster]
[INFO] [11/30/2020 11:16:45.182] [main] [Cluster(akka://RpcSystem)] Cluster Node [akka.tcp://RpcSystem@127.0.0.1:2552] - Started up successfully
[INFO] [11/30/2020 11:16:45.289] [RpcSystem-akka.actor.default-dispatcher-15] [Cluster(akka://RpcSystem)] Cluster Node [akka.tcp://RpcSystem@127.0.0.1:2552] - Metrics will be retreived from MBeans, and may be incorrect on some platforms. To increase metric accuracy add the 'sigar.jar' to the classpath and the appropriate platform-specific native libary to 'java.library.path'. Reason: java.lang.IllegalArgumentException: java.lang.UnsatisfiedLinkError: org.hyperic.sigar.Sigar.getPid()J
[INFO] [11/30/2020 11:16:45.290] [RpcSystem-akka.actor.default-dispatcher-15] [Cluster(akka://RpcSystem)] Cluster Node [akka.tcp://RpcSystem@127.0.0.1:2552] - Metrics collection has started successfully
[INFO] [11/30/2020 11:16:45.714] [RpcSystem-akka.actor.default-dispatcher-3] [Cluster(akka://RpcSystem)] Cluster Node [akka.tcp://RpcSystem@127.0.0.1:2552] - Welcome from [akka.tcp://RpcSystem@127.0.0.1:2551]
wait
notify
hello rpc 2
hello rpc 87
hello rpc 91
hello rpc 95
hello rpc 98
hello rpc 101
hello rpc 104
hello rpc 107
hello rpc 110
hello rpc 114
hello rpc 117
hello rpc 121
hello rpc 124
hello rpc 126
hello rpc 129
hello rpc 132
...
server 端
[INFO] [11/30/2020 11:15:41.608] [RpcSystem-akka.actor.default-dispatcher-15] [akka://RpcSystem/system/cluster/core/daemon/firstSeedNodeProcess-1] Message [akka.dispatch.sysmsg.Terminate] from Actor[akka://RpcSystem/system/cluster/core/daemon/firstSeedNodeProcess-1#-75735353] to Actor[akka://RpcSystem/system/cluster/core/daemon/firstSeedNodeProcess-1#-75735353] was not delivered. [7] dead letters encountered. This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.
[INFO] [11/30/2020 11:15:41.612] [RpcSystem-akka.actor.default-dispatcher-21] [Cluster(akka://RpcSystem)] Cluster Node [akka.tcp://RpcSystem@127.0.0.1:2551] - Node [akka.tcp://RpcSystem@127.0.0.1:2551] is JOINING, roles [RpcServer1]
[INFO] [11/30/2020 11:15:42.600] [RpcSystem-akka.actor.default-dispatcher-16] [Cluster(akka://RpcSystem)] Cluster Node [akka.tcp://RpcSystem@127.0.0.1:2551] - Leader is moving node [akka.tcp://RpcSystem@127.0.0.1:2551] to [Up]
akka://RpcSystem/user/rpcServer2
akka://RpcSystem/user/rpcServer1
akka://RpcSystem/user/rpcServer2
akka://RpcSystem/user/rpcServer1
akka://RpcSystem/user/rpcServer1
akka://RpcSystem/user/rpcServer2
akka://RpcSystem/user/rpcServer2
akka://RpcSystem/user/rpcServer2
akka://RpcSystem/user/rpcServer1
akka://RpcSystem/user/rpcServer1
akka://RpcSystem/user/rpcServer2
akka://RpcSystem/user/rpcServer2
akka://RpcSystem/user/rpcServer1
akka://RpcSystem/user/rpcServer2
akka://RpcSystem/user/rpcServer2
akka://RpcSystem/user/rpcServer2
akka://RpcSystem/user/rpcServer2
akka://RpcSystem/user/rpcServer1
akka://RpcSystem/user/rpcServer2
akka://RpcSystem/user/rpcServer1
akka://RpcSystem/user/rpcServer1
akka://RpcSystem/user/rpcServer2
akka://RpcSystem/user/rpcServer2
...
可以看到，client端的请求被分配到不同的server actor上进行了处理。
本项目github地址：https://github.com/tomliu112/rpc-zk-akka
