Netty笔记手写一个RPC程序

  • Z时代
  • 2024-01-10
  • 分类：综合

编程

源代码仓库 https://github.com/zhshuixian/netty-notes

RPC(Remote Proceduce Call 远程过程调用) 一般用来实现部署在不同机器上的系统之间的方法调用，使程序能够像访问本地系统资源一样，通过网络传输过去访问远端系统资源。

这里将使用 Netty 编写一个非常简单的 RPC 程序，项目大概的示意图如下：

在上一个项目的基础上，新建子项目 03-netty-rpc ,项目的依赖和 Maven 配置见 GitHub 的项目仓库。

1、Protocol RPC 请求协议

新建类 RpcProtocol.java ,定义 RPC 请求的数据的格式。远程过程调用中，有那些数据是必须通过 Netty 网络传输的，通过这个类进行封装，而后将其序列化成二进制数据流传输。这里采用了 Java 原生的的序列化和 Netty 自带的对象编码和解码。

TODO 采用 protobuf / kyro 进行系列化，对返回的数据进行统一的封装

/**
 * 发起 RPC 请求的格式，需要如下字段
 */
@Data
public class RpcProtocol implements Serializable {
    private static final long serialVersionUID = 5933724318897197513L;
    /**
     * 接口名,服务端根据接口名调用远程服务中的实际的实现类
     */
    private String interfaceName;
    /**
     * 方法名，接口方法名
     */
    private String methodName;
    /**
     * 参数的值
     */
    private Object[] paramValues;
    /**
     * 参数的类型
     */
    private Class<?>[] paramTypes;
}

2、Provider 服务端

Provider 服务端主要由三个类构成

  • Provider: 服务端的入口类，启动 Netty 服务

  • ProviderHandler: 根据请求调用在 Provider 注册的具体服务，并将结果写入 ChannelHandlerContext 返回

  • ProviderRegister: 使用 ConcurrentHashMap 保存在 Provider 注册的服务和其对象实例

ProviderRegister，主要提供两个方法，addService 和 getService，其作用是在 Provider 注册一个服务和从 HashMap 获取某个服务。

public class ProviderRegister {
    /**
     * 服务名称和其对象
     */
    private static final Map<String, Object> SERVICE_MAP = new ConcurrentHashMap<>();
    /**
     * 添加 RPC Provider 端的服务
     */
    public <T> void addService(T service, Class<T> clazz) {
        // getCanonicalName() 是获取所传类从java语言规范定义的格式输出
        String serviceName = clazz.getCanonicalName();
        log.info("添加服务，名称是 {}", serviceName);
        if (!SERVICE_MAP.containsKey(serviceName)) {
            // 将服务名和服务对应的对象添加到 SERVICE_MAP
            SERVICE_MAP.put(serviceName, service);
        }
    }
    /**
     * 获取 RPC Provider 端的服务
     */
    public Object getService(String serviceName) {
        Object service = SERVICE_MAP.get(serviceName);
        if (service == null) {
            log.debug("没有找到该 PRC 服务");
            return null;
        }
        log.info("找到服务 {}", serviceName);
        return service;
    }
}

ProviderHandler，根据 RpcProtocol 请求数据中的接口名称，获取其对应的服务，并将结果写入返回。

