一个分布式rpc框架的实现方案（二）

之前实现了一个点对点的rpc功能框架，使用简单的协议将调用接口和参数传给服务端。由于最近在看zk相关的内容，于是准备引入zk进行分布式管理。
先看一下前一个简单版本的实现原理：

这是一个简单的点对点的远程调用实现方案， client绑定了server的ip地址。那么如果有多个server，并且server的变化情况要及时通知client应该怎么实现？引入了zookeeper后这个问题就比较容易解决了。以下两张图表达了基于zk的分布式rpc设计思想。

分布式服务的改造比较简单，将前一个版本的中写死的ip地址改成从zk中查询服务地址，根据负载均衡策略选择一个server的ip, 请求这个ip对应的服务就ok了。使用zookeeper管理server集群的方式如下:

服务方在zookeeper上创建一个临时节点，节点的名称是服务器的ip。客户端读zookeeper同目录下面的子节点列表，拿到server的ip地址，再选择一个ip作为请求方调用服务。如果服务器挂掉了，由于创建的是临时节点，因此zookeeper也会同步删掉这个ZNode，所以客户端就不会再看到对应的服务ip了。
具体代码如下：
1. 实现一个zk 的监控者

public class ZKWatcher implements Watcher {
protected static final int SESSION_TIME = 2000;
protected ZooKeeper zooKeeper;

protected String SERVER_PATH = "/rpcServer";
public void init() throws IOException, KeeperException, InterruptedException{}
public ZKWatcher(){
try {
            init();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

// 防止链接未完成就创建节点
protected   CountDownLatch connectedSemaphore  = new CountDownLatch(1);
public void process(WatchedEvent event) {
if (event.getState() == Event.KeeperState.SyncConnected) {  // 链接成功
            connectedSemaphore.countDown();
        }
    }

    SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss");
public boolean createEphemeralFile(String file){
boolean result = false;
try {
            String f = SERVER_PATH + "/" + file;
            Stat stat = zooKeeper.exists(f, false);
if (stat == null) {
                zNode = zooKeeper.create(f, ("time:" + df.format(new Date()) + " " + getLocalIp()).getBytes(),
                        ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
            }else{
                zooKeeper.setData(f, ("time:" + df.format(new Date()) + " " + getLocalIp()).getBytes(), -1);
            }

            result = true;
        }catch (Exception e){
            System.out.println("error to create zk server file");
            e.printStackTrace();
        }

return result;
    }
}

服务端注册服务，先创建/rpcServer目录

public class ZKServer extends ZKWatcher {

/**
 * 注册服务端
 * @param ip
 * @return
 */
public boolean registServer(String ip) {
    boolean result = false;
try{
if(StringUtils.isEmpty(ip)){
            ip = getLocalIp();
        }
result = this.createEphemeralFile(ip);
    }catch (Exception e){

    }
if (result){
System.out.println("success to regist rpc server:" + ip);
    }else{
System.out.println("fail to regist rpc server:" + ip);
    }

return result;
}

public void init() throws IOException, KeeperException, InterruptedException {
if (zooKeeper == null){
        zooKeeper = new ZooKeeper(getLocalIp(), SESSION_TIME, this);
        connectedSemaphore.await();
    }

Stat stat = zooKeeper.exists(SERVER_PATH, false);
if (stat != null) {
return;
    }
String zNode = zooKeeper.create(SERVER_PATH, ("init:" + getLocalIp()).getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
System.out.println("create server path:" + zNode);
}
}

服务provider在init时将自己ip注册到zk集群上

public class SocketProviderV2{
    ……
public Object provide() throws Throwable {
        ServerSocket serverSocket = new ServerSocket(dataSource.getPort());
// 注册zk服务器
        zkServer.registServer(null);

while (true) {
            Socket socket = null;
try {
// open socket
                socket = serverSocket.accept();
this.getRpcRequest(socket);
// 拿到请求后续处理
               ……
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
 }

rpc客户端查询zk中的服务节点，选择一个作为请求地址

public class ZKClient extends ZKWatcher {
public List<String> getServerIps() throws KeeperException, InterruptedException {
    List<String> ips = new ArrayList<String>();
try{
        ips = zooKeeper.getChildren(SERVER_PATH, false);
    }catch (Exception e){
        e.printStackTrace();
    }

return ips;
}

// 随机数实现负载均衡
Random random = new Random(1000);
public String getBalanceIp() throws KeeperException, InterruptedException {
    List<String> ips = this.getServerIps();

int size = ips.size();
int index = (int)(Math.random() * size);
return ips.get(index);
}

public void init() throws IOException, KeeperException, InterruptedException {
if (zooKeeper == null){
        zooKeeper = new ZooKeeper(getLocalIp(), SESSION_TIME, this);
        connectedSemaphore.await();
    }

    System.out.println("I finished init");
}
}

socket client实现

public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
StringBuffer stream = this.buildRpcRequest(proxy, method, args);

Object object = null;
try{
    long startTime =  System.currentTimeMillis();
System.out.println("New rpc client send " + stream.toString() + " time:" + startTime);
String serverIp = zkClient.getBalanceIp();
System.out.println("server ip is " + serverIp);

    // 调用远程服务器数据
    ……
}catch (IOException e){
    e.printStackTrace();
object = e.toString();
}

return object;
}

其他的代码详见之前版本。

我们看一下调用效果：启动3虚拟机+2个本地zk 服务端，部署分布式zk环境。将rpc provider代码打成jar包传到3个虚拟机中，运营服务程序。
服务1

root@master:/home/caisheng# java -jar lengren-rpc.jar
begin start service v2
success to regist rpc server:192.168.226.128

服务2

root@slave01:/home/caisheng# java -jar lengren-rpc.jar
begin start service v2
success to regist rpc server:192.168.226.129

服务3

root@slave02:/home/caisheng# java -jar lengren-rpc.jar
begin start service v2
success to regist rpc server:192.168.226.130

调出zkclient，查看一下zk集群的node情况。

OK,服务全部注册成功。本地运行rpc client,
Client: 调用到ip 130的机器上

 I finished init
New rpc client send version:1.0.0|##|cypher:default|##|interfaceName:proxyPrintText|##|interface:proxyLearn.beans.PrintText|##|method:print|##|params:java.lang.String+test rpc, time:1467282617593
server ip is 192.168.226.130
client read from service:success aop, result=i am real printtest rpc  time:14
success aop, result=i am real printtest rpc

130 server:

再次运行Client: 这次调用到了ip 128的机器上。

I finished init
New rpc client send version:1.0.0|##|cypher:default|##|interfaceName:proxyPrintText|##|interface:proxyLearn.beans.PrintText|##|method:print|##|params:java.lang.String+test rpc, time:1467280709114
server ip is 192.168.226.128
client read from service:success aop, result=i am real printtest rpc  time:17
success aop, result=i am real printtest rpc

128 server:

未完待续，下次将解决更加细节的问题，例如线程池的维护和NIO。

秒客网

一个分布式rpc框架的实现方案（二）

相关文章