java实现朴素rpc

网络五层协议中，试问 RPC 在第几层？

五层协议

应用层

传输层

网络层

链路层

物理层

进程协作完成任务时，通常需要相互通信。

本质上通信的方式分为本地和网络。

本地通信方式主要有两种形式，第一种是如命名管道基于操作系统暴漏的读写来操作，第二种是如锁机制的基于共享内存的线程队列阻塞机制（内存屏障）。

文章将要介绍一种网络通信方式，实现多台主机上进程相互通信的方式，是计算机网络的应用，是分布式的基础。

接口定义

rpc是远程过程调用，通过存根来进行数据定义，通过网络主机和端口建立传输层的连接和传。

首先先来看下存根，这个非常好理解，因为他其实就是空实现的本地方法调用。

public class EchoService {

	public static EchoRsp EchoSvr(EchoReq req) {
		return new EchoRsp("echo:" + req.content);
	}

	public static void main(String[] args) {
		System.out.println(EchoService.EchoSvr(new EchoReq("hello")).content);
	}
}

class EchoReq {
	String content;

	public EchoReq(String content) {
		this.content = content;
	}
}

class EchoRsp {
	String content;

	public EchoRsp(String content) {
		this.content = content;
	}
}

这是一个回声服务，将传入的数据简单处理后回传。

public static EchoRsp EchoSvr(EchoReq req) throws Exception {
	throw new UnsupportedOperationException();
}

存根相当于客户端并没有回声服务的实现，他只是提供了客户端需要关心的请求和结果。

编码解码

之前说到我们已经可以拿掉请求和结果了，接下来需要实现数据的序列化方式，用于数据在流上的传输。

class EchoReq implements Serializable {
	String content;

	public EchoReq(String content) {
		this.content = content;
	}
}

class EchoRsp implements Serializable {
	String content;

	public EchoRsp(String content) {
		this.content = content;
	}
}

序列化方式使用了Serializable，是语言默认的一种序列化的方式，使用十分简单，只需要实现序列化的接口即可。

定义好序列化方式，还需要定义那些数据在流上传输。

public class Header implements Serializable {
	String stub;
	String method;

	public Header(String stub, String method) {
		this.stub = stub;
		this.method = method;
	}
}

Header 用于制定存根和调用的方法，因此需要在流上传输。除了header外，还应该有服务的请求requset和服务的响应response。

public class Codec {
	InputStream inputStream;
	OutputStream outputStream;

	public Codec(InputStream inputStream, OutputStream outputStream) {
		this.inputStream = inputStream;
		this.outputStream = outputStream;
	}

	public Header Header() throws Exception {
		return (Header) new ObjectInputStream(inputStream).readObject();
	}

	public Object Read() throws Exception {
		return new ObjectInputStream(inputStream).readObject();
	}

	public void Write(Header header, Object obj) throws Exception {
		new ObjectOutputStream(outputStream).writeObject(header);
		new ObjectOutputStream(outputStream).writeObject(obj);
	}
}

定义好传输哪些数据后，编码解码器就好实现了。

服务端

在 rpc 中服务端承载了监听本机端口，并与客户端建立连接并通信数据。

ServerSocket serverSocket = new ServerSocket(8080);
Socket socket = serverSocket.accept();

通过 socket 方式，很容易监听本机的端口，没有客户端连接时会阻塞到 accept。

Codec cc = new Codec(socket.getInputStream(), socket.getOutputStream());
Header header = cc.Header();
Class stub = Class.forName(header.stub);
Map methods = Arrays.asList(stub.getDeclaredMethods()).stream()
	.collect(Collectors.toMap(t -> t.getName(), t -> t));
Method method = methods.get(header.method);
cc.Write(header, method.invoke(null, header, cc.Read()));

服务端通过编码器可以读出客户端要调用的存根和方法，在通过反射的方式可以加载对应的类并进行调用。

现在已经实现了基本的服务端，但是这个服务端的吞吐量受制，需要对客户端的请求排队处理。

ExecutorService iopool = Executors.newFixedThreadPool(100);
Socket socket = serverSocket.accept();
iopool.submit(() -> {
  // ...
});

对于网络请求这种密集io的应用，通常使用池化技术来处理，并发多个线程就可以成倍加快吞吐量。

客户端

大部分客户端都是支持水平扩容的，对客户端来说单机客户端池化加长连接几乎可以覆盖大部分场景。

class Call {
	Long seq;
	Object req;
	Object rsp;
	Thread thread;

	public Call(Long seq, Object req) {
		this.seq = seq;
		this.req = req;
		this.thread = Thread.currentThread();
	}
}

对调用抽象，每个调用需要绑定线程并阻塞，在服务端处理后，会通过标识符回调通知，唤醒阻塞线程。

public class Client {
	Long seq;
	InputStream inputStream;
	OutputStream outputStream;
	ReentrantLock clock;
	ReentrantLock metux;
	Map calls;

	public Client(InputStream inputStream, OutputStream outputStream) {
		this.inputStream = inputStream;
		this.outputStream = outputStream;
		seq = 0L;
		clock = new ReentrantLock();
		metux = new ReentrantLock();
		calls = new HashMap<>();
		new Thread(() -> {
			receive();
		}).start();
	}
}

客户端应该有标识，对共享资源有互斥锁，对流的读写也需要互斥执行。

FutureTask