小司机带你撸一个简单的RPC框架

随着业务的增长，有时候普通的单一型架构不再能满足我们的需求，这就诞生了RPC框架，经过多年的发展，我们可以看到市面上可用性高的开源RPC框架还是比较多的，比如说：Hessian，Dubbo等，这些框架有的支持垮语言，有的只支持特定语言，但不管怎么样，其核心架构都是大同小异。今天小编就带各位小伙伴撸一个简单的RPC

首先是客户端的代码:

package com.minsx.selfrpcb.client;

import com.minsx.selfrpcb.service.RpcServiceRepository;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.stereotype.Component;

import java.io.*;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.net.Socket;

/**
* RpcInvocationHandler
* Created by Joker on 2017/9/18.
*/
@Component
public class ClientService implements InvocationHandler {

private final static String SERVER_ADDRESS = "127.0.0.1";
private final static Integer SERVER_PORT = 8693;
Class clazz = null;

private final Log LOGGER = LogFactory.getLog(this.getClass());

public <T> T getServiceImpl(Class<T> clazz) {
this.clazz = clazz;
return (T) Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class[]{clazz}, this);
}

@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
LOGGER.info("start send method and args to server");
Socket socket = null;
ObjectOutputStream out = null;
ObjectInputStream in = null;
try {
socket = new Socket(SERVER_ADDRESS, SERVER_PORT);
LOGGER.info("opened client socket");
out = new ObjectOutputStream(socket.getOutputStream());
out.writeUTF(clazz.getName());
out.writeUTF(method.getName());
out.writeObject(method.getParameterTypes());
out.writeObject(args);
in = new ObjectInputStream(socket.getInputStream());
} catch (Exception e) {
if (socket != null) socket.close();
if (out != null) out.close();
if (in != null) in.close();
}
LOGGER.info("receive result from server");
if (in==null) {
return null;
}else{
return in.readObject();
}
}
}


package com.minsx.selfrpcb.client;

import com.minsx.selfrpcb.service.RpcServiceRepository;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

/**
* Main
* Created by Joker on 2017/9/18.
*/
@RunWith(SpringJUnit4ClassRunner.class)
@Configuration
@ComponentScan(basePackages = "com.minsx.selfrpcb.client")
@ContextConfiguration(classes = com.minsx.selfrpcb.client.Main.class)
public class Main {

@Autowired
ClientService clientService;

@Test
public void openClient() {
RpcServiceRepository rpcServiceRepository = clientService.getServiceImpl(RpcServiceRepository.class);
System.out.println(String.format("name=%s",rpcServiceRepository.getUserNameById(2)));
System.out.println(String.format("2+3=%s",rpcServiceRepository.add(2,3)));
}

}


服务端代码：
package com.minsx.selfrpcb.server;

import com.minsx.selfrpcb.service.RpcServiceRepository;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
* ServerService
* Created by Joker on 2017/9/19.
*/
@Service
public class ServerService {

private final static ExecutorService EXECUTOR = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

private final static Integer PORT = 8693;

private final Log LOGGER = LogFactory.getLog(this.getClass());

public void start() throws IOException {
ServerSocket server = new ServerSocket();
server.bind(new InetSocketAddress(PORT));
ServiceTask.SERVICE_REGISITRY.put(RpcServiceRepository.class.getName(),SimpleRpcServiceImpl.class);
LOGGER.info("start server");
try {
while (true) {
EXECUTOR.execute(new ServiceTask(server.accept()));
}
} finally {
server.close();
}
}

}

package com.minsx.selfrpcb.server;

import com.minsx.selfrpcb.service.RpcServiceRepository;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.stereotype.Service;

import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.lang.reflect.Method;
import java.net.Socket;
import java.util.HashMap;

