Simple RPC - 06 从零开始设计一个服务端（上）

文章目录

Pre
核心内容
服务端结构概述
注册中心的实现
- 1. 注册中心的架构
- 2. 面向接口编程的设计
- 3. 注册中心的接口设计
- 4. SPI机制的应用
5. 小结

在这里插入图片描述

Pre

Simple RPC - 01 框架原理及总体架构初探

Simple RPC - 02 通用高性能序列化和反序列化设计与实现

Simple RPC - 03 借助Netty实现异步网络通信

Simple RPC - 04 从零开始设计一个客户端（上）

Simple RPC - 05 从零开始设计一个客户端（下）_ 依赖倒置和SPI

核心内容

服务端结构概述：注册中心和RPC服务的结构及作用。
注册中心实现：通过单机版的注册中心实现共享元数据，分析其接口设计和SPI机制。
RPC服务实现：理解服务端处理RPC请求的核心逻辑，包括如何注册服务和处理请求。
请求分发机制：深入了解RequestInvocation和RpcRequestHandler类中的请求分发机制。
代码分析与总结：通过代码实例进一步理解设计思想，并总结整体架构和设计原则。

服务端结构概述

在RPC框架中，服务端可以分为两个主要部分：注册中心和RPC服务。

注册中心：负责管理服务元数据，并提供服务发现的功能。
RPC服务：负责处理客户端发来的RPC请求，并调用相应的业务服务。

简单来说：注册中心的作用是帮助客户端来寻址，找到对应 RPC 服务的物理地址；RPC 服务用于接收客户端桩的请求，调用业务服务的方法，并返回结果。

注册中心的实现

1. 注册中心的架构

通常，一个完整的注册中心包括客户端和服务端两部分：

客户端：向调用方提供 API，负责与注册中心服务端的通信。
服务端：实际管理和记录每个 RPC 服务的注册信息，并将这些信息存储在元数据中。当客户端需要查找服务时，服务端会返回对应的服务地址。

在本例中，出于简化考虑，我们实现了一个单机版的注册中心。这个注册中心只有客户端部分，多个客户端通过读写同一个本地元数据文件实现服务信息的共享。

该注册中心只能在单机环境下运行，不支持跨服务器调用。

2. 面向接口编程的设计

尽管当前实现是单机版的注册中心，但通过“面向接口编程”的设计模式，我们可以在不修改已有代码的情况下，通过 SPI 插件机制，扩展出一个支持跨服务器调用的注册中心（例如，基于 HTTP 协议的实现）。

3. 注册中心的接口设计

在 RPC 框架的接入点接口 RpcAccessPoint 中，增加了一个用于获取注册中心实例的方法：

public interface RpcAccessPoint extends Closeable {
    /**
     * 获取注册中心的引用
     * @param nameServiceUri 注册中心 URI
     * @return 注册中心引用
     */
    NameService getNameService(URI nameServiceUri);
}

该方法接受一个注册中心的 URI 作为参数，并返回一个 NameService 接口的实例。这个 NameService 接口表示注册中心的客户端，可以用来和注册中心服务端通信。

NameService 接口中定义了与注册中心通信的核心方法：

public interface NameService {
    /**
     * 返回所有支持的协议
     */
    Collection<String> supportedSchemes();

    /**
     * 连接注册中心
     * @param nameServiceUri 注册中心地址
     */
    void connect(URI nameServiceUri);
}

supportedSchemes 方法返回注册中心支持的协议（例如 file 或 http）。
connect 方法根据 URI 建立与注册中心的连接。

完整代码如下

/**
 * 注册中心接口定义
 * 该接口用于服务的注册和发现，支持多种通信协议
 *
 * @author artisan
 */
public interface NameService {

    /**
     * 获取所有支持的协议
     *
     * @return 支持的协议集合
     */
    Collection<String> supportedSchemes();

    /**
     * 连接注册中心
     *
     * @param nameServiceUri 注册中心的URI地址
     */
    void connect(URI nameServiceUri);

    /**
     * 注册服务
     *
     * @param serviceName 服务名称
     * @param uri 服务的URI地址
     * @throws IOException 如果连接或注册失败，则抛出此异常
     */
    void registerService(String serviceName, URI uri) throws IOException;

    /**
     * 查询服务地址
     *
     * @param serviceName 服务名称
     * @return 服务的URI地址
     * @throws IOException 如果查找失败，则抛出此异常
     */
    URI lookupService(String serviceName) throws IOException;
}

4. SPI机制的应用

通过 SPI 机制，RpcAccessPoint 可以根据 URI 中指定的协议，动态加载不同的 NameService 实现类。例如，在单机版注册中心中，NameService 的实现类是 LocalFileNameService，其具体功能是读写本地文件，存储和查找服务信息。

public class LocalFileNameService implements NameService {
    @Override
    public Collection<String> supportedSchemes() {
        return Collections.singleton("file");
    }

    @Override
    public void connect(URI nameServiceUri) {
        // 连接到本地文件，初始化文件读写工具
    }

    // 其他方法实现...
}

通过这种方式，新的注册中心实现可以通过 SPI 动态添加到系统中。例如，要实现一个基于 HTTP 的注册中心，只需开发一个新的 NameService 实现类，并将其添加到系统的 CLASSPATH 中即可。

LocalFileNameService代码如下


/**
 * LocalFileNameService 类实现了 NameService 接口，提供了一种基于文件系统来管理服务名称和URI的实现方式
 * 它使用 "file" 协议来操作本地文件，并将服务信息存储在文件中
 * @author artisan
 */
