RPC，即 Remote Procedure Call（远程过程调用），说得通俗一点就是：调用远程计算机上的服务，就像调用本地服务一样。

RPC 可基于 HTTP 或 TCP 协议，Web Service 就是基于 HTTP 协议的 RPC，它具有良好的跨平台性，但其性能却不如基于 TCP 协议的 RPC。会两方面会直接影响 RPC 的性能，一是传输方式，二是序列化。

轻量级分布式 RPC 框架的具体过程，该框架基于 TCP 协议

Spring：它是最强大的依赖注入框架，也是业界的权威标准。
Netty：它使 NIO 编程更加容易，屏蔽了 Java 底层的 NIO 细节。
Protostuff：它基于 Protobuf 序列化框架，面向 POJO，无需编写 .proto 文件。
ZooKeeper：提供服务注册与发现功能，开发分布式系统的必备选择，同时它也具备天生的集群能力。

第一步：编写服务接口

<!-- lang: java -->
public interface HelloService {String hello(String name);
}

第二步：编写服务接口的实现类

<!-- lang: java -->
@RpcService(HelloService.class) // 指定远程接口//使用RpcService注解定义在服务接口的实现类上，需要对该实现类指定远程接口，因为实现类可能会实现多个接口，一定要告诉框架哪个才是远程接口。
public class HelloServiceImpl implements HelloService {@Overridepublic String hello(String name) {return "Hello! " + name;}
}

<!-- lang: java -->
@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Component // 表明可被 Spring 扫描
public @interface RpcService {Class<?> value();
}

第三步：配置服务端

服务端 Spring 配置文件名为spring.xml，内容如下：

<!-- lang: xml -->
<beans ...><context:component-scan base-package="com.xxx.rpc.sample.server"/><context:property-placeholder location="classpath:config.properties"/><!-- 配置服务注册组件 --><bean id="serviceRegistry" class="com.xxx.rpc.registry.ServiceRegistry"><constructor-arg name="registryAddress" value="${registry.address}"/></bean><!-- 配置 RPC 服务器 --><bean id="rpcServer" class="com.xxx.rpc.server.RpcServer"><constructor-arg name="serverAddress" value="${server.address}"/><constructor-arg name="serviceRegistry" ref="serviceRegistry"/></bean>
</beans>

具体的配置参数在config.properties文件中，连接本地的 ZooKeeper 服务器，并在 8000 端口上发布 RPC 服务。

<!-- lang: java -->
# ZooKeeper 服务器
registry.address=127.0.0.1:2181# RPC 服务器
server.address=127.0.0.1:8000

第四步：启动服务器并发布服务

为了加载 Spring 配置文件来发布服务，只需编写一个引导程序即可：

<!-- lang: java -->
public class RpcBootstrap {public static void main(String[] args) {new ClassPathXmlApplicationContext("spring.xml");}
}

运行RpcBootstrap类的main方法即可启动服务端，但还有两个重要的组件尚未实现，它们分别是：ServiceRegistry与RpcServer

第五步：实现服务注册

使用 ZooKeeper 客户端可轻松实现服务注册功能，ServiceRegistry代码如下：

public class ServiceRegistry {private static final Logger LOGGER = LoggerFactory.getLogger(ServiceRegistry.class);private CountDownLatch latch = new CountDownLatch(1);private String registryAddress;public ServiceRegistry(String registryAddress) {this.registryAddress = registryAddress;}public void register(String data) {if (data != null) {ZooKeeper zk = connectServer();if (zk != null) {createNode(zk, data);}}}private ZooKeeper connectServer() {ZooKeeper zk = null;try {zk = new ZooKeeper(registryAddress, Constant.ZK_SESSION_TIMEOUT, new Watcher() {@Overridepublic void process(WatchedEvent event) {if (event.getState() == Event.KeeperState.SyncConnected) {latch.countDown();}}});latch.await();} catch (IOException | InterruptedException e) {LOGGER.error("", e);}return zk;}private void createNode(ZooKeeper zk, String data) {try {byte[] bytes = data.getBytes();String path = zk.create(Constant.ZK_DATA_PATH, bytes, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);LOGGER.debug("create zookeeper node ({} => {})", path, data);} catch (KeeperException | InterruptedException e) {LOGGER.error("", e);}}
}

