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

文章目录

Pre
设计
- 技术点
- 1. 接口设计
- 2. 命令类设计
- 3. 异步通信
- 4. 异常处理与超时机制
- 5. 背压机制
- 6. 响应处理
Code
- 封装通信 Transport 接口
- 抽象数据的请求和响应 Command
- - 关于版本号
- Transport 接口实现类 NettyTransport
- - 兜底的超时机制
- 背压机制实现
总结

在这里插入图片描述

Pre

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

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

设计

技术点

接口设计： Transport 接口的设计思路及其方法的实现。
命令类设计： Command 类及其 Header 的设计和用途。
异步通信： NettyTransport 类中 send 方法的异步实现，以及如何管理请求与响应的配对。
异常处理与超时机制：如何处理发送过程中的异常和可能的超时情况，确保内存资源不被浪费。
背压机制：如何在异步通信中引入背压机制，以避免服务端过载。
响应处理：如何在接收到响应时，使用 ResponseInvocation 类进行响应的处理。

1. 接口设计

在 RPC 框架中，通信模块通过 Transport 接口封装了通信逻辑。

send 方法定义了发送请求并异步获取响应的行为，返回值为 CompletableFuture<Command>，提供了灵活的同步或异步调用方式。

这样的设计使得客户端既能直接获取响应（同步调用），也能在响应返回后执行特定操作（异步调用），大大增加了接口的使用场景。

2. 命令类设计

Command 类封装了要传输的数据，其结构包括 Header 和 payload（命令中要传输的数据）。

其中

Header 包含 requestId、version 和 type 等字段，用于标识请求和响应的匹配、命令版本及命令类型。
ResponseHeader 还增加了 code 和 error 字段，用于标识响应的状态。这种设计提供了必要的信息，确保了通信的正确性和兼容性。

3. 异步通信

NettyTransport 类实现了 send 方法，其核心是使用 CompletableFuture 来处理异步请求。发送请求后，该方法不会阻塞当前线程，而是立即返回 CompletableFuture。

方法内，inFlightRequests 用于存储在途请求，这样当响应返回时，可以通过 requestId 找到对应的 ResponseFuture 并处理。

4. 异常处理与超时机制

异步请求由于其非阻塞特性，可能遇到请求发送失败或接收不到响应等问题。

为此，NettyTransport 类在请求发送和发送异常时均进行了异常捕获处理，确保在异常情况下 ResponseFuture 能及时终止并释放资源。

此外，为了处理可能由于对端未响应导致的孤儿 ResponseFuture，还需要设计超时机制，在超时时间到达后强制终止未完成的请求，防止资源泄漏。

5. 背压机制

为了防止异步通信中客户端发送请求速度过快而导致服务端处理不过来的问题，引入了信号量控制的背压机制。

通过 Semaphore 限制同时处理的请求数，当请求数超过限制时，发送线程会被阻塞，直到有请求完成。这种机制保证了在途请求不会无限增长，从而避免了服务端过载。

6. 响应处理

响应返回时，通过 ResponseInvocation 类来处理。根据响应中的 requestId，在 inFlightRequest 中找到对应的 ResponseFuture，设置响应结果并结束该 ResponseFuture。这使得整个请求-响应过程能有效地闭环运行，保证了系统的稳定性。

Code

封装通信 Transport 接口

我们可以将通信模块封装为一个接口。在这个 RPC 框架中，通信模块的需求很简单：客户端向服务端发送请求，然后服务端返回响应。

因此，通信接口只需要提供一个发送请求的方法：

/**
 * Transport接口定义了传输层的通用方法
 * 该接口中的方法主要用于发送异步命令
 */
public interface Transport {
    /**
     * 发送请求命令
     * 该方法允许异步发送一个命令，并返回一个CompletableFuture对象用于处理响应
     *
     * @param request 请求命令，包含要发送的所有信息
     * @return 返回值是一个Future，用于处理命令的异步执行结果
     *         通过这个Future对象，调用者可以检查命令是否执行成功，以及获取命令的执行结果
     */
    CompletableFuture<Command> send(Command request);
}

send 方法的参数 request 是要发送的请求数据，返回值是一个 CompletableFuture 对象。

通过这个 CompletableFuture，可以灵活地获取响应结果。

可以直接调用它的 get 方法来同步获取响应，
或使用 then 开头的一系列方法来指定响应返回后的操作，支持异步调用。

这样，一个方法既能同步调用，又能异步调用，使用非常方便。

