當前位置：首頁 > 人文社科 > 生活经验 >内容正文

生活经验

轻量级分布式 RPC 框架

發(fā)布時間：2023/11/27 生活经验 28 豆豆

生活随笔收集整理的這篇文章主要介紹了轻量级分布式 RPC 框架小編覺得挺不錯的,現(xiàn)在分享給大家,幫大家做個參考.

RPC，即 Remote Procedure Call（遠程過程調(diào)用），說得通俗一點就是：調(diào)用遠程計算機上的服務(wù)，就像調(diào)用本地服務(wù)一樣。

RPC 可基于 HTTP 或 TCP 協(xié)議，Web Service 就是基于 HTTP 協(xié)議的 RPC，它具有良好的跨平臺性，但其性能卻不如基于 TCP 協(xié)議的 RPC。會兩方面會直接影響 RPC 的性能，一是傳輸方式，二是序列化。

輕量級分布式 RPC 框架的具體過程，該框架基于 TCP 協(xié)議

Spring：它是最強大的依賴注入框架，也是業(yè)界的權(quán)威標準。
Netty：它使 NIO 編程更加容易，屏蔽了 Java 底層的 NIO 細節(jié)。
Protostuff：它基于 Protobuf 序列化框架，面向 POJO，無需編寫 .proto 文件。
ZooKeeper：提供服務(wù)注冊與發(fā)現(xiàn)功能，開發(fā)分布式系統(tǒng)的必備選擇，同時它也具備天生的集群能力。

第一步：編寫服務(wù)接口

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

第二步：編寫服務(wù)接口的實現(xiàn)類

<!-- lang: java -->
@RpcService(HelloService.class) // 指定遠程接口
//使用RpcService注解定義在服務(wù)接口的實現(xiàn)類上，需要對該實現(xiàn)類指定遠程接口，因為實現(xiàn)類可能會實現(xiàn)多個接口，一定要告訴框架哪個才是遠程接口。

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();
}

第三步：配置服務(wù)端

服務(wù)端 Spring 配置文件名為spring.xml，內(nèi)容如下：

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

具體的配置參數(shù)在config.properties文件中，連接本地的 ZooKeeper 服務(wù)器，并在 8000 端口上發(fā)布 RPC 服務(wù)。

<!-- lang: java -->
# ZooKeeper 服務(wù)器
registry.address=127.0.0.1:2181# RPC 服務(wù)器
server.address=127.0.0.1:8000

第四步：啟動服務(wù)器并發(fā)布服務(wù)

為了加載 Spring 配置文件來發(fā)布服務(wù)，只需編寫一個引導(dǎo)程序即可：

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

運行RpcBootstrap類的main方法即可啟動服務(wù)端，但還有兩個重要的組件尚未實現(xiàn)，它們分別是：ServiceRegistry與RpcServer

第五步：實現(xiàn)服務(wù)注冊

使用 ZooKeeper 客戶端可輕松實現(xiàn)服務(wù)注冊功能，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 客戶端命令行創(chuàng)建/registry永久節(jié)點，用于存放所有的服務(wù)臨時節(jié)點

第六步：實現(xiàn) RPC 服務(wù)器

使用 Netty 可實現(xiàn)一個支持 NIO 的 RPC 服務(wù)器，需要使用ServiceRegistry注冊服務(wù)地址，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<>(); // 存放接口名與服務(wù)對象之間的映射關(guān)系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 響應(yīng)進行編碼（為了返回響應(yīng)）.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); // 注冊服務(wù)地址
            }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 響應(yīng)，代碼如下：

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實現(xiàn)序列化：

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即可。當然，更好的實現(xiàn)方式是提供配置項來決定使用哪種序列化方式。

使用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"/><!-- 配置服務(wù)發(fā)現(xiàn)組件 --><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 服務(wù)器
registry.address=127.0.0.1:2181

第八步：實現(xiàn)服務(wù)發(fā)現(xiàn)

同樣使用 ZooKeeper 實現(xiàn)服務(wù)發(fā)現(xiàn)功能，見如下代碼：

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);}}
}

第九步：實現(xiàn) RPC 代理

這里使用 Java 提供的動態(tài)代理技術(shù)實現(xiàn) RPC 代理（當然也可以使用 CGLib 來實現(xiàn)），具體代碼如下：

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(); // 創(chuàng)建并初始化 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(); // 發(fā)現(xiàn)服務(wù)
                    }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 客戶端發(fā)送 RPC 請求并獲取 RPC 響應(yīng)if (response.isError()) {throw response.getError();} else {return response.getResult();}}});}
}

使用RpcClient類實現(xiàn) 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(); // 收到響應(yīng)，喚醒線程
        }}@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 請求進行編碼（為了發(fā)送請求）.addLast(new RpcDecoder(RpcResponse.class)) // 將 RPC 響應(yīng)進行解碼（為了處理響應(yīng)）.addLast(RpcClient.this); // 使用 RpcClient 發(fā)送 RPC 請求
                    }}).option(ChannelOption.SO_KEEPALIVE, true);ChannelFuture future = bootstrap.connect(host, port).sync();future.channel().writeAndFlush(request).sync();synchronized (obj) {obj.wait(); // 未收到響應(yīng)，使線程等待
            }if (response != null) {future.channel().closeFuture().sync();}return response;} finally {group.shutdownGracefully();}}
}

第十步：發(fā)送 RPC 請求

使用 JUnit 結(jié)合 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 實現(xiàn)了一個輕量級 RPC 框架，使用 Spring 提供依賴注入與參數(shù)配置，使用 Netty 實現(xiàn) NIO 方式的數(shù)據(jù)傳輸，使用 Protostuff 實現(xiàn)對象序列化，使用 ZooKeeper 實現(xiàn)服務(wù)注冊與發(fā)現(xiàn)。使用該框架，可將服務(wù)部署到分布式環(huán)境中的任意節(jié)點上，客戶端通過遠程接口來調(diào)用服務(wù)端的具體實現(xiàn)，讓服務(wù)端與客戶端的開發(fā)完全分離，為實現(xiàn)大規(guī)模分布式應(yīng)用提供了基礎(chǔ)支持。

附錄：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

轉(zhuǎn)載于:https://www.cnblogs.com/mxz1994/p/8435917.html

總結(jié)

以上是生活随笔為你收集整理的轻量级分布式 RPC 框架的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。

上一篇：上海比熊网络科技有限公司怎么样？
下一篇：沈腾公交强吻女主是什么电视或者电影