其中，通过Constant配置了所有的常量：

public interface Constant {int ZK_SESSION_TIMEOUT = 5000;String ZK_REGISTRY_PATH = "/registry";String ZK_DATA_PATH = ZK_REGISTRY_PATH + "/data";
}

首先需要使用 ZooKeeper 客户端命令行创建/registry永久节点，用于存放所有的服务临时节点

第六步：实现 RPC 服务器

使用 Netty 可实现一个支持 NIO 的 RPC 服务器，需要使用ServiceRegistry注册服务地址，RpcServer代码如下：

public class RpcServer implements ApplicationContextAware, InitializingBean {private static final Logger LOGGER = LoggerFactory.getLogger(RpcServer.class);private String serverAddress;private ServiceRegistry serviceRegistry;private Map<String, Object> handlerMap = new HashMap<>(); // 存放接口名与服务对象之间的映射关系public RpcServer(String serverAddress) {this.serverAddress = serverAddress;}public RpcServer(String serverAddress, ServiceRegistry serviceRegistry) {this.serverAddress = serverAddress;this.serviceRegistry = serviceRegistry;}@Overridepublic void setApplicationContext(ApplicationContext ctx) throws BeansException {Map<String, Object> serviceBeanMap = ctx.getBeansWithAnnotation(RpcService.class); // 获取所有带有 RpcService 注解的 Spring Beanif (MapUtils.isNotEmpty(serviceBeanMap)) {for (Object serviceBean : serviceBeanMap.values()) {String interfaceName = serviceBean.getClass().getAnnotation(RpcService.class).value().getName();handlerMap.put(interfaceName, serviceBean);}}}@Overridepublic void afterPropertiesSet() throws Exception {EventLoopGroup bossGroup = new NioEventLoopGroup();EventLoopGroup workerGroup = new NioEventLoopGroup();try {ServerBootstrap bootstrap = new ServerBootstrap();bootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<SocketChannel>() {@Overridepublic void initChannel(SocketChannel channel) throws Exception {channel.pipeline().addLast(new RpcDecoder(RpcRequest.class)) // 将 RPC 请求进行解码（为了处理请求）.addLast(new RpcEncoder(RpcResponse.class)) // 将 RPC 响应进行编码（为了返回响应）.addLast(new RpcHandler(handlerMap)); // 处理 RPC 请求
                    }}).option(ChannelOption.SO_BACKLOG, 128).childOption(ChannelOption.SO_KEEPALIVE, true);String[] array = serverAddress.split(":");String host = array[0];int port = Integer.parseInt(array[1]);ChannelFuture future = bootstrap.bind(host, port).sync();LOGGER.debug("server started on port {}", port);if (serviceRegistry != null) {serviceRegistry.register(serverAddress); // 注册服务地址
            }future.channel().closeFuture().sync();} finally {workerGroup.shutdownGracefully();bossGroup.shutdownGracefully();}}
}

以上代码中，有两个重要的 POJO 需要描述一下，它们分别是RpcRequest与RpcResponse。

使用RpcRequest封装 RPC 请求，代码如下：

public class RpcRequest {private String requestId;private String className;private String methodName;private Class<?>[] parameterTypes;private Object[] parameters;// getter/setter...
}

使用RpcResponse封装 RPC 响应，代码如下：

public class RpcResponse {private String requestId;private Throwable error;private Object result;// getter/setter...
}

使用RpcDecoder提供 RPC 解码，只需扩展 Netty 的ByteToMessageDecoder抽象类的decode方法即可，代码如下：