@Slf4j
public class ProviderHandler extends ChannelInboundHandlerAdapter {
    private final ProviderRegister register = new ProviderRegister();
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        Object result;
        RpcProtocol rpcProtocol = (RpcProtocol) msg;
        try {
            // 从 provider 查找有没有这个服务
            Object service = register.getService(rpcProtocol.getInterfaceName());
            // 从 service 根据方法名称和传入参数类型获取具体的方法
            Method method = service.getClass().getMethod(rpcProtocol.getMethodName(),
                    rpcProtocol.getParamTypes());
            // 执行这个方法
            result = method.invoke(service, rpcProtocol.getParamValues());
            // 将结果返回
            ctx.writeAndFlush(result);
            log.info("服务名称：{}，调用的方法是 {}", rpcProtocol.getInterfaceName(), rpcProtocol.getMethodName());
        } catch (NoSuchMethodException | IllegalArgumentException |
                InvocationTargetException | IllegalAccessException e) {
            log.error("服务未找到或者服务发生错误");
        } finally {
            ctx.flush();
            ctx.close();
        }
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

Provider，服务端的入口类，启动 Netty 服务，其跟前面两个 demo 中几乎没有区别，主要是更换了编码和解码器，注册 ProviderHandler 处理具体的事件。

@Slf4j
public class Provider {
    private final int port;
    private final String host;
    private final ProviderRegister register = new ProviderRegister();
    public Provider(String host, int port) {
        this.port = port;
        this.host = host;
    }
    /**
     * 启动 Netty 服务，跟前面的 demo 差不多，不同点在于编码器和解码器
     */
    public void start() {
        log.info("开始启动 RPC 服务，地址是 {} 端口号是：{}", host, port);
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    // 连接的超时时间，超过这个时间还是建立不上的话则代表连接失败
                    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
                    // TCP默认开启了 Nagle 算法，该算法的作用是尽可能的发送大数据快，减少网络传输。
                    // TCP_NODELAY 参数的作用就是控制是否启用 Nagle 算法。
                    .childOption(ChannelOption.TCP_NODELAY, true)
                    // 是否开启 TCP 底层心跳机制
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    //表示系统用于临时存放已完成三次握手的请求的队列的最大长度,如果连接建立频繁，服务器处理创建新连接较慢，
                    // 可以适当调大这个参数
                    .option(ChannelOption.SO_BACKLOG, 128)
                    // Channel 通道的绑定 ChannelPipeline
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            // 使用 JDK 自带的序列化机制 TODO 使用 protobuf 或者 kryo 进行序列化
                            // 对象的解码器
                            pipeline.addLast(new ObjectDecoder(Integer.MAX_VALUE,
                                    ClassResolvers.weakCachingConcurrentResolver(this.getClass().getClassLoader())));
                            // 对象的编码器
                            pipeline.addLast(new ObjectEncoder());
                            pipeline.addLast(new ProviderHandler());
                        }
                    });
            // 使用 bind 监听 host 和 port
            ChannelFuture future = bootstrap.bind(host, port).sync();
            future.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            log.error("开启服务错误:", e);
        } finally {
            log.info("关闭 bossGroup 和 workerGroup");
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
    /**
     * PRC  Provider 服务端注册服务
     *
     * @param service 服务
     * @param clazz   服务接口定义的类
     * @param <T>     服务具体的实现类
     */
    public <T> void addService(T service, Class<T> clazz) {
        register.addService(service, clazz);
    }
}

3、Consumer 客户端

  • Consumer：连接 Provider 服务端，发送请求

  • ConsumerHandler ：将服务端返回的数据进行处理

  • ConsumerProxy ：使用 InvocationHandler 处理动态代理对象的方法调用

// 将返回的结果提取出来即可
public class ConsumerHandler extends ChannelInboundHandlerAdapter {
    private Object result;
    public Object getResult() {
        return result;
    }
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        result = msg;
    }
}

Consumer ,使用 Netty 和 Provider 端通信，基本使用方法和 Netty 客户端方法一样，

public class Consumer {
    private final int port;
    private final String host;
    private final RpcProtocol protocol;
    public Consumer(String host, int port, RpcProtocol protocol) {
        this.port = port;
        this.host = host;
        this.protocol = protocol;
    }
    public Object start() throws InterruptedException {
        // TODO  Netty 连接复用，将这些业务抽取出来
        EventLoopGroup group = new NioEventLoopGroup();
        ConsumerHandler consumerHandler = new ConsumerHandler();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    // 连接的超时时间，超过这个时间还是建立不上的话则代表连接失败
                    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
                    // 是否开启 TCP 底层心跳机制
                    .option(ChannelOption.SO_KEEPALIVE, true)
                    // TCP默认开启了 Nagle 算法，该算法的作用是尽可能的发送大数据快，减少网络传输。TCP_NODELAY 参数的作用就是控制是否启用 Nagle 算法。
                    .option(ChannelOption.TCP_NODELAY, true)
                    // Channel 通道的绑定 ChannelPipeline
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            ChannelPipeline pipeline = ch.pipeline();
                            // 对象参数类型解码器
                            pipeline.addLast(new ObjectDecoder(Integer.MAX_VALUE,
                                    ClassResolvers.cacheDisabled(this.getClass().getClassLoader())));
                            // 对象参数类型编码器
                            pipeline.addLast(new ObjectEncoder());
                            pipeline.addLast(consumerHandler);
                        }
                    });
            // 链接到服务端和使用 ChannelFuture 接收返回的数据
            ChannelFuture future = bootstrap.connect(host, port).sync();
            // 发送请求
            future.channel().writeAndFlush(protocol).sync();
            future.channel().closeFuture().sync();
        } finally {
            group.shutdownGracefully().sync();
        }
        return consumerHandler.getResult();
    }
}

