Java RPC

使用Java原生的序列化、动态代理、反射等实现简单的RPC框架。本示例来源于《分布式服务框架》。

示例

  • EchoService.java
public interface EchoService {
String echo(String ping);
}
  • EchoServiceImpl.java
public class EchoServiceImpl implements EchoService {
@Override
public String echo(String ping) {
return ping != null ? ping + "--> I am OK." : " I am OK.";
}
}
  • RpcExporter.java
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.lang.reflect.Method;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
/**
* Created by lbd on 2017/11/12.
*/
public class RpcExporter {
static Executor executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
public static void exporter(String hostName, int port) throws Exception {
ServerSocket server = new ServerSocket();
server.bind(new InetSocketAddress(hostName, port));
try {
while (true) {
executor.execute(new ExporterTask(server.accept()));
}
} finally {
server.close();
}
}
private static class ExporterTask implements Runnable {
Socket client = null;
public ExporterTask(Socket client) {
this.client = client;
}
@Override
public void run() {
ObjectInputStream input = null;
ObjectOutputStream output = null;
try {
input = new ObjectInputStream(client.getInputStream());
String interfaceName = input.readUTF();
Class<?> service = Class.forName(interfaceName);
String methodName = input.readUTF();
Class<?>[] parameterTypes = (Class<?>[])input.readObject();
Object[] arguments = (Object[])input.readObject();
Method method = service.getMethod(methodName, parameterTypes);
Object result = method.invoke(service.newInstance(), arguments);
output = new ObjectOutputStream(client.getOutputStream());
output.writeObject(result);
} catch (Exception e) {
e.printStackTrace();
} finally {
if (output != null)
try {
output.close();
} catch (IOException e) {
e.printStackTrace();
}
if (input != null)
try {
input.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
}
  • RpcImporter.java
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.net.InetSocketAddress;
import java.net.Socket;
/**
* Created by lbd on 2017/11/12.
*/
public class RpcImporter<S> {
public S importer(final Class<?> serviceClass, final InetSocketAddress addr) {
return (S) Proxy.newProxyInstance(serviceClass.getClassLoader(), new Class<?>[]{serviceClass.getInterfaces()[0]},
new InvocationHandler() {
@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
Socket socket = null;
ObjectOutputStream output = null;
ObjectInputStream input = null;
try {
socket = new Socket();
socket.connect(addr);
output = new ObjectOutputStream(socket.getOutputStream());
output.writeUTF(serviceClass.getName());
output.writeUTF(method.getName());
output.writeObject(method.getParameterTypes());
output.writeObject(args);
input = new ObjectInputStream(socket.getInputStream());
return input.readObject();
} finally {
if (socket != null)
socket.close();
if (output != null)
output.close();
if (input != null)
input.close();
}
}
});
}
}
  • RpcTest.java
import java.net.InetSocketAddress;
/**
* Created by lbd on 2017/11/12.
*/
public class RpcTest {
public static void main(String[] args) {
new Thread(new Runnable() {
@Override
public void run() {
try {
RpcExporter.exporter("localhost", 7890);
} catch (Exception e) {
e.printStackTrace();
}
}
}).start();
RpcImporter<EchoService> proxyImporter = new RpcImporter<>();
EchoService echoService = proxyImporter.importer(EchoServiceImpl.class, new InetSocketAddress("localhost", 7890));
System.out.println(echoService.echo("Are you ok ?"));
}
}

输出

Are you ok ?--> I am OK.

欢迎关注公众号: FullStackPlan 获取更多干货

Copyright © 2016 - 2017 LBD's Blog All Rights Reserved.

访客数 : | 访问量 :