RPC框架 – 源码巴士

1.RPC框架介绍

一句话介绍：RPC技术是为了解决远程调用服务的一种技术，使得调用者像调用本地服务一样方便透明。

技术对比：

 1.RPC采用二进制字节码传输更加安全。

 2.HTTP协议，HTTP 是应用层协议；RPC大多是TCP实现，TCP 是传输层协议。更底层的网络协议，更加高效。

在这里插入图片描述

2.轻量分布式RPC框架实现

在这里插入图片描述

2.1 服务端实现

在这里插入图片描述

编写服务接口
<!-- lang: java -->
public interface HelloService { 
	String hello(String name);
}
将该接口放在独立的客户端 jar 包中，以供应用使用。

2. 编写服务接口的实现类

<!-- lang: java --> 
@RpcService(HelloService.class) // 指定远程接口 
public class HelloServiceImpl implements HelloService {
  @Override 
  public String hello(String name) { 
  	  return "Hello! " + name; 
  } 
}

使用RpcService注解定义在服务接口的实现类上，需要对该实现类指定远程接口，因为实现类可能会实现多个接口，一定要告诉框架哪个才是远程接口。

RpcService代码如下：

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

该注解具备 Spring 的Component注解的特性，可被 Spring 扫描。

该实现类放在服务端 jar 包中，该 jar 包还提供了一些服务端的配置文件与启动服务的引导程序。

3. 配置服务端

服务端 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文件中，内容如下：

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


以上配置表明：连接本地的 ZooKeeper 服务器，并在 8000 端口上发布 RPC 服务。

4. 启动服务器并发布服务

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

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

运行RpcBootstrap类的main方法即可启动服务端，但还有两个重要的组件尚未实现，它们分别是：ServiceRegistry与RpcServer，下文会给出具体实现细节。

5. 实现服务注册

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

<!-- lang: java --> 
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() { 
              @Override 
              public 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配置了所有的常量：

<!-- lang: java --> 
public interface Constant { 
    int ZK_SESSION_TIMEOUT = 5000; 
    String ZK_REGISTRY_PATH = "/registry";
    String ZK_DATA_PATH = ZK_REGISTRY_PATH + "/data"; 
}

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

6.实现 RPC 服务器

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

<!-- lang: java --> 
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; 
    } 
    @Override 
    public void setApplicationContext(ApplicationContext ctx) throws BeansException { 
        Map<String, Object> serviceBeanMap = ctx.getBeansWithAnnotation(RpcService.class); 
        // 获取所有带有 RpcService 注解的 Spring Bean 
        if (MapUtils.isNotEmpty(serviceBeanMap)) { 
            for (Object serviceBean : serviceBeanMap.values()) { 
                String interfaceName = serviceBean.getClass().getAnnotation(RpcService.class).value().getName(); 
                handlerMap.put(interfaceName, serviceBean); 
            } 
        } 
    } 
    @Override 
    public 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>() { 
                @Override 
                public 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 请求，代码如下：

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


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

<!-- lang: java --> 
public class RpcResponse { 
    private String requestId; 
    private Throwable error; 
    private Object result; // getter/setter... 
}


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

<!-- lang: java --> 
public class RpcDecoder extends ByteToMessageDecoder { 
    private Class<?> genericClass; 
    public RpcDecoder(Class<?> genericClass) { 
    this.genericClass = genericClass;
    } 
    @Override 
    public 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);
    } 
}

<!-- lang: java -->
public class RpcDecoder extends ByteToMessageDecoder { 
    private Class<?> genericClass; 
    public RpcDecoder(Class<?> genericClass) { 
    this.genericClass = genericClass; 
    } 
    @Override 
    public 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方法即可，代码如下：

<!-- lang: java --> 
public class RpcEncoder extends MessageToByteEncoder { 
    private Class<?> genericClass; 
    public RpcEncoder(Class<?> genericClass) {
    this.genericClass = genericClass;
    }
    @Override
    public 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实现序列化：

<!-- lang: java --> 
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抽象类即可，代码如下：

<!-- lang: java -->
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;
    }
    @Override
    public 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); 
    } 
    @Override 
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
    LOGGER.error("server caught exception", cause); ctx.close(); 
    } 
}

为了避免使用 Java 反射带来的性能问题，我们可以使用 CGLib 提供的反射 API，如上面用到的FastClass与FastMethod。
在这里插入图片描述

2.2 客户端实现

在这里插入图片描述

1.配置客户端

同样使用 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>


其中config.properties提供了具体的配置：

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

2.实现服务发现

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

<!-- lang: java --> 
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() { 
                @Override 
                public 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() { 
                @Override 
                public 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);
        } 
    }
}

3.实现 RPC 代理

在这里插入图片描述

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

<!-- lang: java --> 
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() {
            @Override 
            public 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抽象类即可，代码如下：

<!-- lang: java --> 
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; 
    }
    @Override 
    public void channelRead0(ChannelHandlerContext ctx, RpcResponse response) throws Exception {
        this.response = response; synchronized (obj) { 
        obj.notifyAll(); // 收到响应，唤醒线程 
        } 
    }
    @Override 
    public 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>() { 
                @Override 
                public 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(); 
        } 
    } 
}

4.发送 RPC 请求

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

<!-- lang: java -->
@RunWith(SpringJUnit4ClassRunner.class) 
@ContextConfiguration(locations = "classpath:spring.xml") 
public class HelloServiceTest { 
    @Autowired 
    private RpcProxy rpcProxy;
    @Test 
    public void helloTest() { 
        HelloService helloService = rpcProxy.create(HelloService.class); 
        String result = helloService.hello("World");
        Assert.assertEquals("Hello! World", result);
    } 
}

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

3.依赖框架简介

3.1 zookeeper在RPC框架中作用

简单来说zookeeper=文件系统+监听通知机制。

提供者RPCServer：服务启动后向zookeeper注册他有的services，并将自己的ip地址和端口作为路径，创建对应的URL临时节点调用者RPCClient：调用相应服务时，找到对应的service节点，获得service所有的子节点，并且watch service节点,然后同样注册自己的znode节点调用过程：每个调用端需明确提供者和调用者的数量以及提供者相应的IP地址，之后调用端获得 service/providers的所有子节点即获得所有的提供者的IP 使用对应负载均衡策略连接其中一个ip地址，进行rpc调度

3.2 netty在RPC框架中的作用

原文链接：https://blog.csdn.net/weixin_44034379/article/details/107773580