/**
* ServiceTask
* Created by Joker on 2017/9/19.
*/
public class ServiceTask implements Runnable {

private Socket socket = null;
public static final HashMap<String, Class> SERVICE_REGISITRY = new HashMap<String, Class>();
private final Log LOGGER = LogFactory.getLog(this.getClass());

public ServiceTask(Socket socket) {
LOGGER.info("receive message from client");
this.socket = socket;
}

@Override
public void run() {
ObjectInputStream input = null;
ObjectOutputStream output = null;
try {
input = new ObjectInputStream(socket.getInputStream());
String serviceName = input.readUTF();
String methodName = input.readUTF();
Class<?>[] parameterTypes = (Class<?>[]) input.readObject();
Object[] arguments = (Object[]) input.readObject();
Class serviceClass = SERVICE_REGISITRY.get(serviceName);
if (serviceClass == null) {
throw new ClassNotFoundException(serviceName + " not found");
}
Method method = serviceClass.getMethod(methodName, parameterTypes);
LOGGER.info("start execute method : "+methodName);
Object result = method.invoke(serviceClass.newInstance(), arguments);
output = new ObjectOutputStream(socket.getOutputStream());
output.writeObject(result);
LOGGER.info("method : "+methodName+" executed over,start return result to client");
} catch (Exception e) {
e.printStackTrace();
}

}

}

package com.minsx.selfrpcb.server;

import com.minsx.selfrpcb.service.RpcServiceRepository;

/**
* SimpleRpcServiceImpl
* Created by Joker on 2017/9/18.
*/
public class SimpleRpcServiceImpl implements RpcServiceRepository {
@Override
public String getUserNameById(Integer id) {
if (id==1) {
return "小明";
}else if(id==2){
return "小强";
}else{
return "小红";
}
}

@Override
public Integer add(Integer A, Integer B) {
return A+B;
}
}

package com.minsx.selfrpcb.server;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

import java.io.IOException;

/**
* Main
* Created by Joker on 2017/9/19.
*/
@RunWith(SpringJUnit4ClassRunner.class)
@Configuration
@ComponentScan(basePackages = "com.minsx.selfrpcb.server")
@ContextConfiguration(classes = com.minsx.selfrpcb.server.Main.class)
public class Main {

@Autowired
ServerService serverService;

@Test
public void openServer() throws IOException {
serverService.start();
}

}

最后是客户端与服务端公用接口：
package com.minsx.selfrpcb.service;

/**
* RpcServiceRepository
* Created by Joker on 2017/9/18.
*/
public interface RpcServiceRepository {

public String getUserNameById(Integer id);

public Integer add(Integer A,Integer B);

}


最后来看一下两端的输出：

客户端输出:

[2017-09-20
13:38:58]-[main- INFO]-[com.minsx.selfrpcb.client.ClientService-invoke(34)] start send method and args to server

[2017-09-20 13:38:58]-[main- INFO]-[com.minsx.selfrpcb.client.ClientService-invoke(40)] opened client socket

[2017-09-20 13:38:58]-[main- INFO]-[com.minsx.selfrpcb.client.ClientService-invoke(52)] receive result from server

name=小强

[2017-09-20 13:38:58]-[main- INFO]-[com.minsx.selfrpcb.client.ClientService-invoke(34)] start send method and args to server

[2017-09-20 13:38:58]-[main- INFO]-[com.minsx.selfrpcb.client.ClientService-invoke(40)] opened client socket

[2017-09-20 13:38:58]-[main- INFO]-[com.minsx.selfrpcb.client.ClientService-invoke(52)] receive result from server

2+3=5

服务端输出:

[2017-09-20
13:38:58]-[pool-1-thread-4- INFO]-[com.minsx.selfrpcb.server.ServiceTask-run(44)] start execute method : getUserNameById

[2017-09-20 13:38:58]-[pool-1-thread-4- INFO]-[com.minsx.selfrpcb.server.ServiceTask-run(48)] method : getUserNameById executed over,start return result to client

[2017-09-20 13:38:58]-[main- INFO]-[com.minsx.selfrpcb.server.ServiceTask-<init>(25)] receive message from client

[2017-09-20 13:38:58]-[pool-1-thread-1- INFO]-[com.minsx.selfrpcb.server.ServiceTask-run(44)] start execute method : add

[2017-09-20 13:38:58]-[pool-1-thread-1- INFO]-[com.minsx.selfrpcb.server.ServiceTask-run(48)] method : add executed over,start return result to client

当然，这里为了简便，小编把这个公用接口放在了同一包下，在开发过程中我们不推荐这种用法，建议接口部分独立打包，通过依赖的方式引入进来。值得一提的是正式开发中客户端与服务端的连接需要保持长连接与数据校对，此处我们不在阐述。