抽象数据的请求和响应 Command

在这个接口中，数据的请求和响应被抽象为一个 Command 类。

public class Command {
    protected Header header;
    private byte[] payload;
    //...
}
 
public class Header {
    private int requestId;
    private int version;
    private int type;
    // ...
}
public class ResponseHeader extends Header {
    private int code;
    private String error;
    // ...
}

Command 类由一个 Header 和一个 payload 字节数组组成。payload 代表命令中需要传输的数据，在这里要求该数据已被序列化为字节数组。

Header 包含三个关键属性：

requestId：唯一标识一个请求命令，尤其在异步双向通信中，这个 ID 可以用来将请求和响应正确配对。
version：标识命令的版本号。
type：标识命令的类型，方便接收方识别命令的种类并将其路由到相应的处理器。

此外，响应的 Header (ResponseHeader)还增加了两个字段：

code：使用数字表示响应状态，0 表示成功，其他值表示各种错误状态，这个设计类似于 HTTP 协议中的 StatusCode。
error：用于传递错误信息。

关于版本号

在设计通信协议时，保持协议的可持续升级能力和向下兼容性至关重要。由于需求的变化，传输协议也可能会发生变化。为了确保使用该传输协议的程序能够正常工作并兼容旧版本，协议中必须包含一个版本号，标明该数据使用的是哪个版本的协议。

在发送命令时，发送方必须包含该命令的版本号。接收方收到命令后，首先需要检查版本号。如果接收方支持该版本的命令，就正常处理，否则应拒绝该命令，并返回响应告知对方：“我不认识这个命令。” 这样的设计确保了通信协议的完备性和可持续性。

这里的版本号是指命令的版本号，也即传输协议的版本号，而不是程序的版本号。需要实现这部分校验。

Transport 接口实现类 NettyTransport

我们来看一下 Transport 接口的实现类 NettyTransport。send 方法是一个典型的异步方法，它在将请求数据发送出去之后立即返回，而不会阻塞当前线程等待响应。

@Override
  /**
     * 发送命令方法
     * 该方法将命令发送到远程服务器，并管理与该请求相关的未来响应
     *
     * @param request   要发送的命令请求
     * @return 返回一个CompletableFuture，用于处理异步响应
     */
    @Override
    public CompletableFuture<Command> send(Command request) {
        // 构建返回值
        CompletableFuture<Command> completableFuture = new CompletableFuture<>();
        try {
            // 将在途请求放到inFlightRequests中，以便在收到响应时可以找到对应的Future
            inFlightRequests.put(new ResponseFuture(request.getHeader().getRequestId(), completableFuture));
            // 发送命令
            channel.writeAndFlush(request).addListener((ChannelFutureListener) channelFuture -> {
                // 处理发送失败的情况
                if (!channelFuture.isSuccess()) {
                    completableFuture.completeExceptionally(channelFuture.cause());
                    channel.close();
                }
            });
        } catch (Throwable t) {
            // 处理发送异常
            inFlightRequests.remove(request.getHeader().getRequestId());
            completableFuture.completeExceptionally(t);
        }
        return completableFuture;
    }

这段代码主要做了两件事：

请求与返回值的关联：
它首先将请求中的 requestId 和返回的 CompletableFuture 对象一起封装成一个 ResponseFuture 对象，并将其存入 inFlightRequests。inFlightRequests 维护了所有已发送但尚未收到响应的请求，这些请求的 ResponseFuture 对象都保存在这里。
发送请求：
然后，它调用 Netty 的方法将 request 命令发送给服务端。需要注意的是，发送请求可能会遇到各种异常情况，如网络连接中断或对方进程崩溃，导致无法收到响应。为了防止这些“孤儿”ResponseFuture 在内存中积累，代码对可能的异常情况进行了处理，以确保相关的 ResponseFuture 及时结束。

具体来说，代码在两个地方进行了异常处理：

如果发送请求失败，CompletableFuture 将通过 completeExceptionally 方法完成，并且关闭连接。
如果在整个发送过程中抛出异常，inFlightRequests 中的对应请求将被移除，且 CompletableFuture 也会通过 completeExceptionally 方法完成。

通过这些处理，确保即使在异常情况下，系统也能正确地释放资源，不会出现资源泄漏或内存问题。

兜底的超时机制

即使我们对所有可以捕获的异常都做了处理，也无法保证所有的 ResponseFuture 都能正常或异常地结束。例如，如果对端程序的开发人员编写的代码有问题，导致它接收到请求后未能返回响应，那么这些请求就可能永远挂起。为了解决这种情况，我们还需要引入一个兜底的超时机制。