public class RpcDecoder extends ByteToMessageDecoder {private Class<?> genericClass;public RpcDecoder(Class<?> genericClass) {this.genericClass = genericClass;}@Overridepublic void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {if (in.readableBytes() < 4) {return;}in.markReaderIndex();int dataLength = in.readInt();if (dataLength < 0) {ctx.close();}if (in.readableBytes() < dataLength) {in.resetReaderIndex();return;}byte[] data = new byte[dataLength];in.readBytes(data);Object obj = SerializationUtil.deserialize(data, genericClass);out.add(obj);}
}

使用RpcEncoder提供 RPC 编码，只需扩展 Netty 的MessageToByteEncoder抽象类的encode方法即可，代码如下：

public class RpcEncoder extends MessageToByteEncoder {private Class<?> genericClass;public RpcEncoder(Class<?> genericClass) {this.genericClass = genericClass;}@Overridepublic void encode(ChannelHandlerContext ctx, Object in, ByteBuf out) throws Exception {if (genericClass.isInstance(in)) {byte[] data = SerializationUtil.serialize(in);out.writeInt(data.length);out.writeBytes(data);}}
}

编写一个SerializationUtil工具类，使用Protostuff实现序列化：

public class SerializationUtil {private static Map<Class<?>, Schema<?>> cachedSchema = new ConcurrentHashMap<>();private static Objenesis objenesis = new ObjenesisStd(true);private SerializationUtil() {}@SuppressWarnings("unchecked")private static <T> Schema<T> getSchema(Class<T> cls) {Schema<T> schema = (Schema<T>) cachedSchema.get(cls);if (schema == null) {schema = RuntimeSchema.createFrom(cls);if (schema != null) {cachedSchema.put(cls, schema);}}return schema;}@SuppressWarnings("unchecked")public static <T> byte[] serialize(T obj) {Class<T> cls = (Class<T>) obj.getClass();LinkedBuffer buffer = LinkedBuffer.allocate(LinkedBuffer.DEFAULT_BUFFER_SIZE);try {Schema<T> schema = getSchema(cls);return ProtostuffIOUtil.toByteArray(obj, schema, buffer);} catch (Exception e) {throw new IllegalStateException(e.getMessage(), e);} finally {buffer.clear();}}public static <T> T deserialize(byte[] data, Class<T> cls) {try {T message = (T) objenesis.newInstance(cls);Schema<T> schema = getSchema(cls);ProtostuffIOUtil.mergeFrom(data, message, schema);return message;} catch (Exception e) {throw new IllegalStateException(e.getMessage(), e);}}
}

以上了使用 Objenesis 来实例化对象，它是比 Java 反射更加强大。

注意：如需要替换其它序列化框架，只需修改SerializationUtil即可。当然，更好的实现方式是提供配置项来决定使用哪种序列化方式。

使用RpcHandler中处理 RPC 请求，只需扩展 Netty 的SimpleChannelInboundHandler抽象类即可，代码如下：

public class RpcHandler extends SimpleChannelInboundHandler<RpcRequest> {private static final Logger LOGGER = LoggerFactory.getLogger(RpcHandler.class);private final Map<String, Object> handlerMap;public RpcHandler(Map<String, Object> handlerMap) {this.handlerMap = handlerMap;}@Overridepublic void channelRead0(final ChannelHandlerContext ctx, RpcRequest request) throws Exception {RpcResponse response = new RpcResponse();response.setRequestId(request.getRequestId());try {Object result = handle(request);response.setResult(result);} catch (Throwable t) {response.setError(t);}ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);}private Object handle(RpcRequest request) throws Throwable {String className = request.getClassName();Object serviceBean = handlerMap.get(className);Class<?> serviceClass = serviceBean.getClass();String methodName = request.getMethodName();Class<?>[] parameterTypes = request.getParameterTypes();Object[] parameters = request.getParameters();/*Method method = serviceClass.getMethod(methodName, parameterTypes);method.setAccessible(true);return method.invoke(serviceBean, parameters);*/FastClass serviceFastClass = FastClass.create(serviceClass);FastMethod serviceFastMethod = serviceFastClass.getMethod(methodName, parameterTypes);return serviceFastMethod.invoke(serviceBean, parameters);}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {LOGGER.error("server caught exception", cause);ctx.close();}
}