public class LocalFileNameService implements NameService {
    private static final Logger logger = LoggerFactory.getLogger(LocalFileNameService.class);
    /**
     * 支持的协议集合，本实现仅支持 "file" 协议
     */
    private static final Collection<String> schemes = Collections.singleton("file");

    /**
     * 用于存储服务信息的文件对象
     */
    private File file;

    /**
     * 返回此服务支持的协议集合
     *
     * @return 支持的协议集合
     */
    @Override
    public Collection<String> supportedSchemes() {
        return schemes;
    }

    /**
     * 连接到指定的名称服务URI，如果支持该URI的协议，则将URI解析为本地文件
     * 此方法首先检查给定的URI是否使用受支持的协议如果协议受支持，则将URI转换为本地文件路径
     * 如果不支持该协议，则抛出运行时异常
     *
     * @param nameServiceUri 名称服务的URI，用于连接和解析
     * @throws RuntimeException 如果URI的协议不受支持，则抛出此异常
     */
    @Override
    public void connect(URI nameServiceUri) {
        // 检查URI的协议是否在支持的协议列表中
        if (schemes.contains(nameServiceUri.getScheme())) {
            // 如果协议受支持，则将URI转换为本地文件路径
            file = new File(nameServiceUri);
        } else {
            // 如果协议不受支持，则抛出异常
            throw new RuntimeException("Unsupported scheme!");
        }
    }


    /**
     * 注册服务，将服务名称和服务URI写入到文件中
     * 此方法是同步的，以确保并发访问时的数据一致性
     *
     * @param serviceName 服务名称
     * @param uri 服务的URI
     * * @throws IOException 如果发生I/O错误
     */
    @Override
    public synchronized void registerService(String serviceName, URI uri) throws IOException {
        // 记录服务注册的日志信息
        logger.info("Register service: {}, uri: {}.", serviceName, uri);
        // 使用RandomAccessFile和FileChannel来读写文件，并确保资源在使用后能够正确关闭
        try (RandomAccessFile raf = new RandomAccessFile(file, "rw");
             FileChannel fileChannel = raf.getChannel()) {
            // 获取文件锁，以确保并发访问时的数据一致性
            FileLock lock = fileChannel.lock();
            try {
                // 获取文件长度，用于后续判断文件是否为空
                int fileLength = (int) raf.length();
                Metadata metadata;
                byte[] bytes;
                // 如果文件长度大于0，说明文件非空，读取并解析文件内容
                if (fileLength > 0) {
                    bytes = new byte[(int) raf.length()];
                    ByteBuffer buffer = ByteBuffer.wrap(bytes);
                    // 循环读取文件内容到ByteBuffer中
                    while (buffer.hasRemaining()) {
                        fileChannel.read(buffer);
                    }

                    // 解析字节码为Metadata对象
                    metadata = SerializeSupport.parse(bytes);
                } else {
                    // 如果文件为空，创建一个新的Metadata对象
                    metadata = new Metadata();
                }
                // 根据服务名获取或创建一个空的URI列表
                List<URI> uris = metadata.computeIfAbsent(serviceName, k -> new ArrayList<>());
                // 如果列表中不存在该URI，则添加进去
                if (!uris.contains(uri)) {
                    uris.add(uri);
                }
                // 记录更新后的Metadata信息
                logger.info(metadata.toString());

                // 将Metadata对象序列化为字节码
                bytes = SerializeSupport.serialize(metadata);
                // 清空文件，为写入新的字节码做准备
                fileChannel.truncate(bytes.length);
                // 将文件指针移到文件开头，准备写入
                fileChannel.position(0L);
                // 将字节码写入文件
                fileChannel.write(ByteBuffer.wrap(bytes));
                // 强制将写入操作刷入磁盘
                fileChannel.force(true);
            } finally {
                // 释放文件锁
                lock.release();
            }
        }
    }


    /**
     * 根据服务名称查找服务的URI
     * 如果文件中存在对应的服务URI，则随机返回一个
     *
     * @param serviceName 服务名称
     * @return 服务的URI，如果找不到则返回null
     * @throws IOException 如果发生I/O错误
     */
    @Override
    public URI lookupService(String serviceName) throws IOException {
        Metadata metadata;
        // 使用try-with-resources语句确保文件资源正确关闭
        try (RandomAccessFile raf = new RandomAccessFile(file, "rw");
             FileChannel fileChannel = raf.getChannel()) {
            // 获取文件锁以确保数据的一致性
            FileLock lock = fileChannel.lock();
            try {
                // 读取文件内容到字节数组
                byte[] bytes = new byte[(int) raf.length()];
                ByteBuffer buffer = ByteBuffer.wrap(bytes);
                // 循环读取直到文件末尾
                while (buffer.hasRemaining()) {
                    fileChannel.read(buffer);
                }
                // 如果文件非空，则反序列化为Metadata对象，否则创建新的空Metadata对象
                metadata = bytes.length == 0 ? new Metadata() : SerializeSupport.parse(bytes);
                // 记录日志
                logger.info(metadata.toString());
            } finally {
                // 释放文件锁
                lock.release();
            }
        }

        // 从Metadata中获取服务的所有URI
        List<URI> uris = metadata.get(serviceName);
        // 如果没有找到对应的URI列表，返回null
        if (null == uris || uris.isEmpty()) {
            return null;
        } else {
            // 随机选择一个URI返回
            return uris.get(ThreadLocalRandom.current().nextInt(uris.size()));
        }
    }
}