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

生活经验

轻量级分布式 RPC 框架

發布時間：2023/11/27 生活经验 38 豆豆

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

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 框架的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。