Java 怎么实现基于 Java NIO 的自定义 RPC 传输层-java教程-PHP中文网

java基于nio实现rpc传输层的核心是selector+channel替代bio，通过服务端、客户端、编解码器、请求上下文四大组件分离职责，采用自定义二进制协议解决粘包与序列化，确保高并发低线程数下的高效通信。

java 怎么实现基于 java nio 的自定义 rpc 传输层

Java 基于 NIO 实现自定义 RPC 传输层，核心是用 Selector + Channel（如 SocketChannel / ServerSocketChannel）替代传统 BIO 的阻塞 I/O，支撑高并发、低线程数的远程调用通信。重点不在“造轮子”，而在于清晰分离协议编解码、连接管理、请求路由与事件调度。

一、设计关键组件与职责划分

一个轻量但可用的 NIO RPC 传输层需包含以下角色：

服务端（NioRpcServer）：绑定端口，注册 ServerSocketChannel 到 Selector，接受新连接；为每个客户端连接创建并注册对应的 SocketChannel
客户端（NioRpcClient）：维护连接池（可复用 SocketChannel），异步发起连接；支持连接失败重试与空闲检测
消息编解码器（Encoder/Decoder）：定义二进制协议（如魔数 + 长度字段 + 序列化体），在读写时完成 ByteBuffer 与 RpcRequest/RpcResponse 的转换
请求上下文管理（RequestContext）：客户端发出请求时生成唯一 requestId，本地缓存 Promise 或 CompletableFuture，收到响应后唤醒对应等待线程

二、服务端 NIO 主循环示例（精简版）

服务端主线程只做事件分发，不执行业务逻辑：

public class NioRpcServer {
    private final Selector selector;
    private final Map<socketchannel rpchandler> channelHandlers = new ConcurrentHashMap();

    public void start(int port) throws IOException {
        ServerSocketChannel serverChannel = ServerSocketChannel.open();
        serverChannel.configureBlocking(false);
        serverChannel.bind(new InetSocketAddress(port));
        serverChannel.register(selector, SelectionKey.OP_ACCEPT);

        while (!Thread.interrupted()) {
            selector.select(); // 阻塞直到有事件
            Set<selectionkey> keys = selector.selectedKeys();
            Iterator<selectionkey> iter = keys.iterator();
            while (iter.hasNext()) {
                SelectionKey key = iter.next();
                iter.remove();
                if (key.isAcceptable()) {
                    handleAccept(serverChannel);
                } else if (key.isReadable()) {
                    handleRead((SocketChannel) key.channel());
                } else if (key.isWritable()) {
                    handleWrite((SocketChannel) key.channel());
                }
            }
        }
    }

    private void handleAccept(ServerSocketChannel server) throws IOException {
        SocketChannel client = server.accept();
        client.configureBlocking(false);
        client.register(selector, SelectionKey.OP_READ);
        channelHandlers.put(client, new RpcHandler(client));
    }

    private void handleRead(SocketChannel channel) {
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        int read = channel.read(buffer);
        if (read > 0) {
            buffer.flip();
            RpcRequest req = decoder.decode(buffer); // 自定义解码器
            RpcResponse resp = service.invoke(req);  // 真正的业务调用（建议扔进业务线程池）
            channelHandlers.get(channel).enqueueResponse(resp);
            channel.register(selector, SelectionKey.OP_WRITE); // 触发写就绪
        }
    }
}</selectionkey></selectionkey></socketchannel>

三、客户端异步调用与响应匹配

客户端需解决两个问题：如何非阻塞发请求、如何把响应准确还给发起方。

Java Development Manual

Java开发手册规约集合，基于阿里巴巴Java开发手册（嵩山版）。涵盖7大维度：编程规约、异常日志、单元测试、安全规约、MySQL数据库、工程结构、设计规约。当用户需要：(1) 编写或审查Java代码 (2) 检查命名/代码规范 (3) 处理异常和日志 (4) 编写单元测试 (5) 安全编码 (6) 数据库设...

下载

使用 ConcurrentHashMap<long completablefuture>></long> 缓存待响应的请求，requestId 作为 key
发送请求前生成唯一 requestId，写入协议头，并将 CompletableFuture 存入 map
收到响应后，根据响应中的 requestId 查找并 complete() 对应 future，调用方通过 future.get() 或 thenApply 获取结果
设置超时机制：启动定时任务或用 future.orTimeout()（JDK9+）自动 fail