为了避免使用 Java 反射带来的性能问题，我们可以使用 CGLib 提供的反射 API，如上面用到的FastClass与FastMethod。

第七步：配置客户端

同样使用 Spring 配置文件来配置 RPC 客户端，spring.xml代码如下：

<!-- lang: java -->
<beans ...><context:property-placeholder location="classpath:config.properties"/><!-- 配置服务发现组件 --><bean id="serviceDiscovery" class="com.xxx.rpc.registry.ServiceDiscovery"><constructor-arg name="registryAddress" value="${registry.address}"/></bean><!-- 配置 RPC 代理 --><bean id="rpcProxy" class="com.xxx.rpc.client.RpcProxy"><constructor-arg name="serviceDiscovery" ref="serviceDiscovery"/></bean>
</beans>

<!-- lang: java -->
# ZooKeeper 服务器
registry.address=127.0.0.1:2181

第八步：实现服务发现

同样使用 ZooKeeper 实现服务发现功能，见如下代码：

public class ServiceDiscovery {private static final Logger LOGGER = LoggerFactory.getLogger(ServiceDiscovery.class);private CountDownLatch latch = new CountDownLatch(1);private volatile List<String> dataList = new ArrayList<>();private String registryAddress;public ServiceDiscovery(String registryAddress) {this.registryAddress = registryAddress;ZooKeeper zk = connectServer();if (zk != null) {watchNode(zk);}}public String discover() {String data = null;int size = dataList.size();if (size > 0) {if (size == 1) {data = dataList.get(0);LOGGER.debug("using only data: {}", data);} else {data = dataList.get(ThreadLocalRandom.current().nextInt(size));LOGGER.debug("using random data: {}", data);}}return data;}private ZooKeeper connectServer() {ZooKeeper zk = null;try {zk = new ZooKeeper(registryAddress, Constant.ZK_SESSION_TIMEOUT, new Watcher() {@Overridepublic void process(WatchedEvent event) {if (event.getState() == Event.KeeperState.SyncConnected) {latch.countDown();}}});latch.await();} catch (IOException | InterruptedException e) {LOGGER.error("", e);}return zk;}private void watchNode(final ZooKeeper zk) {try {List<String> nodeList = zk.getChildren(Constant.ZK_REGISTRY_PATH, new Watcher() {@Overridepublic void process(WatchedEvent event) {if (event.getType() == Event.EventType.NodeChildrenChanged) {watchNode(zk);}}});List<String> dataList = new ArrayList<>();for (String node : nodeList) {byte[] bytes = zk.getData(Constant.ZK_REGISTRY_PATH + "/" + node, false, null);dataList.add(new String(bytes));}LOGGER.debug("node data: {}", dataList);this.dataList = dataList;} catch (KeeperException | InterruptedException e) {LOGGER.error("", e);}}
}

第九步：实现 RPC 代理

这里使用 Java 提供的动态代理技术实现 RPC 代理（当然也可以使用 CGLib 来实现），具体代码如下：

