手寫實現RPC框架（帶註冊中心）
阿新 • • 發佈：2018-12-19
可能這個圖不太準確，但是大體意思就是服務端在註冊中心中註冊服務，客戶端在註冊中心獲取服務地址進行呼叫，中間可能還會有一些LB等：
定義一個註冊服務的頂層介面IRegistryCenter：
package dgb.nospring.myrpc.registry; /** * 註冊中心頂層介面 * @author Dongguabai * @date 2018/11/1 19:05 */ public interface IRegistryCenter { /** * 註冊服務 * @param serviceName 服務名稱 * @param serviceAddress 服務地址 */ void register(String serviceName,String serviceAddress); }
實現類RegistryCenterImpl：
package dgb.nospring.myrpc.registry; import lombok.extern.slf4j.Slf4j; import org.apache.curator.framework.CuratorFramework; import org.apache.curator.framework.CuratorFrameworkFactory; import org.apache.curator.retry.ExponentialBackoffRetry; import org.apache.zookeeper.CreateMode; /** * 註冊中心實現 * * @author Dongguabai * @date 2018/11/1 19:10 */ @Slf4j public class RegistryCenterImpl implements IRegistryCenter { private CuratorFramework curatorFramework; { curatorFramework = CuratorFrameworkFactory.builder() .connectString(RegistryCenterConfig.CONNECTING_STR) .sessionTimeoutMs(RegistryCenterConfig.SESSION_TIMEOUT) .retryPolicy(new ExponentialBackoffRetry(1000, 10)).build(); curatorFramework.start(); } //註冊相應服務 @Override public void register(String serviceName, String serviceAddress) { String serviceNodePath = RegistryCenterConfig.NAMESPACE + "/" + serviceName; try { //如果serviceNodePath（/rpcNode/userService）不存在就建立 if (curatorFramework.checkExists().forPath(serviceNodePath)==null){ //持久化節點 curatorFramework.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(serviceNodePath,RegistryCenterConfig.DEFAULT_VALUE); } //註冊的服務的節點路徑 String addressPath = serviceNodePath+"/"+serviceAddress; //臨時節點 String rsNode = curatorFramework.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(addressPath, RegistryCenterConfig.DEFAULT_VALUE); log.info("服務註冊成功：{}",rsNode); } catch (Exception e) { throw new RuntimeException("註冊服務出現異常！",e); } } }
註冊中心的一些配置引數RegistryCenterConfig：
package dgb.nospring.myrpc.registry; /** * @author Dongguabai * @date 2018/11/1 19:13 */ public interface RegistryCenterConfig { /** * ZK地址int */ String CONNECTING_STR = "192.168.220.136,192.168.220.137"; int SESSION_TIMEOUT = 4000; /** * 註冊中心namespace */ String NAMESPACE = "/rpcNode"; /** * value一般來說作用不大；一般主要是利用節點特性搞點事情 */ byte[] DEFAULT_VALUE = "0".getBytes(); }
為了方便，增加了一個服務釋出的註解RpcAnnotation，在介面的實現類上標明這個註解表示對外發布這個介面：
package dgb.nospring.myrpc.registry;
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
/**
* @author Dongguabai
* @date 2018/11/2 8:54
*/
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
public @interface RpcAnnotation {
/**
* 對外發布服務的介面
*
* @return
*/
Class<?> value();
/**
* 版本，用來區分不同版本
* @return
*/
String version() default "";
}
這個版本號的作用在本次Demo中沒有體現出來，不過其實使用也是很簡單的，可以將版本號與ZK node地址中的serviceName拼接或者繫結起來，然後根據版本號+serviceName獲取相應的服務呼叫地址。那麼客戶端在發現服務的時候也要傳入相應的版本進去。
首先改造服務端，服務端要將服務釋出到註冊中心，RpcServer：
package dgb.nospring.myrpc;
import dgb.nospring.myrpc.registry.IRegistryCenter;
import dgb.nospring.myrpc.registry.RpcAnnotation;
import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
* @author Dongguabai
* @date 2018/11/1 15:53
*/
@Slf4j
public class RpcServer {
/**
* 註冊中心
*/
private IRegistryCenter registryCenter;
/**
* 服務的釋出地址
*/
private String addressService;
/**
* 服務名稱和服務物件之間的關係
*/
private static final Map<String,Object> HANDLER_MAPPING = new HashMap<>();
//不建議通過Executors建立執行緒池，這裡為了方便
private static final ExecutorService executor = Executors.newCachedThreadPool();
/*public void publisher(final Object service, int port) {
//啟動一個服務監聽
try (ServerSocket serverSocket = new ServerSocket(port)) {
while (true){
//通過ServerSocket獲取請求
Socket socket = serverSocket.accept();
executor.execute(new ProcessorHandler(socket,service));
}
} catch (IOException e) {
e.printStackTrace();
}
}*/
/**
* 改造後的釋出服務的方法
*/
public void publisher() {
//啟動一個服務監聽
//獲取埠
int port = Integer.parseInt(addressService.split(":")[1]);
try (ServerSocket serverSocket = new ServerSocket(port)) {
//迴圈獲取所有的介面Name
HANDLER_MAPPING.keySet().forEach(interfaceName->{
registryCenter.register(interfaceName,addressService);
log.info("註冊服務成功：【serviceName：{}，address：{}】",interfaceName,addressService);
});
while (true){
//通過ServerSocket獲取請求
Socket socket = serverSocket.accept();
executor.execute(new ProcessorHandler(socket,HANDLER_MAPPING));
}
} catch (IOException e) {
e.printStackTrace();
}
}
/**
* 繫結服務名稱和服務物件
* @param services
*/
public void bind(Object... services){
for (Object service : services) {
//獲取釋出的服務介面
RpcAnnotation rpcAnnotation = service.getClass().getAnnotation(RpcAnnotation.class);
if (rpcAnnotation==null){
continue;
}
//釋出介面的class
String serviceName = rpcAnnotation.value().getName();
//將serviceName和service進行繫結
HANDLER_MAPPING.put(serviceName,service);
}
}
public RpcServer(IRegistryCenter registryCenter, String addressService) {
this.registryCenter = registryCenter;
this.addressService = addressService;
}
}
改造任務處理類ProcessorHandler：
package dgb.nospring.myrpc;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.net.Socket;
import java.util.Map;
/**
* 任務處理類
*
* @author Dongguabai
* @date 2018/11/1 16:10
*/
public class ProcessorHandler implements Runnable {
private Socket socket;
/**
* 服務端釋出的服務
*/
private Map<String,Object> handlerMap;
/**
* 通過構造傳入Map
* @param socket
* @param handlerMap
*/
public ProcessorHandler(Socket socket, Map<String, Object> handlerMap) {
this.socket = socket;
this.handlerMap = handlerMap;
}
//處理請求
@Override
public void run() {
ObjectInputStream objectInputStream = null;
try {
objectInputStream = new ObjectInputStream(socket.getInputStream());
//反序列化
RpcRequest rpcRequest = (RpcRequest) objectInputStream.readObject();
Object result = invoke(rpcRequest);
//將結果返回給客戶端
ObjectOutputStream objectOutputStream = new ObjectOutputStream(socket.getOutputStream());
objectOutputStream.writeObject(result);
objectOutputStream.flush();
objectInputStream.close();
} catch (Exception e) {
e.printStackTrace();
} finally {
if (objectInputStream != null) {
try {
objectInputStream.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
/**
* 反射呼叫
*
* @param rpcRequest
*/
/*private Object invoke(RpcRequest rpcRequest) throws NoSuchMethodException, InvocationTargetException, IllegalAccessException {
System.out.println("服務端開始呼叫------");
Object[] parameters = rpcRequest.getParameters();
Class[] parameterTypes = new Class[parameters.length];
for (int i = 0, length = parameters.length; i < length; i++) {
parameterTypes[i] = parameters[i].getClass();
}
Method method = service.getClass().getMethod(rpcRequest.getMethodName(), parameterTypes);
return method.invoke(service, parameters);
}*/
/**
* 反射呼叫（之前通過Service進行反射呼叫，現在通過Map獲取service）
*
* @param rpcRequest
*/
private Object invoke(RpcRequest rpcRequest) throws NoSuchMethodException, InvocationTargetException, IllegalAccessException {
System.out.println("服務端開始呼叫------");
Object[] parameters = rpcRequest.getParameters();
Class[] parameterTypes = new Class[parameters.length];
for (int i = 0, length = parameters.length; i < length; i++) {
parameterTypes[i] = parameters[i].getClass();
}
//從Map中獲得Service（根據客戶端請求的ServiceName，獲得相應的服務），依舊是通過反射發起呼叫
Object service = handlerMap.get(rpcRequest.getClassName());
Method method = service.getClass().getMethod(rpcRequest.getMethodName(), parameterTypes);
return method.invoke(service, parameters);
}
}
測試服務釋出Demo：
package dgb.nospring.myrpc.demo;
import dgb.nospring.myrpc.RpcServer;
import dgb.nospring.myrpc.registry.IRegistryCenter;
import dgb.nospring.myrpc.registry.RegistryCenterImpl;
/**
* @author Dongguabai
* @date 2018/11/1 18:07
*/
public class ServerDemo {
public static void main(String[] args) {
//之前釋出服務
/*
RpcServer rpcServer = new RpcServer();
rpcServer.publisher(new HelloServiceImpl(),12345);
*/
//改造後
IRegistryCenter registryCenter = new RegistryCenterImpl();
//這裡為了方便，獲取ip地址就直接寫了
RpcServer rpcServer = new RpcServer(registryCenter,"127.0.0.1:12345");
//繫結服務
rpcServer.bind(new HelloServiceImpl());
rpcServer.publisher();
}
}
執行結果：
在ZK客戶端：
服務客戶釋出後，現在要解決的就是服務發現的問題。
定義一個頂層服務發現介面IServiceDiscovery：
package dgb.nospring.myrpc.registry;
/**
* @author Dongguabai
* @date 2018/11/2 9:55
*/
public interface IServiceDiscovery {
/**
* 根據介面名稱發現服務呼叫地址
* @param serviceName
* @return
*/
String discover(String serviceName);
}
實現類：
package dgb.nospring.myrpc.registry;
import dgb.nospring.myrpc.registry.loadbalance.LoadBalance;
import dgb.nospring.myrpc.registry.loadbalance.RandomLoadBanalce;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import java.util.List;
/**
* 服務發現實現類
* @author Dongguabai
* @date 2018/11/2 9:56
*/
@Slf4j
public class ServiceDiscoveryImpl implements IServiceDiscovery {
/**
* /rpcNode/dgb.nospring.myrpc.demo.IHelloService
* 當前服務下所有的協議地址
*/
private List<String> repos;
/**
* ZK地址
*/
private String zkAddress;
private CuratorFramework curatorFramework;
@Override
public String discover(String serviceName) {
//獲取/rpcNode/dgb.nospring.myrpc.demo.IHelloService下所有協議地址
String nodePath = RegistryCenterConfig.NAMESPACE+"/"+serviceName;
try {
repos = curatorFramework.getChildren().forPath(nodePath);
} catch (Exception e) {
throw new RuntimeException("服務發現獲取子節點異常！",e);
}
//動態發現服務節點變化，需要註冊監聽
registerWatcher(nodePath);
//這裡為了方便，直接使用隨機負載
LoadBalance loadBalance = new RandomLoadBanalce();
return loadBalance.selectHost(repos);
}
/**
* 監聽節點變化，給repos重新賦值
* @param path
*/
private void registerWatcher(String path){
PathChildrenCache pathChildrenCache = new PathChildrenCache(curatorFramework,path,true);
PathChildrenCacheListener pathChildrenCacheListener = new PathChildrenCacheListener() {
@Override
public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
repos = curatorFramework.getChildren().forPath(path);
}
};
pathChildrenCache.getListenable().addListener(pathChildrenCacheListener);
try {
pathChildrenCache.start();
} catch (Exception e) {
throw new RuntimeException("監聽節點變化異常！",e);
}
}
public ServiceDiscoveryImpl(String zkAddress) {
this.zkAddress = zkAddress;
curatorFramework = CuratorFrameworkFactory.builder()
.connectString(RegistryCenterConfig.CONNECTING_STR)
.sessionTimeoutMs(RegistryCenterConfig.SESSION_TIMEOUT)
.retryPolicy(new ExponentialBackoffRetry(1000, 10)).build();
curatorFramework.start();
}
}
還有一套負載演算法（這裡簡單實現了一個隨機負載）：
package dgb.nospring.myrpc.registry.loadbalance;
import java.util.List;
/**
* 負載頂層介面
* @author Dongguabai
* @date 2018/11/2 10:11
*/
public interface LoadBalance {
String selectHost(List<String> repos);
}
package dgb.nospring.myrpc.registry.loadbalance;
import org.apache.commons.collections.CollectionUtils;
import java.util.List;
/**
* @author Dongguabai
* @date 2018/11/2 10:15
*/
public abstract class AbstractLoadBanance implements LoadBalance{
/**
* 通過模板方法，做一些牽制操作
* @param repos
* @return
*/
@Override
public String selectHost(List<String> repos) {
if(CollectionUtils.isEmpty(repos)){
return null;
}
if(repos.size()==1){
return repos.get(0);
}
return doSelect(repos);
}
/**
* 實現具體的實現負載演算法
* @param repos
* @return
*/
protected abstract String doSelect(List<String> repos);
}
package dgb.nospring.myrpc.registry.loadbalance;
import java.util.List;
import java.util.Random;
/**
* 隨機負載演算法
* @author Dongguabai
* @date 2018/11/2 10:17
*/
public class RandomLoadBanalce extends AbstractLoadBanance{
@Override
protected String doSelect(List<String> repos) {
return repos.get(new Random().nextInt(repos.size()));
}
}
還有獲取服務的RpcClientProxy需要進行改造，其實就是改了一個引數傳遞而已：
package dgb.nospring.myrpc;
import dgb.nospring.myrpc.registry.IServiceDiscovery;
import java.lang.reflect.Proxy;
/**
* 客戶端代理
* @author Dongguabai
* @date 2018/11/1 16:18
*/
public class RpcClientProxy {
private IServiceDiscovery serviceDiscovery;
/* public <T> T clientProxy(final Class<T> interfaceClass,final String host,final int port){
return (T) Proxy.newProxyInstance(interfaceClass.getClassLoader(),new Class[]{interfaceClass},new RemoteInvocationHandler(host, port));
}*/
public <T> T clientProxy(final Class<T> interfaceClass){
return (T) Proxy.newProxyInstance(interfaceClass.getClassLoader(),new Class[]{interfaceClass},new RemoteInvocationHandler(serviceDiscovery));
}
public RpcClientProxy(IServiceDiscovery serviceDiscovery) {
this.serviceDiscovery = serviceDiscovery;
}
}
同樣的，動態代理的InvocationHandler也要修改：
package dgb.nospring.myrpc;
import dgb.nospring.myrpc.registry.IServiceDiscovery;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
/**
* @author Dongguabai
* @date 2018/11/1 16:20
*/
public class RemoteInvocationHandler implements InvocationHandler{
private IServiceDiscovery serviceDiscovery;
/**
*發起客戶端和服務端的遠端呼叫。呼叫客戶端的資訊進行傳輸
*/
@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
RpcRequest rpcRequest = new RpcRequest();
rpcRequest.setClassName(method.getDeclaringClass().getName());
rpcRequest.setMethodName(method.getName());
rpcRequest.setParameters(args);
//從ZK中獲取地址 127.0.0.1:12345
String discover = serviceDiscovery.discover(rpcRequest.getClassName());
TcpTransport tcpTransport = new TcpTransport(discover);
return tcpTransport.send(rpcRequest);
}
public RemoteInvocationHandler(IServiceDiscovery serviceDiscovery) {
this.serviceDiscovery = serviceDiscovery;
}
}
同樣的。TCPTransport也要進行改造：
package dgb.nospring.myrpc;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.Socket;
/**
* socket傳輸
*
* @author Dongguabai
* @date 2018/11/1 16:25
*/
public class TcpTransport {
private String serviceAddress;
private Socket newSocket() {
System.out.println("準備建立Socket連線，"+serviceAddress);
String[] split = serviceAddress.split(":");
try {
Socket socket = new Socket(split[0], Integer.parseInt(split[1]));
return socket;
} catch (IOException e) {
throw new RuntimeException("Socket連線建立失敗！" + serviceAddress);
}
}
public TcpTransport(String serviceAddress) {
this.serviceAddress = serviceAddress;
}
public Object send(RpcRequest rpcRequest) {
Socket socket = null;
try {
socket = newSocket();
try {
ObjectOutputStream outputStream = new ObjectOutputStream(socket.getOutputStream());
outputStream.writeObject(rpcRequest);
outputStream.flush();
ObjectInputStream inputStream = new ObjectInputStream(socket.getInputStream());
Object result = inputStream.readObject();
inputStream.close();
outputStream.close();
return result;
} catch (IOException | ClassNotFoundException e) {
throw new RuntimeException("發起遠端呼叫異常！",e);
}
} finally {
if (socket != null) {
try {
socket.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
}
客戶端Demo:
package dgb.nospring.myrpc.demo;
import dgb.nospring.myrpc.RpcClientProxy;
import dgb.nospring.myrpc.registry.IServiceDiscovery;
import dgb.nospring.myrpc.registry.RegistryCenterConfig;
import dgb.nospring.myrpc.registry.ServiceDiscoveryImpl;
/**
* @author Dongguabai
* @date 2018/11/1 18:10
*/
public class ClientDemo {
public static void main(String[] args) {
/*RpcClientProxy proxy = new RpcClientProxy();
IHelloService helloService = proxy.clientProxy(IHelloService.class, "127.0.0.1", 12345);
String name = helloService.sayHello("張三");
System.out.println(name);*/
IServiceDiscovery serviceDiscovery = new ServiceDiscoveryImpl(RegistryCenterConfig.CONNECTING_STR);
RpcClientProxy proxy = new RpcClientProxy(serviceDiscovery);
IHelloService service = proxy.clientProxy(IHelloService.class);
System.out.println(service.sayHello("張三"));
}
}
控制檯輸出：
如果需要驗證叢集環境下，我們可以建立兩個ServerDemo：
兩個服務均註冊到註冊中心：
客戶端呼叫還是不變：
連續呼叫兩次客戶端：