这个超时机制的作用是在所有情况下确保 ResponseFuture 最终结束。无论发生什么情况，只要超过设定的超时时间仍未收到响应，我们就认为该 ResponseFuture 已经失败，随后结束并从内存中删除它。这个机制可以防止系统中出现挂起的请求从而导致资源泄漏。

InFlightRequests 负责管理所有在途请求，包括记录它们的状态、处理响应以及超时清理等操作。通过定时任务或其他方式监控每个请求的超时时间，当超时后，会自动将对应的 ResponseFuture 标记为失败并清理相关资源。

这种设计可以有效防止系统因请求堆积而出现性能问题或内存泄漏，确保系统的稳定性和可靠性。

背压机制实现

同步请求时，客户端必须等待服务端的响应，服务端处理请求所需的时间决定了客户端等待的时间。这种等待机制实际上形成了一种天然的背压机制（Back Pressure），即服务端的处理速度会限制客户端的请求速度。

在异步请求中，客户端不会等待服务端的响应，而是连续地发送请求。这种情况下，缺少了同步请求中的天然背压机制，可能导致问题。

如果服务端的处理速度跟不上客户端的请求速度，客户端的请求速度不会减慢，导致在途的请求越来越多，这些请求堆积在服务端的内存中。当内存耗尽时，进一步的请求将会失败。如果服务端已经处理不过来，客户端还在不停地发送请求，这显然是无意义的。

为了避免这种情况，我们需要在服务端处理不过来的时候限制客户端的请求速度，即增加一个背压机制。

这个背压机制可以通过 InFlightRequests 类中的信号量来实现：

private final Semaphore semaphore = new Semaphore(10);

这个信号量初始时有 10 个许可。每次我们将一个 ResponseFuture 加入 inFlightRequests 时，需要先从信号量中获取一个许可。如果此时没有许可可用，发送请求的线程将会被阻塞，直到有许可被归还，线程才能继续发送请求。当一个在途请求结束时，我们归还一个许可。这样就确保了在途请求的数量不会超过 10 个，服务端处理中的请求也不会超过 10 个，从而有效地实现了一个简单而有效的背压机制。

/**
 * InFlightRequests 类用于管理在在途的请求
 * 它通过使用信号量和一个映射来跟踪和管理请求，以确保同时只有有限数量的请求处于处理状态
 * 该类还提供了一个机制来定期清理超时的请求
 * @author artisan
 */
public class InFlightRequests implements Closeable {
    // 定义超时时间为10秒
    private final static long TIMEOUT_SEC = 10L;
    // 信号量用于控制同时处理的请求数量
    private final Semaphore semaphore = new Semaphore(10);
    // 使用ConcurrentHashMap来存储每个请求的响应未来对象
    private final Map<Integer, ResponseFuture> futureMap = new ConcurrentHashMap<>();
    // 使用单线程的ScheduledExecutorService来定期执行清理超时请求的任务
    private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
    // 定期执行的任务用于清理超时的请求
    private final ScheduledFuture scheduledFuture;

    /**
     * 构造函数初始化ScheduledFuture，用于定期执行清理超时请求的任务
     */
    public InFlightRequests() {
        scheduledFuture = scheduledExecutorService.scheduleAtFixedRate(this::removeTimeoutFutures, TIMEOUT_SEC, TIMEOUT_SEC, TimeUnit.SECONDS);
    }

    /**
     * 将一个请求添加到管理中，如果无法在超时时间内获取信号量，则抛出TimeoutException
     *
     * @param responseFuture 要添加的响应未来对象
     * @throws InterruptedException 如果线程被中断
     * @throws TimeoutException 如果在规定超时时间内无法获取信号量
     */
    public void put(ResponseFuture responseFuture) throws InterruptedException, TimeoutException {
        if(semaphore.tryAcquire(TIMEOUT_SEC, TimeUnit.SECONDS)) {
            futureMap.put(responseFuture.getRequestId(), responseFuture);
        } else {
            throw new TimeoutException();
        }
    }

    /**
     * 私有方法，用于定期清理超时的请求，并释放信号量
     */
    private void removeTimeoutFutures() {
        futureMap.entrySet().removeIf(entry -> {
            if( System.nanoTime() - entry.getValue().getTimestamp() > TIMEOUT_SEC * 1000000000L) {
                semaphore.release();
                return true;
            } else {
                return false;
            }
        });
    }