public class RpcProxy {private String serverAddress;private ServiceDiscovery serviceDiscovery;public RpcProxy(String serverAddress) {this.serverAddress = serverAddress;}public RpcProxy(ServiceDiscovery serviceDiscovery) {this.serviceDiscovery = serviceDiscovery;}@SuppressWarnings("unchecked")public <T> T create(Class<?> interfaceClass) {return (T) Proxy.newProxyInstance(interfaceClass.getClassLoader(),new Class<?>[]{interfaceClass},new InvocationHandler() {@Overridepublic Object invoke(Object proxy, Method method, Object[] args) throws Throwable {RpcRequest request = new RpcRequest(); // 创建并初始化 RPC 请求
                    request.setRequestId(UUID.randomUUID().toString());request.setClassName(method.getDeclaringClass().getName());request.setMethodName(method.getName());request.setParameterTypes(method.getParameterTypes());request.setParameters(args);if (serviceDiscovery != null) {serverAddress = serviceDiscovery.discover(); // 发现服务
                    }String[] array = serverAddress.split(":");String host = array[0];int port = Integer.parseInt(array[1]);RpcClient client = new RpcClient(host, port); // 初始化 RPC 客户端RpcResponse response = client.send(request); // 通过 RPC 客户端发送 RPC 请求并获取 RPC 响应if (response.isError()) {throw response.getError();} else {return response.getResult();}}});}
}

使用RpcClient类实现 RPC 客户端，只需扩展 Netty 提供的SimpleChannelInboundHandler抽象类即可，代码如下：

public class RpcClient extends SimpleChannelInboundHandler<RpcResponse> {private static final Logger LOGGER = LoggerFactory.getLogger(RpcClient.class);private String host;private int port;private RpcResponse response;private final Object obj = new Object();public RpcClient(String host, int port) {this.host = host;this.port = port;}@Overridepublic void channelRead0(ChannelHandlerContext ctx, RpcResponse response) throws Exception {this.response = response;synchronized (obj) {obj.notifyAll(); // 收到响应，唤醒线程
        }}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {LOGGER.error("client caught exception", cause);ctx.close();}public RpcResponse send(RpcRequest request) throws Exception {EventLoopGroup group = new NioEventLoopGroup();try {Bootstrap bootstrap = new Bootstrap();bootstrap.group(group).channel(NioSocketChannel.class).handler(new ChannelInitializer<SocketChannel>() {@Overridepublic void initChannel(SocketChannel channel) throws Exception {channel.pipeline().addLast(new RpcEncoder(RpcRequest.class)) // 将 RPC 请求进行编码（为了发送请求）.addLast(new RpcDecoder(RpcResponse.class)) // 将 RPC 响应进行解码（为了处理响应）.addLast(RpcClient.this); // 使用 RpcClient 发送 RPC 请求
                    }}).option(ChannelOption.SO_KEEPALIVE, true);ChannelFuture future = bootstrap.connect(host, port).sync();future.channel().writeAndFlush(request).sync();synchronized (obj) {obj.wait(); // 未收到响应，使线程等待
            }if (response != null) {future.channel().closeFuture().sync();}return response;} finally {group.shutdownGracefully();}}
}

第十步：发送 RPC 请求

使用 JUnit 结合 Spring 编写一个单元测试，代码如下：

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:spring.xml")
public class HelloServiceTest {@Autowiredprivate RpcProxy rpcProxy;@Testpublic void helloTest() {HelloService helloService = rpcProxy.create(HelloService.class);String result = helloService.hello("World");Assert.assertEquals("Hello! World", result);}
}

本文通过 Spring + Netty + Protostuff + ZooKeeper 实现了一个轻量级 RPC 框架，使用 Spring 提供依赖注入与参数配置，使用 Netty 实现 NIO 方式的数据传输，使用 Protostuff 实现对象序列化，使用 ZooKeeper 实现服务注册与发现。使用该框架，可将服务部署到分布式环境中的任意节点上，客户端通过远程接口来调用服务端的具体实现，让服务端与客户端的开发完全分离，为实现大规模分布式应用提供了基础支持。

附录：Maven 依赖

<!-- lang: xml -->
<!-- JUnit -->
<dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>4.11</version><scope>test</scope>
</dependency><!-- SLF4J -->
<dependency><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId><version>1.7.7</version>
</dependency><!-- Spring -->
<dependency><groupId>org.springframework</groupId><artifactId>spring-context</artifactId><version>3.2.12.RELEASE</version>
</dependency>
<dependency><groupId>org.springframework</groupId><artifactId>spring-test</artifactId><version>3.2.12.RELEASE</version><scope>test</scope>
</dependency><!-- Netty -->
<dependency><groupId>io.netty</groupId><artifactId>netty-all</artifactId><version>4.0.24.Final</version>
</dependency><!-- Protostuff -->
<dependency><groupId>com.dyuproject.protostuff</groupId><artifactId>protostuff-core</artifactId><version>1.0.8</version>
</dependency>
<dependency><groupId>com.dyuproject.protostuff</groupId><artifactId>protostuff-runtime</artifactId><version>1.0.8</version>
</dependency><!-- ZooKeeper -->
<dependency><groupId>org.apache.zookeeper</groupId><artifactId>zookeeper</artifactId><version>3.4.6</version>
</dependency><!-- Apache Commons Collections -->
<dependency><groupId>org.apache.commons</groupId><artifactId>commons-collections4</artifactId><version>4.0</version>
</dependency><!-- Objenesis -->
<dependency><groupId>org.objenesis</groupId><artifactId>objenesis</artifactId><version>2.1</version>
</dependency><!-- CGLib -->
<dependency><groupId>cglib</groupId><artifactId>cglib</artifactId><version>3.1</version>
</dependency>

http://javatar.iteye.com/blog/1123915

转载于:https://www.cnblogs.com/mxz1994/p/8435917.html

轻量级分布式 RPC 框架相关推荐

一个轻量级分布式RPC框架--NettyRpc
1.背景最近在搜索Netty和Zookeeper方面的文章时,看到了这篇文章<轻量级分布式 RPC 框架>,作者用Zookeeper.Netty和Spring写了一个轻量级的分布式RPC ...
轻量级分布式 RPC 框架DIY（转）
http://www.csdn123.com/html/topnews201408/67/12167.htm RPC,即 Remote Procedure Call(远程过程调用),说得通俗一点 ...
如何实现一个分布式 RPC 框架
远程过程调用(Remote Procedure Call,RPC)是一个计算机通信协议.该协议允许运行于一台计算机的程序调用另一台计算机的子程序,而程序员无需额外地为这个交互作用编程.RPC的主要目标 ...
dubbo-快速入门-分布式RPC框架Apache Dubbo
文章目录分布式RPC框架Apache Dubbo 1. 软件架构的演进过程 1.1 单体架构 1.2 垂直架构 1.3 SOA架构 1.4 微服务架构 2. Apache Dubbo概述 2.1 D ...
分布式RPC框架Apache Dubbo
分布式RPC框架Apache Dubbo 1. 软件架构的演进过程软件架构的发展经历了由单体架构.垂直架构.SOA架构到微服务架构的演进过程,下面我们分别了解一下这几个架构. 1.1 单体架构 1. ...
自己动手从0开始实现一个分布式RPC框架
简介: 如果一个程序员能清楚的了解RPC框架所具备的要素,掌握RPC框架中涉及的服务注册发现.负载均衡.序列化协议.RPC通信协议.Socket通信.异步调用.熔断降级等技术,可以全方位的提升基本素质 ...
还发愁项目经验吗？基于Netty实现分布式RPC框架[附完整代码]
写给大家的话最近我收到很多读者的来信,对如何学习分布式.如何进行项目实践和提高编程能力,存在很多疑问. 分布式那么难,怎么学?为什么看了那么多书还是掌握不了? 开源的框架比如Dubbo代码太多了,完 ...
基于zeromq的高性能分布式RPC框架Zerorpc 性能测试
Zeromq 是基于zeromq.gevent和 msgpack开发的分布式RPC框架zerorpc-python.这个框架简单.易用. 1. 安装zeromq 1 2 3 4 5 6 yum -y ...
LTS 轻量级分布式任务调度框架(Light Task Schedule) - 推酷
LTS 轻量级分布式任务调度框架(Light Task Schedule) - 推酷

轻量级分布式 RPC 框架