ConsumerProxy ,实现 InvocationHandler 接口，在调用动态代理的方法时候，实际是调用其中的 invoke() 方法。

public class ConsumerProxy implements InvocationHandler {
    private final String host;
    private final int port;
    public ConsumerProxy(String host, int port) {
        this.host = host;
        this.port = port;
    }
    @SuppressWarnings("unchecked")
    public <T> T getProxy(Class<T> clazz) {
        // this 表示将 ConsumerProxy 的实例传入，调用动态代理对象的时候实际调用的是 ConsumerProxy.invoke() 的方法。
        return (T) Proxy.newProxyInstance(clazz.getClassLoader(), new Class<?>[]{clazz}, this);
    }
    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        // 封装成 RPC 的请求
        RpcProtocol protocol = new RpcProtocol();
        // 获取方法对应的类名
        protocol.setInterfaceName(method.getDeclaringClass().getName());
        // 方法名
        protocol.setMethodName(method.getName());
        // 方法传入的参数类型
        protocol.setParamTypes(method.getParameterTypes());
        // 方法传入参数的实际值
        protocol.setParamValues(args);
        // 启动 PRC Consumer
        Consumer consumer = new Consumer(host, port, protocol);
        return consumer.start();
    }
}

4、使用这个 PRC 程序

  • interfaces ：客户端和服务端服务定义的接口类

  • provider : 服务端服务和实现 interfaces 接口类

  • consumer：客户端，使用本项目的 PRC 程序调用服务端的服务

interfaces：定义接口服务

public interface HelloService {
    /** 返回一个字符串 */
    String hello(String username);
}

@Data
@AllArgsConstructor
public class Message implements Serializable {
    private static final long serialVersionUID = -4268077260739000146L;
    private int code;
    private String message;
}
public interface MessageService {
    /** 返回一个自定义 Java 对象 */
    Message sayMessage(String name);
}

provider：注册 RPC 服务和监听 RPC 请求

package org.xian.rpc.test.provider
// 实现对应的接口 HelloServiceImpl.java
public class HelloServiceImpl implements HelloService {
    @Override
    public String hello(String username) {
        return "Hi " + username;
    }
}
// MessageServiceImpl.java
public class MessageServiceImpl implements MessageService {
    @Override
    public Message sayMessage(String name) {
        return new Message(200, "Your name is " + name);
    }
}
// 启动 PRC 服务端， ProviderApplication.java
public class ProviderApplication {
    public static void main(String[] args) {
        Provider server = new Provider("127.0.0.1", 8080);
        // 注册 HelloService 和 MessageService 服务
        server.addService(new HelloServiceImpl(), HelloService.class);
        server.addService(new MessageServiceImpl(), MessageService.class);
        // 启动 RPC 服务端
        server.start();
    }
}

consumer：调用 RPC 远程服务

package org.xian.rpc.test.consumer;
@Slf4j
public class ConsumerApplication {
    public static void main(String[] args) {
        ConsumerProxy proxy = new ConsumerProxy("127.0.0.1", 8080);
        // 生成动态代理对象
        HelloService helloService = proxy.getProxy(HelloService.class);
        // 实际调用的是 invoke(Object proxy, Method method, Object[] args)
        String result = helloService.hello("xian");
        log.info("HelloService 调用结果是 {}", result);
        MessageService messageService = proxy.getProxy(MessageService.class);
        Message message = messageService.sayMessage("xiaoxian");
        log.info("MessageService 调用结果是 {}", message.toString());
    }
}

在本示例中，使用 Netty 实现了一个非常简单的 RPC 服务的程序，当然还有许多需要的优化的点，比如客户端连接的复用，没有使用 Zookeeper 或者其他进行服务发现等。

参考资料：

《Netty 4核心原理与写RPC框架实战》

github.com/Snailclimb/guide-rpc-framework

以上是 Netty笔记手写一个RPC程序 的全部内容， 来源链接： utcz.com/z/519018.html

回到顶部