    /**
     * 根据请求ID移除对应的请求，并释放信号量
     *
     * @param requestId 请求的ID
     * @return 被移除的请求的响应未来对象，如果不存在则返回null
     */
    public ResponseFuture remove(int requestId) {
        ResponseFuture future = futureMap.remove(requestId);
        if(null != future) {
            semaphore.release();
        }
        return future;
    }

    /**
     * 关闭InFlightRequests，取消定期执行的任务，并关闭线程池
     */
    @Override
    public void close() {
        scheduledFuture.cancel(true);
        scheduledExecutorService.shutdown();
    }
}

在 ResponseInvocation 类中，我们异步接收所有服务端返回的响应。处理逻辑相对简单，根据响应头中的 requestId，在 inFlightRequests 中查找对应的 ResponseFuture，设置返回值并结束该 ResponseFuture。


/**
 * 一个Channel处理器，用于处理入站的响应消息
 * 它继承自SimpleChannelInboundHandler，并且被设计来只处理Command类型的对象
 * 主要功能是将接收到的响应与之前的请求进行匹配，并通知等待的调用者
 *
 * @author artisan
 */
@ChannelHandler.Sharable
public class ResponseInvocation extends SimpleChannelInboundHandler<Command> {

    private static final Logger logger = LoggerFactory.getLogger(ResponseInvocation.class);
    /**
     * 用于跟踪和管理未完成的请求
     */
    private final InFlightRequests inFlightRequests;

    /**
     * 构造函数，初始化ResponseInvocation实例
     *
     * @param inFlightRequests InFlightRequests实例，用于管理飞行中的请求
     */
    ResponseInvocation(InFlightRequests inFlightRequests) {
        this.inFlightRequests = inFlightRequests;
    }

    /**
     * 读取并处理接收到的响应命令
     * <p>
     * 当一个响应被读取时，这个方法会尝试找到与之匹配的请求，并完成相应的响应未来
     * 如果找不到匹配的请求，它会记录一条警告日志
     *
     * @param channelHandlerContext 上下文环境，提供通道和处理方法的相关信息
     * @param response              接收到的响应命令
     */
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, Command response) {
        // 从飞行中的请求中移除匹配的请求，并获取其未来对象
        ResponseFuture future = inFlightRequests.remove(response.getHeader().getRequestId());
        if (null != future) {
            // 如果找到了匹配的请求，完成其未来对象
            future.getFuture().complete(response);
        } else {
            // 如果找不到匹配的请求，记录警告日志
            logger.warn("Drop response: {}", response);
        }
    }

    /**
     * 当发生异常时被捕获并处理
     * <p>
     * 主要功能是记录异常信息，并关闭通道上下文
     *
     * @param ctx   通道上下文，提供对通道的访问和关闭操作
     * @param cause 异常原因
     * @throws Exception 如果异常无法被处理，它将被抛出
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        // 记录异常警告日志
        logger.warn("Exception: ", cause);
        // 调用父类的异常处理方法
        super.exceptionCaught(ctx, cause);
        // 获取当前通道，并检查其活动状态，如果活跃则关闭通道
        Channel channel = ctx.channel();
        if (channel.isActive()) {
            ctx.close();
        }
    }
}

总结

首先定义了对外提供服务的接口。这样，服务的使用者只需要依赖这个接口，而不需要关心其具体实现。

这种设计的好处在于，它有效地解耦了接口的使用者和实现者，使我们能够安全地替换接口的实现。通过将接口定义得尽量通用，接口就可以独立于具体的使用场景，从而实现高度的复用性。

例如，RPC框架中的网络传输和序列化代码，不仅能在这个框架中使用，甚至可以直接应用到其他系统中，而无需进行修改。

在协议设计方面，我们为每个命令设计了一个固定的头部信息。这样做的好处是，当我们接收到命令时，可以首先解析头部，从而进行版本检查和路由分发等操作，而无需立即解析整个命令的内容。为了应对需求变化并确保协议能够持续升级，每个命令都携带一个协议版本号。接收方在处理命令时需要检查这个版本号，以确保它支持该版本的协议。

在实现异步网络通信时，还需要配合实现一个背压机制。背压机制可以防止客户端请求速度过快，导致服务端无法及时处理而引发的大量请求失败。通过对在途请求的数量进行限制，我们确保系统在高负载情况下依然能够稳定运行。

在这里插入图片描述